Spaces:
Paused
Paused
| """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 | |