From f5fa61959813157047f5470aa236b732d0604499 Mon Sep 17 00:00:00 2001 From: Boring Date: Sun, 4 Oct 2026 04:41:10 +0000 Subject: [PATCH] Harden personal Phast MCP provisioning --- server.py | 528 ++++++++++++++++++++++++++++++++++++++++++++---------- 1 file changed, 432 insertions(+), 96 deletions(-) diff --git a/server.py b/server.py index 69ea9fe..cf3958b 100644 --- a/server.py +++ b/server.py @@ -1,111 +1,447 @@ from __future__ import annotations -import json,os,re + +import asyncio +import json +import os +import re from pathlib import Path -import httpx,uvicorn +from typing import Any + +import httpx +import uvicorn from fastmcp import FastMCP -from fastmcp.server.dependencies import get_access_token -from fastmcp.server.auth.providers.introspection import IntrospectionTokenVerifier from fastmcp.server.auth.oauth_proxy import OAuthProxy +from fastmcp.server.auth.providers.introspection import IntrospectionTokenVerifier +from fastmcp.server.dependencies import get_access_token from prefab_ui.app import PrefabApp -from prefab_ui.components import Badge,Card,Column,Heading,Row,Text -from starlette.responses import HTMLResponse,JSONResponse -DATA=Path(os.getenv("PHAST_DATA_DIR","/data/phast"));DATA.mkdir(parents=True,exist_ok=True) -GITEA=os.getenv("GITEA_URL","https://teahub.xyz").rstrip("/");CID=os.environ["GITEA_OAUTH_CLIENT_ID"];SECRET=os.environ["GITEA_OAUTH_CLIENT_SECRET"];JWT=os.environ["JWT_SIGNING_KEY"] -USERS={x.strip() for x in os.getenv("ALLOWED_GITEA_USERS","matcha,chai").split(",") if x.strip()};RX=re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$") +from prefab_ui.components import Badge, Card, Column, Heading, Row, Text +from starlette.responses import HTMLResponse, JSONResponse, Response + +DATA = Path(os.getenv("PHAST_DATA_DIR", "/data/phast")) +DATA.mkdir(parents=True, exist_ok=True) +GITEA = os.getenv("GITEA_URL", "https://teahub.xyz").rstrip("/") +CID = os.environ["GITEA_OAUTH_CLIENT_ID"] +SECRET = os.environ["GITEA_OAUTH_CLIENT_SECRET"] +JWT = os.environ["JWT_SIGNING_KEY"] +GITEA_TOKEN = os.getenv("GITEA_TOKEN", "") +USERS = {x.strip() for x in os.getenv("ALLOWED_GITEA_USERS", "matcha,chai").split(",") if x.strip()} +RX = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$") + + class V(IntrospectionTokenVerifier): - async def verify_token(self,t): - a=await super().verify_token(t) - if not a:return None - async with httpx.AsyncClient(timeout=10) as c:r=await c.get(GITEA+"/login/oauth/userinfo",headers={"Authorization":"Bearer "+t}) - if r.status_code!=200:return None - u=str(r.json().get("preferred_username") or "") - if u not in USERS:return None - a.claims={**(a.claims or {}),"login":u};return a -def auth(base): - v=V(introspection_url=GITEA+"/login/oauth/introspect",client_id=CID,client_secret=SECRET,client_auth_method="client_secret_basic",required_scopes=["openid"]) - return OAuthProxy(upstream_authorization_endpoint=GITEA+"/login/oauth/authorize",upstream_token_endpoint=GITEA+"/login/oauth/access_token",upstream_client_id=CID,upstream_client_secret=SECRET,token_verifier=v,base_url=base,resource_base_url=base,valid_scopes=["openid","profile","email"],forward_pkce=True,forward_resource=False,token_endpoint_auth_method="client_secret_basic",jwt_signing_key=JWT,require_authorization_consent="remember",enable_cimd=True) -def actor(): - a=get_access_token();u=str(((a.claims if a else {}) or {}).get("login") or "") - if u not in USERS:raise RuntimeError("actor not enabled") - return u -def slug(s): - s=s.strip().lower() - if not RX.fullmatch(s) or s in {"my","www","mcp"}:raise ValueError("invalid slug") - return s -def path(s):return DATA/slug(s)/"phast.json" -def read(s): - p=path(s) - if not p.exists():raise KeyError("unknown Phast") - return json.loads(p.read_text()) -def write(s,d): - p=path(s);p.parent.mkdir(parents=True,exist_ok=True);q=p.with_suffix(".tmp");q.write_text(json.dumps(d,indent=2,sort_keys=True)+"\n");q.replace(p);return d -def mine(s): - d=read(s) - if d.get("owner")!=actor():raise PermissionError("not your Phast") - return d -def create(s,name=""): - s=slug(s) - if path(s).exists():raise ValueError("already exists") - return write(s,{"schema_version":1,"kind":"phast","slug":s,"name":name.strip() or s,"owner":actor(),"host":s+".phast.app","surfaces":{"human":"/","mcp":"/mcp"},"tools":{"installed":[]},"connections":[],"prefab":{"version":1,"home":{"mode":"grid","edit":True,"widgets":[]}}}) -def prefab(d): - ws=d["prefab"]["home"]["widgets"] - with PrefabApp(title=d["name"],css_class="max-w-4xl mx-auto p-6",mode="light") as a: - with Column(gap=5): - with Row(justify="between",align="center"):Heading(d["name"],level=1);Badge(str(len(d["tools"]["installed"]))+" tools",variant="secondary") - Text("https://"+d["host"]+"/mcp",code=True) - if not ws: - with Card(css_class="p-5"):Heading("Blank home",level=2);Text("Add a provider or widget from Chat or my.phast.app.") - for w in ws: - with Card(css_class="p-5"):Heading(str(w.get("title") or w["id"]),level=2);Text(str(w.get("description",""))) - return a -def human(d): - ws=d["prefab"]["home"]["widgets"];cards="".join("

