Spaces:
Paused
Paused
File size: 4,950 Bytes
56bf429 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | """Authorize the configured PC MCP from Space Secrets without exposing them."""
from __future__ import annotations
import html
import hmac
import json
import re
from collections.abc import Mapping
from html.parser import HTMLParser
from pathlib import Path
from urllib.parse import parse_qs, urljoin, urlsplit
import httpx
MCP_URL = "https://naldo-mcp-render.onrender.com/mcp"
ISSUER = "https://naldo-mcp-render.onrender.com"
class _LoginForm(HTMLParser):
def __init__(self):
super().__init__(convert_charrefs=True)
self.inputs: dict[str, str] = {}
def handle_starttag(self, tag, attrs):
values = dict(attrs)
if tag == "input" and values.get("name"):
self.inputs[values["name"]] = values.get("value", "")
def _stored_credentials(data_home: Path) -> bool:
try:
data = json.loads((data_home / "opencode" / "mcp-auth.json").read_text())
entry = data.get("naldo_pc", {})
return entry.get("serverUrl") == MCP_URL and bool(entry.get("tokens", {}).get("accessToken"))
except (OSError, ValueError, AttributeError, TypeError):
return False
def _callback_code(response: httpx.Response, expected_state: str) -> str:
if response.is_redirect:
candidates = [urljoin(str(response.url), response.headers.get("location", ""))]
else:
text = html.unescape(response.text).replace("\\/", "/").replace("\\u0026", "&")
candidates = re.findall(r'https?://[^\s"\'<>\\]+', text)
for candidate in candidates:
url = urlsplit(candidate)
params = parse_qs(url.query)
state = params.get("state", [""])[0]
if (url.path == "/mcp/oauth/callback"
and hmac.compare_digest(state, expected_state)
and params.get("code")):
return params["code"][0]
raise RuntimeError("OAuth callback is missing or has a different state")
def ensure_pc_mcp_auth(*, base_url: str, data_home: Path, env: Mapping[str, str]) -> bool:
"""Use normal OAuth/PKCE and CSRF checks; credentials stay in the process."""
web_user = env.get("OPENCODE_SERVER_USERNAME", "")
web_password = env.get("OPENCODE_SERVER_PASSWORD", "")
if not web_user or not web_password:
return False
with httpx.Client(base_url=base_url, auth=(web_user, web_password),
timeout=20, follow_redirects=False) as api:
response = api.get("/mcp")
response.raise_for_status()
status = response.json().get("naldo_pc", {}).get("status")
if status == "connected" and _stored_credentials(data_home):
return True
user = env.get("NALDO_MCP_ADMIN_USER", "")
password = env.get("NALDO_MCP_ADMIN_PASSWORD", "")
if not user or not password:
return False
response = api.post("/mcp/naldo_pc/auth")
response.raise_for_status()
pending = response.json()
authorization_url = pending.get("authorizationUrl", "")
state = pending.get("oauthState", "")
if not authorization_url or not isinstance(state, str) or not state:
raise RuntimeError("MCP did not start an OAuth authorization")
url = urlsplit(authorization_url)
if url.scheme != "https" or url.netloc != urlsplit(ISSUER).netloc:
raise RuntimeError("Unexpected OAuth issuer")
with httpx.Client(timeout=20, follow_redirects=False) as login:
response = login.get(authorization_url)
for _ in range(5):
if not response.is_redirect:
break
destination = urljoin(str(response.url), response.headers.get("location", ""))
target = urlsplit(destination)
if target.scheme != "https" or target.netloc != urlsplit(ISSUER).netloc:
raise RuntimeError("Unexpected OAuth login redirect")
response = login.get(destination)
response.raise_for_status()
form = _LoginForm()
form.feed(response.text)
if not form.inputs.get("state") or not form.inputs.get("csrf"):
raise RuntimeError("OAuth login form has no state or CSRF value")
payload = {key: form.inputs[key] for key in ("state", "csrf")}
payload.update(username=user, password=password)
response = login.post(ISSUER + "/login", data=payload,
headers={"Origin": ISSUER, "Referer": str(response.url)})
if not response.is_redirect:
response.raise_for_status()
code = _callback_code(response, state)
response = api.post("/mcp/naldo_pc/auth/callback", json={"code": code})
response.raise_for_status()
if response.json().get("status") != "connected" or not _stored_credentials(data_home):
raise RuntimeError("MCP authorization did not persist a connected credential")
return True
|