diff --git a/.github/workflows/mcp-browser.yml b/.github/workflows/mcp-browser.yml
new file mode 100644
index 0000000..b5e26ab
--- /dev/null
+++ b/.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
diff --git a/MCP.md b/MCP.md
index a45d5f1..742e647 100644
--- a/MCP.md
+++ b/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
diff --git a/app/mcp_auth.py b/app/mcp_auth.py
index d9ab258..554e0d4 100644
--- a/app/mcp_auth.py
+++ b/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(
""
""
@@ -552,9 +561,12 @@ def page(title: str, body: str) -> HTMLResponse:
f"YABus · 帳號連接
{html.escape(title)}
{body}",
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"{name.title()} "
body += ""
- response = page("授權讀取收藏", body)
+ response = page(
+ "授權讀取收藏", body, form_redirect_uri=str(params.redirect_uri)
+ )
cookie(response, request, CSRF_COOKIE, csrf)
return response
return page("連接你的收藏", body)
diff --git a/test_mcp.py b/test_mcp.py
index c008f9d..bd7d665 100644
--- a/test_mcp.py
+++ b/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:
diff --git a/test_mcp_browser.py b/test_mcp_browser.py
new file mode 100644
index 0000000..982984c
--- /dev/null
+++ b/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()