"+str(w.get("title") or w["id"])+"

"+str(w.get("description",""))+"

" for w in ws) or "

Blank home

Add a widget from Chat or my.phast.app.

" - return ''+d["name"]+' - phast
phast.'+d["host"]+'/mcp

'+d["name"]+'

'+cards+'
' -m=FastMCP("My Phast",version="0.1.0",instructions="Create and manage portable Phast MCP tool belts and Prefab homes.",auth=auth("https://my.phast.app"),mask_error_details=True) + async def verify_token(self, token): + access = await super().verify_token(token) + if not access: + return None + async with httpx.AsyncClient(timeout=10) as client: + response = await client.get( + GITEA + "/login/oauth/userinfo", + headers={"Authorization": "Bearer " + token}, + ) + if response.status_code != 200: + return None + login = str(response.json().get("preferred_username") or "") + if login not in USERS: + return None + access.claims = {**(access.claims or {}), "login": login} + return access + + +def auth(base: str) -> OAuthProxy: + verifier = V( + introspection_url=GITEA + "/login/oauth/introspect", + client_id=CID, + client_secret=SECRET, + client_auth_method="client_secret_basic", + required_scopes=["openid"], + ) + return OAuthProxy( + upstream_authorization_endpoint=GITEA + "/login/oauth/authorize", + upstream_token_endpoint=GITEA + "/login/oauth/access_token", + upstream_client_id=CID, + upstream_client_secret=SECRET, + token_verifier=verifier, + base_url=base, + resource_base_url=base, + valid_scopes=["openid", "profile", "email"], + forward_pkce=True, + forward_resource=False, + token_endpoint_auth_method="client_secret_basic", + jwt_signing_key=JWT, + require_authorization_consent="remember", + enable_cimd=True, + ) + + +def actor() -> str: + access = get_access_token() + login = str(((access.claims if access else {}) or {}).get("login") or "") + if login not in USERS: + raise RuntimeError("actor not enabled") + return login + + +def slug(value: str) -> str: + value = value.strip().lower() + if not RX.fullmatch(value) or value in {"my", "www", "mcp"}: + raise ValueError("invalid slug") + return value + + +def path(value: str) -> Path: + return DATA / slug(value) / "phast.json" + + +def read(value: str) -> dict[str, Any]: + p = path(value) + if not p.exists(): + raise KeyError("unknown Phast") + return json.loads(p.read_text()) + + +def write(value: str, data: dict[str, Any]) -> dict[str, Any]: + p = path(value) + p.parent.mkdir(parents=True, exist_ok=True) + try: + p.parent.chmod(0o700) + except OSError: + pass + tmp = p.with_suffix(".tmp") + tmp.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n") + tmp.chmod(0o600) + tmp.replace(p) + try: + p.chmod(0o600) + except OSError: + pass + return data + + +def mine(value: str) -> dict[str, Any]: + data = read(value) + if data.get("owner") != actor(): + raise PermissionError("not your Phast") + 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(): + raise ValueError("already exists") + # Provision auth before durable state so a failed OAuth mutation cannot leave + # a half-created personal MCP behind. + ensure_oauth_callback(value) + return write( + value, + { + "schema_version": 1, + "kind": "phast", + "slug": value, + "name": name.strip() or value, + "owner": actor(), + "host": value + ".phast.app", + "surfaces": {"human": "/", "mcp": "/mcp"}, + "tools": {"installed": []}, + "connections": [], + "prefab": {"version": 1, "home": {"mode": "grid", "edit": True, "widgets": []}}, + }, + ) + + +def prefab(data: dict[str, Any]) -> PrefabApp: + widgets = data["prefab"]["home"]["widgets"] + with PrefabApp(title=data["name"], css_class="max-w-4xl mx-auto p-6", mode="light") as app: + with Column(gap=5): + with Row(justify="between", align="center"): + Heading(data["name"], level=1) + Badge(str(len(data["tools"]["installed"])) + " tools", variant="secondary") + Text("https://" + data["host"] + "/mcp", code=True) + if not widgets: + with Card(css_class="p-5"): + Heading("Blank home", level=2) + Text("Add a provider or widget from Chat or my.phast.app.") + for widget in widgets: + with Card(css_class="p-5"): + Heading(str(widget.get("title") or widget["id"]), level=2) + Text(str(widget.get("description", ""))) + return app + + +def human(data: dict[str, Any]) -> str: + widgets = data["prefab"]["home"]["widgets"] + cards = "".join( + "

