fix: repair browser OAuth consent and callback redirects

.github/workflows/mcp-browser.yml

@@ -0,0 +1,28 @@
+name: MCP browser consent
+
+on:
+  pull_request:
+  push:
+    branches: [main]
+
+permissions:
+  contents: read
+
+jobs:
+  consent:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@v5
+      - uses: actions/setup-python@v6
+        with:
+          python-version: '3.13'
+          cache: pip
+      - name: Install test dependencies
+        run: python -m pip install -r requirements.txt pytest playwright
+      - name: Install Chromium
+        run: python -m playwright install --with-deps chromium
+      - name: Exercise OAuth forms in Chromium and the API
+        env:
+          RUN_MCP_BROWSER_TESTS: '1'
+        run: python -m pytest test_mcp_browser.py test_mcp.py test_auth_api.py test_account_page.py -q

MCP.md

@@ -62,6 +62,11 @@ Google/Discord sign-in completes on the existing server callback, with a browser
 state. The temporary application token is immediately revoked and replaced with a
 10-minute HttpOnly consent session. Application tokens are never sent to an AI client.
 Consent POSTs require a CSRF token and same-origin Origin header.
+Consent and authorization-management HTML uses `Referrer-Policy: strict-origin`
+so native browser form POSTs keep their Origin while omitting paths/query parameters
+from Referer. OAuth redirects retain `no-referrer`. Consent's `form-action` CSP
+permits only this server and the registered callback origin for that transaction;
+this lets browsers follow the POST redirect back to the AI client.
 
 Authorization codes expire after two minutes and are consumed atomically. Access
 tokens last at most one hour; refresh tokens have a 30-day inactivity limit and an
@@ -103,6 +108,20 @@ refresh, revocation and public/private separation without contacting an identity
 provider. Real provider login, platform directory submission and Android Gemini
 preview access still require the deployed service and platform account.
 
