Harden personal Phast MCP provisioning

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
+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"]
+DATA = Path(os.getenv("PHAST_DATA_DIR", "/data/phast"))
-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])?$")
+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):
+    async def verify_token(self, token):
-  a=await super().verify_token(t)
+        access = await super().verify_token(token)
-  if not a:return None
+        if not access:
-  async with httpx.AsyncClient(timeout=10) as c:r=await c.get(GITEA+"/login/oauth/userinfo",headers={"Authorization":"Bearer "+t})
+            return None
-  if r.status_code!=200:return None
+        async with httpx.AsyncClient(timeout=10) as client:
-  u=str(r.json().get("preferred_username") or "")
+            response = await client.get(
-  if u not in USERS:return None
+                GITEA + "/login/oauth/userinfo",
-  a.claims={**(a.claims or {}),"login":u};return a
+                headers={"Authorization": "Bearer " + token},
-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"])
+        if response.status_code != 200:
- 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)
+            return None
-def actor():
+        login = str(response.json().get("preferred_username") or "")
- a=get_access_token();u=str(((a.claims if a else {}) or {}).get("login") or "")
+        if login not in USERS:
- if u not in USERS:raise RuntimeError("actor not enabled")
+            return None
- return u
+        access.claims = {**(access.claims or {}), "login": login}
-def slug(s):
+        return access
- s=s.strip().lower()
+
- if not RX.fullmatch(s) or s in {"my","www","mcp"}:raise ValueError("invalid slug")
+
- return s
+def auth(base: str) -> OAuthProxy:
-def path(s):return DATA/slug(s)/"phast.json"
+    verifier = V(
-def read(s):
+        introspection_url=GITEA + "/login/oauth/introspect",
- p=path(s)
+        client_id=CID,
- if not p.exists():raise KeyError("unknown Phast")
+        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(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):
+def write(value: str, data: dict[str, Any]) -> dict[str, Any]:
- d=read(s)
+    p = path(value)
- if d.get("owner")!=actor():raise PermissionError("not your Phast")
+    p.parent.mkdir(parents=True, exist_ok=True)
- return d
+    try:
-def create(s,name=""):
+        p.parent.chmod(0o700)
- s=slug(s)
+    except OSError:
- if path(s).exists():raise ValueError("already exists")
+        pass
- 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":[]}}})
+    tmp = p.with_suffix(".tmp")
-def prefab(d):
+    tmp.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n")
- ws=d["prefab"]["home"]["widgets"]
+    tmp.chmod(0o600)
- with PrefabApp(title=d["name"],css_class="max-w-4xl mx-auto p-6",mode="light") as a:
+    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(d["name"],level=1);Badge(str(len(d["tools"]["installed"]))+" tools",variant="secondary")
+            with Row(justify="between", align="center"):
-   Text("https://"+d["host"]+"/mcp",code=True)
+                Heading(data["name"], level=1)
-   if not ws:
+                Badge(str(len(data["tools"]["installed"])) + " tools", variant="secondary")
-    with Card(css_class="p-5"):Heading("Blank home",level=2);Text("Add a provider or widget from Chat or my.phast.app.")
+            Text("https://" + data["host"] + "/mcp", code=True)
-   for w in ws:
+            if not widgets:
-    with Card(css_class="p-5"):Heading(str(w.get("title") or w["id"]),level=2);Text(str(w.get("description","")))
+                with Card(css_class="p-5"):
- return a
+                    Heading("Blank home", level=2)
-def human(d):
+                    Text("Add a provider or widget from Chat or my.phast.app.")
- ws=d["prefab"]["home"]["widgets"];cards="".join("<article><h2>"+str(w.get("title") or w["id"])+"</h2><p>"+str(w.get("description",""))+"</p></article>" for w in ws) or "<article><h2>Blank home</h2><p>Add a widget from Chat or my.phast.app.</p></article>"
+            for widget in widgets:
- return '<!doctype html><meta name="viewport" content="width=device-width"><title>'+d["name"]+' - phast</title><style>body{margin:0;background:#f5f3ee;color:#121211;font:14px system-ui}main{max-width:1000px;margin:auto;padding:32px}header{display:flex;justify-content:space-between}b{font-size:28px}em{color:#ff4d00}h1{font-size:72px;letter-spacing:-5px;margin:80px 0 40px}section{display:grid;grid-template-columns:repeat(3,1fr);gap:14px}article{border:1px solid #d5d1c8;padding:18px;min-height:150px}p{color:#77746c}@media(max-width:700px){section{grid-template-columns:1fr}}</style><main><header><b>phast<em>.</em></b><code>'+d["host"]+'/mcp</code></header><h1>'+d["name"]+'</h1><section>'+cards+'</section></main>'
+                with Card(css_class="p-5"):
-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)
+                    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(
+        "<article><h2>"
+        + str(widget.get("title") or widget["id"])
+        + "</h2><p>"
+        + str(widget.get("description", ""))
+        + "</p></article>"
+        for widget in widgets
+    ) or "<article><h2>Blank home</h2><p>Add a widget from Chat or my.phast.app.</p></article>"
+    return (
+        '<!doctype html><meta name="viewport" content="width=device-width">'
+        + "<title>"
+        + data["name"]
+        + " - phast</title>"
+        + "<style>body{margin:0;background:#f5f3ee;color:#121211;font:14px system-ui}"
+        + "main{max-width:1000px;margin:auto;padding:32px}header{display:flex;justify-content:space-between}"
+        + "b{font-size:28px}em{color:#ff4d00}h1{font-size:72px;letter-spacing:-5px;margin:80px 0 40px}"
+        + "section{display:grid;grid-template-columns:repeat(3,1fr);gap:14px}"
+        + "article{border:1px solid #d5d1c8;padding:18px;min-height:150px}p{color:#77746c}"
+        + "@media(max-width:700px){section{grid-template-columns:1fr}}</style>"
+        + "<main><header><b>phast<em>.</em></b><code>"
+        + data["host"]
+        + "/mcp</code></header><h1>"
+        + data["name"]
+        + "</h1><section>"
+        + cards
+        + "</section></main>"
+    )
+
+
+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()]}
+    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"}
+    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 providers_search(query: str) -> dict:
+    return {"query": query, "source": "oracle", "status": "adapter-next"}
+
+
+@m.tool()
+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:
- d=mine(slug);ws=d["prefab"]["home"]["widgets"]
+    data = mine(slug)
- if any(w["id"]==widget_id for w in ws):raise ValueError("widget exists")
+    widgets = data["prefab"]["home"]["widgets"]
- w={"id":widget_id,"title":title,"kind":"prefab","description":description};ws.append(w);write(slug,d);return w
+    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:
- 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}
+    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))
+def phast_preview(slug: str) -> PrefabApp:
-mgmt=m.http_app(path="/mcp",transport="streamable-http",stateless_http=True);cache={}
+    return prefab(mine(slug))
-def inst(s):
+
- s=slug(s)
+
- if s in cache:return cache[s]
+mgmt = m.http_app(path="/mcp", transport="streamable-http", stateless_http=True)
- 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)
+cache: dict[str, Any] = {}
- @x.tool()
+started: dict[str, Any] = {}
- def phast_manifest()->dict:return read(s)
+locks: dict[str, asyncio.Lock] = {}
- @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 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
+            try:
-  h=dict(scope.get("headers",[])).get(b"host",b"").decode().split(":")[0].lower();p=scope.get("path","/")
+                await mgmt(scope, receive, send)
-  if h=="my.phast.app":await mgmt(scope,receive,send);return
+            finally:
-  if h.endswith(".phast.app"):
+                for cm in reversed(list(started.values())):
-   s=h[:-10]
+                    try:
-   try:d=read(s)
+                        await cm.__aexit__(None, None, None)
-   except Exception:await JSONResponse({"error":"unknown Phast"},status_code=404)(scope,receive,send);return
+                    except Exception:
-   if p=="/" and scope.get("method")=="GET":await HTMLResponse(human(d))(scope,receive,send);return
+                        pass
-   await inst(s)(scope,receive,send);return
+            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")))
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="::", port=int(os.getenv("PORT", "8081")))