" + + str(widget.get("title") or widget["id"]) + + "

" + + str(widget.get("description", "")) + + "

" + for widget in widgets + ) or "

Blank home

Add a widget from Chat or my.phast.app.

" + return ( + '' + + "" + + data["name"] + + " - phast" + + "" + + "
phast." + + data["host"] + + "/mcp

" + + data["name"] + + "

" + + cards + + "
" + ) + + +m = FastMCP( + "My Phast", + version="0.2.0", + instructions="Create and manage portable Phast MCP tool belts and Prefab homes.", + auth=auth("https://my.phast.app"), + mask_error_details=True, +) + + @m.tool() -def phast_status()->dict:return {"ok":True,"actor":actor()} +def phast_status() -> dict: + return {"ok": True, "actor": actor(), "version": "0.2.0"} + + @m.tool() -def phasts_list()->dict: - return {"phasts":[d for p in DATA.glob("*/phast.json") if (d:=json.loads(p.read_text())).get("owner")==actor()]} +def phasts_list() -> dict: + current = actor() + rows = [] + for p in DATA.glob("*/phast.json"): + try: + data = json.loads(p.read_text()) + except Exception: + continue + if data.get("owner") == current: + rows.append(data) + return {"phasts": sorted(rows, key=lambda item: item["slug"])} + + @m.tool() -def phasts_get(slug:str)->dict:return mine(slug) +def phasts_get(slug: str) -> dict: + return mine(slug) + + @m.tool() -def phasts_create(slug:str,name:str="")->dict: - d=create(slug,name);return {"phast":d,"human_url":"https://"+d["host"]+"/","mcp_url":"https://"+d["host"]+"/mcp"} +def phasts_create(slug: str, name: str = "") -> dict: + data = create(slug, name) + return { + "phast": data, + "human_url": "https://" + data["host"] + "/", + "mcp_url": "https://" + data["host"] + "/mcp", + } + + @m.tool() -def providers_search(query:str)->dict:return {"query":query,"source":"oracle","status":"adapter-next"} +def phasts_update(slug: str, name: str) -> dict: + data = mine(slug) + clean = name.strip() + if not clean: + raise ValueError("name is required") + data["name"] = clean + return write(slug, data) + + @m.tool() -def widgets_add(slug:str,widget_id:str,title:str,description:str="")->dict: - d=mine(slug);ws=d["prefab"]["home"]["widgets"] - if any(w["id"]==widget_id for w in ws):raise ValueError("widget exists") - w={"id":widget_id,"title":title,"kind":"prefab","description":description};ws.append(w);write(slug,d);return w +def providers_search(query: str) -> dict: + return {"query": query, "source": "oracle", "status": "adapter-next"} + + @m.tool() -def widgets_remove(slug:str,widget_id:str)->dict: - d=mine(slug);d["prefab"]["home"]["widgets"]=[w for w in d["prefab"]["home"]["widgets"] if w["id"]!=widget_id];write(slug,d);return {"removed":widget_id} +def widgets_list(slug: str) -> dict: + return {"widgets": list(mine(slug)["prefab"]["home"]["widgets"])} + + +@m.tool() +def widgets_add(slug: str, widget_id: str, title: str, description: str = "") -> dict: + data = mine(slug) + widgets = data["prefab"]["home"]["widgets"] + if any(widget["id"] == widget_id for widget in widgets): + raise ValueError("widget exists") + widget = {"id": widget_id, "title": title, "kind": "prefab", "description": description} + widgets.append(widget) + write(slug, data) + return widget + + +@m.tool() +def widgets_update(slug: str, widget_id: str, title: str = "", description: str = "") -> dict: + data = mine(slug) + for widget in data["prefab"]["home"]["widgets"]: + if widget["id"] == widget_id: + if title.strip(): + widget["title"] = title.strip() + if description: + widget["description"] = description + write(slug, data) + return widget + raise KeyError("unknown widget") + + +@m.tool() +def widgets_remove(slug: str, widget_id: str) -> dict: + data = mine(slug) + widgets = data["prefab"]["home"]["widgets"] + kept = [widget for widget in widgets if widget["id"] != widget_id] + if len(kept) == len(widgets): + raise KeyError("unknown widget") + data["prefab"]["home"]["widgets"] = kept + write(slug, data) + return {"removed": widget_id} + + @m.tool(app=True) -def phast_preview(slug:str)->PrefabApp:return prefab(mine(slug)) -mgmt=m.http_app(path="/mcp",transport="streamable-http",stateless_http=True);cache={} -def inst(s): - s=slug(s) - if s in cache:return cache[s] - d=read(s);x=FastMCP(d["name"],version="0.1.0",instructions="Portable personal Phast MCP.",auth=auth("https://"+d["host"]),mask_error_details=True) - @x.tool() - def phast_manifest()->dict:return read(s) - @x.tool(app=True) - def home_app()->PrefabApp:return prefab(read(s)) - cache[s]=x.http_app(path="/mcp",transport="streamable-http",stateless_http=True);return cache[s] +def phast_preview(slug: str) -> PrefabApp: + return prefab(mine(slug)) + + +mgmt = m.http_app(path="/mcp", transport="streamable-http", stateless_http=True) +cache: dict[str, Any] = {} +started: dict[str, Any] = {} +locks: dict[str, asyncio.Lock] = {} + + +def inst(value: str): + value = slug(value) + if value in cache: + return cache[value] + data = read(value) + personal = FastMCP( + data["name"], + version="0.2.0", + instructions="Portable personal Phast MCP.", + auth=auth("https://" + data["host"]), + mask_error_details=True, + ) + + @personal.tool() + def phast_status() -> dict: + owned = mine(value) + return { + "ok": True, + "slug": value, + "owner": owned["owner"], + "tools": len(owned["tools"]["installed"]), + "widgets": len(owned["prefab"]["home"]["widgets"]), + } + + @personal.tool() + def phast_manifest() -> dict: + return mine(value) + + @personal.tool(app=True) + def home_app() -> PrefabApp: + return prefab(mine(value)) + + cache[value] = personal.http_app(path="/mcp", transport="streamable-http", stateless_http=True) + return cache[value] + + +async def ready_inst(value: str): + value = slug(value) + app = inst(value) + if value in started: + return app + lock = locks.setdefault(value, asyncio.Lock()) + async with lock: + if value not in started: + cm = app.router.lifespan_context(app) + await cm.__aenter__() + started[value] = cm + return app + + class Router: - async def __call__(self,scope,receive,send): - if scope.get("type")=="lifespan": - await mgmt(scope,receive,send);return - h=dict(scope.get("headers",[])).get(b"host",b"").decode().split(":")[0].lower();p=scope.get("path","/") - if h=="my.phast.app":await mgmt(scope,receive,send);return - if h.endswith(".phast.app"): - s=h[:-10] - try:d=read(s) - except Exception:await JSONResponse({"error":"unknown Phast"},status_code=404)(scope,receive,send);return - if p=="/" and scope.get("method")=="GET":await HTMLResponse(human(d))(scope,receive,send);return - await inst(s)(scope,receive,send);return - await JSONResponse({"error":"unknown host"},status_code=404)(scope,receive,send) -app=Router() -if __name__=="__main__":uvicorn.run(app,host="::",port=int(os.getenv("PORT","8081"))) + async def __call__(self, scope, receive, send): + if scope.get("type") == "lifespan": + try: + await mgmt(scope, receive, send) + finally: + for cm in reversed(list(started.values())): + try: + await cm.__aexit__(None, None, None) + except Exception: + pass + return + + host = dict(scope.get("headers", [])).get(b"host", b"").decode().split(":")[0].lower() + request_path = scope.get("path", "/") + method = scope.get("method", "GET") + + if host == "my.phast.app": + await mgmt(scope, receive, send) + return + + if host.endswith(".phast.app"): + value = host[:-10] + try: + data = read(value) + except Exception: + await JSONResponse({"error": "unknown Phast"}, status_code=404)(scope, receive, send) + return + + if request_path == "/" and method in {"GET", "HEAD"}: + body = human(data) + if method == "HEAD": + await Response(content=b"", media_type="text/html", status_code=200)(scope, receive, send) + else: + await HTMLResponse(body)(scope, receive, send) + return + + personal = await ready_inst(value) + await personal(scope, receive, send) + return + + await JSONResponse({"error": "unknown host"}, status_code=404)(scope, receive, send) + + +app = Router() + +if __name__ == "__main__": + uvicorn.run(app, host="::", port=int(os.getenv("PORT", "8081")))