fix(cors): scope moz-extension wildcard to aw-watcher-web endpoints

.github/workflows/lint.yml

@@ -12,7 +12,8 @@ jobs:
     steps:
       - uses: actions/checkout@v3
       - uses: actions/setup-python@v4
-      - uses: jpetrucciani/ruff-check@main
+      # Keep lint reproducible; the floating action tag can adopt new default rules.
+      - uses: jpetrucciani/ruff-check@5839e3c65007bdb626c1f3362153e45be347654f  # ruff 0.15.22
 
   format:
     runs-on: ubuntu-latest

aw_server/extension_cors.py

@@ -0,0 +1,134 @@
+"""
+Endpoint scoping for the moz-extension:// CORS wildcard.
+
+Firefox assigns every extension its own random origin, so aw-server uses
+``moz-extension://*`` to allow aw-watcher-web without knowing the ID in advance.
+That wildcard also permits every other installed extension to reach the full API —
+including ``/api/0/export``, ``/api/0/import``, queries, and settings — with no
+host permission and therefore no install-time browser prompt naming ActivityWatch.
+
+This module registers a ``before_request`` hook that restricts wildcard-matched
+extension origins to the three endpoints aw-watcher-web actually needs:
+
+  GET  /api/0/info                                  — hostname/version detection
+  POST /api/0/buckets/aw-watcher-web-<id>           — ensure its bucket exists
+  POST /api/0/buckets/aw-watcher-web-<id>/heartbeat — heartbeat recording
+
+All other paths return 403 before the handler executes. flask-cors CORS headers are
+still added by the after_request hook, but the 403 status prevents JavaScript from
+treating the response as a successful cross-origin fetch (and more importantly, the
+server side never processes the request).
+
+User-configured origins (``cors_origins`` / ``cors_regex`` in config) are exempted:
+those are explicit opt-ins by the server owner, unlike the built-in wildcard.
+
+Path matching uses split segments rather than the raw path string to avoid
+percent-encoding bypasses — the same bug class as aw-server-rust#588 and #636.
+
+See also: ActivityWatch/aw-server-rust#637 (the Rust sibling of this fix).
+"""
+
+import logging
+import re
+from typing import List, Optional
+
+from flask import Flask, abort, request
+
+logger = logging.getLogger(__name__)
+
+_EXTENSION_SCHEME = "moz-extension://"
+
+
+def register(app: Flask, user_origins: List[str]) -> None:
+    """Register the extension CORS scope hook on *app*.
+
+    *user_origins* — origins the user configured explicitly (captured before
+    the built-in ``moz-extension://*`` wildcard is appended).  These are
+    explicit opt-ins and bypass the scope narrowing.
+    """
+
+    @app.before_request
+    def _restrict_extension_cors() -> Optional[object]:
+        origin = request.headers.get("Origin", "")
+        if not origin.lower().startswith(_EXTENSION_SCHEME):
+            return None  # not a moz-extension origin — let flask-cors handle it
+
+        # flask-cors 4 treats strings containing regex metacharacters as regular
+        # expressions and otherwise compares them case-insensitively. Keep this
+        # exemption consistent with that contract.
+        for pattern in user_origins:
+            if _matches_configured_origin(origin, pattern):
+                return None
+
+        segments = [s for s in request.path.split("/") if s]
+        if _is_allowed(request.method, segments):
+            return None
+
+        abort(403)
+
+
+def _matches_configured_origin(origin: str, pattern: str) -> bool:
+    """Match an origin using flask-cors 4's configured-origin semantics."""
+    regex_chars = "*\\]?$^[()"
+    if any(char in pattern for char in regex_chars):
+        try:
+            return re.match(pattern, origin, flags=re.IGNORECASE) is not None
+        except re.error:
+            return False
+    return origin.lower() == pattern.lower()
+
+
+def _is_allowed(method: str, segments: List[str]) -> bool:
+    """Return True if *method* + *segments* is a path aw-watcher-web actually uses.
+
+    For OPTIONS preflights the path is checked against the set of allowed paths
+    (not the Access-Control-Request-Method header) to keep the logic simple while
+    still blocking preflights for disallowed paths such as ``/api/0/export``.
+    """
+    if method == "OPTIONS":
+        return _is_allowed_path(segments)
+
+    # GET /api/0/info
+    if method == "GET" and segments == ["api", "0", "info"]:
+        return True
+
+    # POST /api/0/buckets/aw-watcher-web-<id>
+    if (
+        method == "POST"
+        and len(segments) == 4
+        and segments[:3] == ["api", "0", "buckets"]
+        and segments[3].startswith("aw-watcher-web-")
+    ):
+        return True
+
+    # POST /api/0/buckets/aw-watcher-web-<id>/heartbeat
+    if (
+        method == "POST"
+        and len(segments) == 5
+        and segments[:3] == ["api", "0", "buckets"]
+        and segments[3].startswith("aw-watcher-web-")
+        and segments[4] == "heartbeat"
+    ):
+        return True
+
+    return False
+
+
+def _is_allowed_path(segments: List[str]) -> bool:
+    """Return True if *segments* is on any allowed endpoint (for OPTIONS checks)."""
+    if segments == ["api", "0", "info"]:
+        return True
+    if (
+        len(segments) == 4
+        and segments[:3] == ["api", "0", "buckets"]
+        and segments[3].startswith("aw-watcher-web-")
+    ):
+        return True
+    if (
+        len(segments) == 5
+        and segments[:3] == ["api", "0", "buckets"]
+        and segments[3].startswith("aw-watcher-web-")
+        and segments[4] == "heartbeat"
+    ):
+        return True
+    return False

