diff --git a/server.py b/server.py index cf3958b..4e0eace 100644 --- a/server.py +++ b/server.py @@ -47,7 +47,7 @@ class V(IntrospectionTokenVerifier): return access -def auth(base: str) -> OAuthProxy: +def auth(base: str, resource: str | None = None) -> OAuthProxy: verifier = V( introspection_url=GITEA + "/login/oauth/introspect", client_id=CID, @@ -62,7 +62,7 @@ def auth(base: str) -> OAuthProxy: upstream_client_secret=SECRET, token_verifier=verifier, base_url=base, - resource_base_url=base, + resource_base_url=resource or base, valid_scopes=["openid", "profile", "email"], forward_pkce=True, forward_resource=False, @@ -124,43 +124,6 @@ def mine(value: str) -> dict[str, Any]: return data -def _oauth_apps() -> list[dict[str, Any]]: - if not GITEA_TOKEN: - raise RuntimeError("GITEA_TOKEN is required for Phast provisioning") - with httpx.Client(timeout=10) as client: - response = client.get( - GITEA + "/api/v1/user/applications/oauth2", - headers={"Authorization": "token " + GITEA_TOKEN, "Accept": "application/json"}, - ) - response.raise_for_status() - return list(response.json() or []) - - -def ensure_oauth_callback(value: str) -> str: - value = slug(value) - callback = f"https://{value}.phast.app/auth/callback" - apps = [app for app in _oauth_apps() if str(app.get("client_id") or "") == CID] - if len(apps) != 1: - raise RuntimeError("could not uniquely resolve the shared Phast OAuth application") - app = apps[0] - redirects = list(app.get("redirect_uris") or []) - if callback in redirects: - return callback - redirects.append(callback) - with httpx.Client(timeout=10) as client: - response = client.patch( - GITEA + f"/api/v1/user/applications/oauth2/{app['id']}", - headers={"Authorization": "token " + GITEA_TOKEN, "Accept": "application/json"}, - json={ - "name": app.get("name") or "Phast", - "redirect_uris": sorted(set(redirects)), - "confidential_client": bool(app.get("confidential_client", True)), - }, - ) - response.raise_for_status() - return callback - - def create(value: str, name: str = "") -> dict[str, Any]: value = slug(value) if path(value).exists(): @@ -356,7 +319,7 @@ def inst(value: str): data["name"], version="0.2.0", instructions="Portable personal Phast MCP.", - auth=auth("https://" + data["host"]), + auth=auth("https://my.phast.app", "https://" + data["host"]), mask_error_details=True, )