+Native Chromium form regressions additionally exercise approval, denial and
+revocation without injecting an Origin header. They use isolated browser storage,
+synthetic accounts and an in-process HTTP bridge; no identity provider is contacted.
+
+```powershell
+python -m pip install playwright
+python -m playwright install chromium
+$env:RUN_MCP_BROWSER_TESTS = "1"
+python -m pytest test_mcp_browser.py -q
+```
+
+`MCP_TEST_BROWSER_EXECUTABLE` can optionally select an existing Chromium executable.
+Without `RUN_MCP_BROWSER_TESTS=1`, these tests are skipped by the offline suite.
+
 For the existing offline suite, exclude the two live TDX probe scripts:
 
 ```powershell

app/mcp_auth.py

@@ -14,7 +14,7 @@
 import re
 import secrets
 import time
-from urllib.parse import parse_qs, urlencode, urlsplit
+from urllib.parse import parse_qs, quote, urlencode, urlsplit
 from uuid import uuid4
 
 from fastapi import HTTPException
@@ -539,7 +539,16 @@ def browser_principal(provider, request):
     return authenticate_token(provider.settings, token) if token else None
 
 
-def page(title: str, body: str) -> HTMLResponse:
+def page(
+    title: str, body: str, *, form_redirect_uri: str | None = None
+) -> HTMLResponse:
+    form_sources = "'self'"
+    if form_redirect_uri is not None:
+        # Chromium also applies form-action to the redirect after a form POST.
+        # Allow only this transaction's registered callback origin. Quote CSP
+        # delimiters so even unusual registered URLs cannot add directives.
+        target = urlsplit(form_redirect_uri)
+        form_sources += " " + quote(f"{target.scheme}://{target.netloc}", safe=":/[]")
     return HTMLResponse(
         "<!doctype html><html lang='zh-Hant'><meta charset='utf-8'>"
         "<meta name='viewport' content='width=device-width,initial-scale=1'>"
@@ -552,9 +561,12 @@ def page(title: str, body: str) -> HTMLResponse:
         f"</style><main><p>YABus · 帳號連接</p><h1>{html.escape(title)}</h1>{body}</main></html>",
         headers={
             "Cache-Control": "no-store",
-            "Referrer-Policy": "no-referrer",
+            # no-referrer makes native HTML form POSTs send Origin: null.
+            # Keep the same-origin CSRF check usable while omitting URL paths
+            # and query parameters from Referer (including OAuth request IDs).
+            "Referrer-Policy": "strict-origin",
             "X-Frame-Options": "DENY",
-            "Content-Security-Policy": "default-src 'none'; style-src 'unsafe-inline'; form-action 'self'; frame-ancestors 'none'; base-uri 'none'",
+            "Content-Security-Policy": f"default-src 'none'; style-src 'unsafe-inline'; form-action {form_sources}; frame-ancestors 'none'; base-uri 'none'",
         },
     )
 
@@ -616,7 +628,9 @@ async def consent_page(provider, request):
         for name in ("google", "discord"):
             body += f"<a href='/oauth/login/{name}?{urlencode({'request_id': pending_id})}'>{name.title()}</a> "
         body += "</p>"
-        response = page("授權讀取收藏", body)
+        response = page(
+            "授權讀取收藏", body, form_redirect_uri=str(params.redirect_uri)
+        )
         cookie(response, request, CSRF_COOKIE, csrf)
         return response
     return page("連接你的收藏", body)

test_mcp.py

@@ -290,6 +290,64 @@ def test_pkce_redirect_resource_scope_csrf_and_denial(self):
         self.assertEqual(denial["error"], ["access_denied"])
         self.assertNotIn("code", denial)
 
+    def test_consent_form_preserves_origin_without_weakening_csrf(self):
+        client = self.register()
+        response = self.begin(client)
+        self.client.cookies.set("yabus_auth_token", self.login.token)
+        consent = self.client.get(response.headers["location"])
+        self.assertEqual(consent.headers["referrer-policy"], "strict-origin")
+        self.assertIn(
+            "form-action 'self' https://client.example;",
+            consent.headers["content-security-policy"],
+        )
+        pending_id = parse_qs(urlsplit(response.headers["location"]).query)[
+            "request_id"
+        ][0]
+        csrf = re.search("name='csrf' value='([^']+)'", consent.text)[1]
+        form = {"request_id": pending_id, "csrf": csrf, "decision": "approve"}
+        for headers in (
+            {},
+            {"Origin": "null"},
+            {"Origin": "https://evil.example"},
+            {"Origin": "https://client.example"},
+        ):
+            with self.subTest(headers=headers):
+                self.assertEqual(
+                    self.client.post(
+                        "/oauth/consent", data=form, headers=headers
+                    ).status_code,
+                    403,
+                )
+        for invalid in ("", "invalid-csrf"):
+            self.assertEqual(
+                self.client.post(
+                    "/oauth/consent",
+                    data={**form, "csrf": invalid},
+                    headers={"Origin": self.provider.issuer},
+                ).status_code,
+                403,
+            )
+        result = self.client.post(
+            "/oauth/consent", data=form, headers={"Origin": self.provider.issuer}
+        )
+        self.assertEqual(result.status_code, 303)
+        self.assertEqual(result.headers["referrer-policy"], "no-referrer")
+        manage = self.client.get("/oauth/connections")
+        self.assertEqual(manage.headers["referrer-policy"], "strict-origin")
+        self.assertIn("form-action 'self';", manage.headers["content-security-policy"])
+
+    def test_callback_origin_cannot_inject_csp_directives(self):
+        from app.mcp_auth import page
+
+        rendered = page(
+            "Consent",
+            "",
+            form_redirect_uri="https://client.example;script-src%20*/callback",
+        )
+        policy = rendered.headers["content-security-policy"]
+        self.assertNotIn(";script-src", policy)
+        self.assertNotIn(" *", policy)
+
     def test_secret_hashed_at_rest_and_basic_auth(self):
         client = self.register("client_secret_basic")
         with get_connection(self.settings.app_db_path) as db:

test_mcp_browser.py

@@ -0,0 +1,168 @@
+"""Opt-in native browser regressions, using temporary accounts and no network.
+
+RUN_MCP_BROWSER_TESTS=1 python -m pytest test_mcp_browser.py -q
+Install Playwright plus Chromium first. MCP_TEST_BROWSER_EXECUTABLE optionally
+selects an existing Chromium executable; no personal browser profile is used.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import logging
+import os
+import unittest
+from urllib.parse import parse_qs, urlsplit
+
+import test_mcp
+
+
+@unittest.skipUnless(
+    os.getenv("RUN_MCP_BROWSER_TESTS") == "1"
+    and importlib.util.find_spec("playwright") is not None,
+    "Set RUN_MCP_BROWSER_TESTS=1 and install Playwright/Chromium",
+)
+class MCPBrowserConsentTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        from playwright.sync_api import sync_playwright
+
+        cls.playwright = sync_playwright().start()
+        cls.addClassCleanup(cls.playwright.stop)
+        # Use full Chromium's new headless mode, matching desktop Chrome. The
+        # separate headless-shell bypasses route interception on this redirect
+        # and attempts DNS for the deliberately non-existent callback domain.
+        options = {"headless": True, "channel": "chromium"}
+        executable = os.getenv("MCP_TEST_BROWSER_EXECUTABLE")
+        if executable:
+            options["executable_path"] = executable
+        cls.browser = cls.playwright.chromium.launch(**options)
+        cls.addClassCleanup(cls.browser.close)
+
+    def setUp(self):
+        # The fixture has synthetic accounts, temporary databases and mocked
+        # transit services. Suppress HTTP client URLs containing test OAuth data.
+        previous_level = logging.root.manager.disable
+        logging.disable(logging.CRITICAL)
+        self.addCleanup(logging.disable, previous_level)
+        self.server = test_mcp.MCPIntegrationTests()
+        self.server.setUp()
+        self.addCleanup(self.server.tearDown)
+        self.origin = self.server.provider.issuer
+        self.posts = []
+        self.callbacks = []
+        self.context = self.browser.new_context()
+        self.addCleanup(self.context.close)
+        self.context.add_cookies(
+            [
+                {
+                    "name": "yabus_auth_token",
+                    "value": self.server.login.token,
+                    "url": self.origin,
+                    "httpOnly": True,
+                    "secure": True,
+                    "sameSite": "Lax",
+                },
+            ]
+        )
+        self.context.route("**/*", self.bridge)
+        self.page = self.context.new_page()
+
+    def bridge(self, route):
+        request = route.request
+        target = urlsplit(request.url)
+        if target.hostname == "client.example" and target.path == "/callback":
+            self.callbacks.append(
+                {
+                    "referer": request.headers.get("referer"),
+                    "code": bool(parse_qs(target.query).get("code")),
+                    "error": parse_qs(target.query).get("error", [None])[0],
+                }
+            )
+            route.fulfill(
+                status=200, content_type="text/html", body="Synthetic OAuth callback"
+            )
+            return
+        if not request.url.startswith(self.origin + "/"):
+            route.abort()
+            return
+        # Trust only the browser's headers/cookies. TestClient's cookie jar must
+        # not conceal missing cookies or manufacture an Origin on its behalf.
+        self.server.client.cookies.clear()
+        response = self.server.client.request(
+            request.method,
+            request.url,
+            headers=request.all_headers(),
+            content=request.post_data_buffer,
+        )
+        if request.method == "POST":
+            self.posts.append(
+                {
+                    "path": target.path,
+                    "origin": request.headers.get("origin"),
+                    "referer": request.headers.get("referer"),
+                    "status": response.status_code,
+                }
+            )
+        headers = dict(response.headers)
+        headers.pop("content-length", None)
+        route.fulfill(
+            status=response.status_code, headers=headers, body=response.content
+        )
+
+    def open_consent(self):
+        client = self.server.register()
+        response = self.server.begin(client)
+        self.assertEqual(response.status_code, 302)
+        page = self.page.goto(response.headers["location"])
+        self.assertEqual(page.status, 200)
+        return client
+
+    def assert_post(self, path):
+        self.assertEqual(
+            self.posts[-1],
+            {
+                "path": path,
+                "origin": self.origin,
+                "referer": self.origin + "/",
+                "status": 303,
+            },
+        )
+
+    def test_approve_with_browser_generated_origin_reaches_callback(self):
+        self.open_consent()
+        self.page.locator("button[value=approve]").click()
+        self.page.wait_for_url("https://client.example/callback*", timeout=5000)
+        self.assert_post("/oauth/consent")
+        self.assertEqual(urlsplit(self.page.url).hostname, "client.example")
+        self.assertEqual(
+            self.callbacks, [{"referer": None, "code": True, "error": None}]
+        )
+        self.assertEqual(
+            len(self.server.provider.account_grants(self.server.login.account_id)), 1
+        )
+
+    def test_deny_with_browser_generated_origin_reaches_callback(self):
+        self.open_consent()
+        self.page.locator("button[value=deny]").click()
+        self.page.wait_for_url("https://client.example/callback*", timeout=5000)
+        self.assert_post("/oauth/consent")
+        self.assertEqual(
+            self.callbacks, [{"referer": None, "code": False, "error": "access_denied"}]
+        )
+        self.assertEqual(
+            self.server.provider.account_grants(self.server.login.account_id), []
+        )
+
+    def test_revoke_with_browser_generated_origin(self):
+        client = self.server.register()
+        self.server.consent(client)
+        self.page.goto(self.origin + "/oauth/connections")
+        self.page.locator("button").click()
+        self.assert_post("/oauth/connections")
+        self.assertEqual(
+            self.server.provider.account_grants(self.server.login.account_id), []
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()