aw_server/server.py

@@ -14,7 +14,7 @@
 )
 from flask_cors import CORS
 
-from . import rest
+from . import extension_cors, rest
 from .api import ServerAPI
 from .custom_static import get_custom_static_blueprint
 from .log import FlaskLogHandler
@@ -105,13 +105,21 @@ def _config_cors(cors_origins: List[str], testing: bool):
         # Used for development of aw-webui
         cors_origins.append("http://127.0.0.1:27180/*")
 
+    # Capture user-configured origins before appending the built-in wildcard.
+    # extension_cors uses this list to exempt explicit opt-ins from scope narrowing.
+    user_origins = list(cors_origins)
+
     # TODO: This could probably be more specific
     #       See https://github.com/ActivityWatch/aw-server/pull/43#issuecomment-386888769
     cors_origins.append("moz-extension://*")
 
     # See: https://flask-cors.readthedocs.org/en/latest/
     CORS(current_app, resources={r"/api/*": {"origins": cors_origins}})
 
+    # Narrow the moz-extension wildcard to only the endpoints aw-watcher-web needs.
+    # See aw_server/extension_cors.py and ActivityWatch/aw-server-rust#637.
+    extension_cors.register(current_app._get_current_object(), user_origins)
+
 
 # Only to be called from aw_server.main function!
 def _start(

tests/test_extension_cors.py

@@ -0,0 +1,171 @@
+"""Tests for moz-extension CORS endpoint scoping (extension_cors module)."""
+
+import pytest
+
+from aw_server.extension_cors import (
+    _is_allowed,
+    _is_allowed_path,
+    _matches_configured_origin,
+)
+from aw_server.server import AWFlask
+
+_EXT_ORIGIN = "moz-extension://aabbccddeeff00112233445566778899"
+_HOST = "127.0.0.1"
+
+
+@pytest.fixture(scope="module")
+def client():
+    app = AWFlask(_HOST, testing=True)
+    return app.test_client()
+
+
+# ---------------------------------------------------------------------------
+# Unit tests for the path-matching helpers
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "method,path,expected",
+    [
+        ("GET", "/api/0/info", True),
+        ("POST", "/api/0/buckets/aw-watcher-web-hostname", True),
+        ("POST", "/api/0/buckets/aw-watcher-web-hostname/heartbeat", True),
+        # Blocked paths
+        ("GET", "/api/0/export", False),
+        ("POST", "/api/0/export", False),
+        ("GET", "/api/0/buckets/", False),
+        ("GET", "/api/0/buckets/aw-watcher-web-hostname/events", False),
+        ("POST", "/api/0/query/", False),
+        ("GET", "/api/0/settings", False),
+        # Non-watcher bucket
+        ("POST", "/api/0/buckets/aw-watcher-window-hostname", False),
+        # Heartbeat on non-watcher bucket
+        ("POST", "/api/0/buckets/aw-watcher-afk-hostname/heartbeat", False),
+        # Percent-encoding should not bypass via raw path (segments are used)
+        ("GET", "/api/0/%65xport", False),  # %65 = 'e', decodes to 'export'
+    ],
+)
+def test_is_allowed_unit(method, path, expected):
+    segments = [s for s in path.split("/") if s]
+    assert _is_allowed(method, segments) is expected
+
+
+@pytest.mark.parametrize(
+    "path,expected",
+    [
+        ("/api/0/info", True),
+        ("/api/0/buckets/aw-watcher-web-hostname", True),
+        ("/api/0/buckets/aw-watcher-web-hostname/heartbeat", True),
+        ("/api/0/export", False),
+        ("/api/0/buckets/", False),
+    ],
+)
+def test_is_allowed_path_unit(path, expected):
+    segments = [s for s in path.split("/") if s]
+    assert _is_allowed_path(segments) is expected
+
+
+# ---------------------------------------------------------------------------
+# Integration tests via Flask test client
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "method,path",
+    [
+        ("GET", "/api/0/info"),
+        # Bucket creation: 400 without body is expected but NOT 403
+        ("POST", "/api/0/buckets/aw-watcher-web-testhost"),
+        # Heartbeat: 404 (bucket doesn't exist) is expected but NOT 403
+        ("POST", "/api/0/buckets/aw-watcher-web-testhost/heartbeat"),
+    ],
+)
+def test_extension_allowed(client, method, path):
+    """moz-extension origins are permitted at aw-watcher-web's endpoints."""
+    headers = {"Origin": _EXT_ORIGIN}
+    r = client.open(path, method=method, headers=headers)
+    assert (
+        r.status_code != 403
+    ), f"{method} {path} should not be blocked; got {r.status_code}"
+
+
+@pytest.mark.parametrize(
+    "method,path",
+    [
+        ("GET", "/api/0/export"),
+        ("GET", "/api/0/buckets/"),
+        ("POST", "/api/0/import"),
+        ("POST", "/api/0/query/"),
+        ("GET", "/api/0/settings"),
+        # Events read from a watcher bucket
+        ("GET", "/api/0/buckets/aw-watcher-web-testhost/events"),
+        # Non-watcher bucket
+        ("POST", "/api/0/buckets/aw-watcher-window-testhost"),
+        ("POST", "/api/0/buckets/aw-watcher-afk-testhost/heartbeat"),
+    ],
+)
+def test_extension_blocked(client, method, path):
+    """moz-extension origins are blocked at endpoints beyond aw-watcher-web's needs."""
+    headers = {"Origin": _EXT_ORIGIN}
+    r = client.open(path, method=method, headers=headers)
+    assert (
+        r.status_code == 403
+    ), f"{method} {path} should be blocked (403); got {r.status_code}"
+
+
+@pytest.mark.parametrize(
+    "pattern,origin,expected",
+    [
+        ("moz-extension://aabbcc", "moz-extension://aabbcc", True),
+        ("MOZ-EXTENSION://AABBCC", "moz-extension://aabbcc", True),
+        (r"moz-extension://.*", "moz-extension://aabbcc", True),
+        (r"moz-extension://[a-f0-9]+", "moz-extension://aabbcc", True),
+        (r"moz-extension://[0-9]+", "moz-extension://aabbcc", False),
+        ("moz-extension://other", "moz-extension://aabbcc", False),
+    ],
+)
+def test_matches_configured_origin(pattern, origin, expected):
+    assert _matches_configured_origin(origin, pattern) is expected
+
+
+def test_regex_configured_extension_origin_bypasses_scope_guard():
+    """Owner-configured regex origins retain unrestricted endpoint access."""
+    app = AWFlask(_HOST, testing=False, cors_origins=[r"moz-extension://.*"])
+    client = app.test_client()
+
+    response = client.get("/api/0/export", headers={"Origin": _EXT_ORIGIN})
+
+    assert response.status_code != 403
+
+
+def test_non_extension_origin_passthrough(client):
+    """Non-moz-extension origins are not affected by the scope guard."""
+    headers = {"Origin": "http://127.0.0.1:27180"}
+    r = client.get("/api/0/info", headers=headers)
+    assert r.status_code != 403
+
+
+def test_no_origin_passthrough(client):
+    """Requests without an Origin header (native watchers, curl) are not blocked."""
+    r = client.get("/api/0/info")
+    assert r.status_code != 403
+
+
+def test_options_allowed_path(client):
+    """OPTIONS preflight on an allowed path is permitted."""
+    headers = {
+        "Origin": _EXT_ORIGIN,
+        "Access-Control-Request-Method": "GET",
+    }
+    r = client.options("/api/0/info", headers=headers)
+    assert r.status_code != 403
+
+
+def test_options_blocked_path(client):
+    """OPTIONS preflight on a blocked path is rejected."""
+    headers = {
+        "Origin": _EXT_ORIGIN,
+        "Access-Control-Request-Method": "GET",
+    }
+    r = client.options("/api/0/export", headers=headers)
+    assert r.status_code == 403