mcp 2.2: domain search needs the MCP's sign-in; the Domains tab sends people through it once

/domains/sign-in and /domains/callback do Tea MCP's own Gitea OAuth (the
Green Tea MCP app) in the browser, keep only the name from Gitea, and leave a
signed, HttpOnly cookie on mcp.teahub.xyz for 30 days. GET /domains wants it
(401 with a sign-in link otherwise) and rate-limits per person. The Domains
tab sends that cookie, offers the sign-in on a 401, and runs the waiting
search on the way back. Tested end to end against a Gitea 28 and Cloudflare.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CYHXFvWMCs1GUmRmpySqQ6

mcp/tea_mcp/domains.py

@@ -1,33 +1,45 @@
-"""GET /domains: Cloudflare Registrar search for the Domains tab of Teahub's dashboard.
+"""GET /domains: Cloudflare Registrar search for the Domains tab of Teahub's dashboard, for people signed in.
 
-The dashboard on GITEA_URL asks with ?q=<a name, a phrase or a domain>. Exact names (the domain as given, or the
-name under each of ENDINGS) are checked live with Cloudflare's domain-check; suggestions come from its domain-search.
-Each comes with availability and Cloudflare's at-cost price. Nothing is registered or bought here. The Cloudflare
-token stays on this server and anyone may ask, so each client gets a few searches a minute, everyone together a few
-more, and answers are kept for ten minutes.
+Signing in is the MCP's own sign-in, done in the browser. /domains/sign-in sends the browser to Gitea's OAuth with this
+server's OAuth app (the one Claude's connector uses, so anyone who has connected is let straight through), and
+/domains/callback learns from Gitea who it is and remembers them for 30 days in a signed, HttpOnly cookie on this host.
+The dashboard on GITEA_URL then asks with ?q=<a name, a phrase or a domain> and that cookie (teahub.xyz and
+mcp.teahub.xyz are one site, so the browser sends it). Exact names (the domain as given, or the name under each of
+ENDINGS) are checked live with Cloudflare's domain-check; suggestions come from its domain-search, each with
+availability and Cloudflare's at-cost price. Nothing is registered or bought. Each person gets a few searches a minute,
+everyone together a few more, and answers are kept for ten minutes.
 """
 from __future__ import annotations
 
 import asyncio
+import base64
+import binascii
+import hashlib
+import hmac
+import json
 import os
 import re
+import secrets
 import time
 from collections import OrderedDict, deque
 from typing import Any
+from urllib.parse import urlencode
 
 import httpx
 from starlette.requests import Request
-from starlette.responses import JSONResponse, Response
+from starlette.responses import JSONResponse, PlainTextResponse, RedirectResponse, Response
 
 CLOUDFLARE = "https://api.cloudflare.com/client/v4"
 QUERY = re.compile(r"[A-Za-z0-9][A-Za-z0-9 ._-]{0,99}")
 ENDINGS = ("com", "net", "org", "co", "io", "xyz", "app", "dev", "ai", "me")
-PER_CLIENT, IN_ALL, WINDOW = 8, 20, 60.0  # searches per WINDOW seconds that go to Cloudflare
+PER_PERSON, IN_ALL, WINDOW = 8, 20, 60.0  # searches per WINDOW seconds that go to Cloudflare
 KEEP_SECONDS, KEEP_ANSWERS = 600.0, 256
+SIGNED_IN, SIGNING_IN = "tea_domains", "tea_domains_signin"  # cookies, both scoped to /domains on this host
+SIGNED_IN_SECONDS, SIGNING_IN_SECONDS = 30 * 86400, 600
 
-transport: httpx.AsyncBaseTransport | None = None  # the tests' fake Cloudflare
+transport: httpx.AsyncBaseTransport | None = None  # the tests' fake Cloudflare and Gitea
 _kept: OrderedDict[tuple[str, int], tuple[float, dict[str, Any]]] = OrderedDict()
-_by_client: dict[str, deque[float]] = {}
+_by_person: dict[str, deque[float]] = {}
 _by_all: deque[float] = deque()
 
 
@@ -35,6 +47,44 @@ class Unanswered(Exception):
     """Cloudflare gave no usable answer."""
 
 
+def _url(name: str) -> str:
+    return os.getenv(name, "").strip().rstrip("/")
+
+
+def _b64(data: bytes) -> str:
+    return base64.urlsafe_b64encode(data).decode().rstrip("=")
+
+
+def _mac(body: str) -> str:
+    key = hmac.new(os.environ["JWT_SIGNING_KEY"].encode(), b"teahub domains cookie", hashlib.sha256).digest()
+    return _b64(hmac.new(key, body.encode(), hashlib.sha256).digest())
+
+
+def seal(data: dict[str, Any], seconds: int) -> str:
+    """`data` and an expiry, signed: readable by anyone, made only here."""
+    body = _b64(json.dumps({**data, "exp": int(time.time()) + seconds}, separators=(",", ":")).encode())
+    return f"{body}.{_mac(body)}"
+
+
+def unseal(value: str | None) -> dict[str, Any] | None:
+    body, _, mac = (value or "").partition(".")
+    if not body or not hmac.compare_digest(mac.encode(), _mac(body).encode()):
+        return None
+    try:
+        data = json.loads(base64.urlsafe_b64decode(body + "=" * (-len(body) % 4)))
+    except (ValueError, binascii.Error):
+        return None
+    return data if isinstance(data, dict) and isinstance(data.get("exp"), int) and data["exp"] > time.time() else None
+
+
+def _json(response: httpx.Response) -> dict[str, Any]:
+    try:
+        body = response.json()
+    except ValueError:
+        return {}
+    return body if isinstance(body, dict) else {}
+
+
 def exact_names(query: str) -> list[str]:
     """A domain is checked as given; a name or phrase becomes one label under each of ENDINGS."""
     value = query.strip().lower()
@@ -47,13 +97,10 @@ def exact_names(query: str) -> list[str]:
 def _rows(answer: httpx.Response | BaseException, *, invalid_ok: bool = False) -> list[dict[str, Any]]:
     if isinstance(answer, BaseException):
         raise Unanswered(type(answer).__name__)
-    try:
-        body = answer.json()
-    except ValueError:
-        body = {}
     if answer.status_code == 400 and invalid_ok:
         return []  # none of the names is a domain Cloudflare could register (code 1008)
-    if answer.status_code >= 400 or not isinstance(body, dict) or body.get("success") is False:
+    body = _json(answer)
+    if answer.status_code >= 400 or body.get("success") is False:
         raise Unanswered(f"HTTP {answer.status_code}")
     rows = []
     for item in (body.get("result") or {}).get("domains") or []:
@@ -81,32 +128,36 @@ async def search(query: str, limit: int, token: str, account: str) -> dict[str,
             "domains": [row for row in found if str(row["name"]).lower() not in names]}
 
 
-def _allowed(client: str, now: float) -> bool:
-    mine = _by_client.setdefault(client, deque())
+def _allowed(person: str, now: float) -> bool:
+    mine = _by_person.setdefault(person, deque())
     for times in (_by_all, mine):
         while times and now - times[0] > WINDOW:
             times.popleft()
-    if len(mine) >= PER_CLIENT or len(_by_all) >= IN_ALL:
+    if len(mine) >= PER_PERSON or len(_by_all) >= IN_ALL:
         return False
     mine.append(now)
     _by_all.append(now)
-    if len(_by_client) > 2048:
-        for name in [name for name, times in _by_client.items() if not times or now - times[-1] > WINDOW]:
-            del _by_client[name]
+    if len(_by_person) > 2048:
+        for name in [name for name, times in _by_person.items() if not times or now - times[-1] > WINDOW]:
+            del _by_person[name]
     return True
 
 
-async def domains_http(request: Request) -> Response:
-    origin = os.getenv("GITEA_URL", "").rstrip("/")
-    cors = {"Vary": "Origin"}
+def _cors(request: Request) -> dict[str, str]:
+    origin = _url("GITEA_URL")
     if origin and request.headers.get("origin") == origin:
-        cors["Access-Control-Allow-Origin"] = origin
+        return {"Access-Control-Allow-Origin": origin, "Access-Control-Allow-Credentials": "true", "Vary": "Origin"}
+    return {"Vary": "Origin"}
+
+
+async def domains_http(request: Request) -> Response:
+    cors = _cors(request)
     if request.method == "OPTIONS":
         return Response(status_code=204, headers={**cors, "Access-Control-Allow-Methods": "GET",
                                                   "Access-Control-Max-Age": "86400"})
 
     def answer(body: dict[str, Any], status: int = 200) -> JSONResponse:
-        keep = "public, max-age=300" if status == 200 else "no-store"
+        keep = "private, max-age=300" if status == 200 else "no-store"
         return JSONResponse(body, status, headers={**cors, "Cache-Control": keep})
 
     query = (request.query_params.get("q") or "").strip()
@@ -122,13 +173,15 @@ async def domains_http(request: Request) -> Response:
     account = os.getenv("CLOUDFLARE_ACCOUNT_ID", "").strip()
     if not (token and account):
         return answer({"error": "Domain search is not set up here."}, 503)
+    person = (unseal(request.cookies.get(SIGNED_IN)) or {}).get("u")
+    if not person:
+        return answer({"error": "Sign in to search domains.", "sign_in": f"{_url('PUBLIC_URL')}/domains/sign-in"}, 401)
 
     key, now = (query.lower(), limit), time.monotonic()
     kept = _kept.get(key)
     if kept and kept[0] > now:
         return answer(kept[1])
-    client = request.headers.get("fly-client-ip") or (request.client.host if request.client else "")
-    if not _allowed(client, now):
+    if not _allowed(person, now):
         return answer({"error": "Too many searches. Try again in a minute."}, 429)
     try:
         result = await search(query, limit, token, account)
@@ -139,3 +192,52 @@ async def domains_http(request: Request) -> Response:
     while len(_kept) > KEEP_ANSWERS:
         _kept.popitem(last=False)
     return answer(result)
+
+
+async def sign_in_http(request: Request) -> Response:
+    """Off to Gitea's OAuth with this server's app; `next`, a page on GITEA_URL, is where the browser comes back to."""
+    gitea, public = _url("GITEA_URL"), _url("PUBLIC_URL")
+    back = request.query_params.get("next") or ""
+    if not back.startswith(f"{gitea}/"):
+        back = f"{gitea}/"
+    state, verifier = secrets.token_urlsafe(24), secrets.token_urlsafe(48)
+    to = f"{gitea}/login/oauth/authorize?" + urlencode({
+        "client_id": os.environ["GITEA_OAUTH_CLIENT_ID"], "redirect_uri": f"{public}/domains/callback",
+        "response_type": "code", "scope": "openid profile", "state": state,
+        "code_challenge": _b64(hashlib.sha256(verifier.encode()).digest()), "code_challenge_method": "S256"})
+    response = RedirectResponse(to, 302)
+    response.set_cookie(SIGNING_IN, seal({"s": state, "v": verifier, "n": back}, SIGNING_IN_SECONDS),
+                        max_age=SIGNING_IN_SECONDS, path="/domains", secure=True, httponly=True, samesite="lax")
+    return response
+
+
+async def callback_http(request: Request) -> Response:
+    """Gitea's answer: the code becomes a Gitea token, the token says who signed in, and only the name is kept."""
+    pending = unseal(request.cookies.get(SIGNING_IN))
+    code, state = request.query_params.get("code", ""), request.query_params.get("state", "")
+    if not pending or not code or not hmac.compare_digest(state.encode(), str(pending.get("s", "")).encode()):
+        return PlainTextResponse("That sign-in has expired or was not started here. Start again from the Domains tab.",
+                                 400)
+    gitea, public = _url("GITEA_URL"), _url("PUBLIC_URL")
+    person = None
+    try:
+        async with httpx.AsyncClient(timeout=12, transport=transport) as client:
+            granted = await client.post(f"{gitea}/login/oauth/access_token", headers={"Accept": "application/json"},
+                                        data={"grant_type": "authorization_code", "code": code,
+                                              "redirect_uri": f"{public}/domains/callback",
+                                              "client_id": os.environ["GITEA_OAUTH_CLIENT_ID"],
+                                              "client_secret": os.environ["GITEA_OAUTH_CLIENT_SECRET"],
+                                              "code_verifier": str(pending.get("v", ""))})
+            access = _json(granted).get("access_token") if granted.status_code == 200 else None
+            if access:
+                me = await client.get(f"{gitea}/login/oauth/userinfo", headers={"Authorization": f"Bearer {access}"})
+                person = _json(me).get("preferred_username") if me.status_code == 200 else None
+    except httpx.HTTPError:
+        person = None
+    if not person or not isinstance(person, str):
+        return PlainTextResponse("Teahub could not sign you in just now. Try again from the Domains tab.", 502)
+    response = RedirectResponse(str(pending.get("n") or f"{gitea}/"), 303)
+    response.set_cookie(SIGNED_IN, seal({"u": person}, SIGNED_IN_SECONDS), max_age=SIGNED_IN_SECONDS,
+                        path="/domains", secure=True, httponly=True, samesite="lax")
+    response.delete_cookie(SIGNING_IN, path="/domains", secure=True, httponly=True, samesite="lax")
+    return response

mcp/tea_mcp/server.py

@@ -28,7 +28,7 @@ from starlette.requests import Request
 from starlette.responses import JSONResponse
 from starlette.routing import Route
 
-from .domains import domains_http
+from .domains import callback_http, domains_http, sign_in_http
 
 
 def required_env(name: str) -> str:
@@ -673,7 +673,10 @@ async def healthz(_: Request) -> JSONResponse:
 app = mcp.http_app(path="/mcp", transport="streamable-http", stateless_http=True, json_response=True,
                    allowed_hosts=ALLOWED_HOSTS)
 app.routes.append(Route("/healthz", healthz, methods=["GET"]))
-app.routes.append(Route("/domains", domains_http, methods=["GET", "OPTIONS"]))  # the dashboard's Domains tab
+# The dashboard's Domains tab: its search, and the browser sign-in it needs (this server's Gitea OAuth app).
+app.routes.append(Route("/domains", domains_http, methods=["GET", "OPTIONS"]))
+app.routes.append(Route("/domains/sign-in", sign_in_http, methods=["GET"]))
+app.routes.append(Route("/domains/callback", callback_http, methods=["GET"]))
 
 if __name__ == "__main__":
     uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("PORT", "8080")), proxy_headers=True, forwarded_allow_ips="*")