From c1935a97e2a2d768e3402141fcf0fa13278e5be5 Mon Sep 17 00:00:00 2001 From: runyaga Date: Tue, 6 Oct 2026 02:02:13 -0500 Subject: [PATCH] feat: host-side forge proxy keeps git-forge OAuth tokens out of workspaces klangkd runs the PKCE authorization for a proxied forge on the host, keeps the access and refresh tokens per workspace and forge host, and injects them into a reverse proxy that containers reach at /forge-proxy// with their workspace JWT. The browser-delegate relay refuses container-built authorization flows, PAT prompts and credential-cache reads for a proxied forge, so a container cannot obtain a forge credential through the bridge either. Policy is explicit and refuses everything by default: - forge_proxy_features: read, git-read, git-push, issues, issue-edit, site-create, collaborators, api-passthrough, or '*'. - forge_proxy_allowed_repos: owner/repo, owner/*, or '*', with ':ro' (reads only) and ':issues' (reads plus issue work) suffixes. - forge_proxy_template_repos / forge_proxy_template_owners / forge_proxy_collaborator_accounts for site creation from a template and read-only collaborators. A push that deletes a ref is refused, compressed pushes are refused so the ref updates stay inspectable, paths with encoding or URL delimiters are refused, and forge error bodies on API routes are replaced with a local error. Provider entries accept an optional OAuth scope. --- docs/changes.md | 7 + .../tools/git-credential-klangk | 123 +- src/klangk/klangk/api/__init__.py | 3 + src/klangk/klangk/api/browser_delegate.py | 49 + src/klangk/klangk/api/forge_proxy.py | 370 ++++ src/klangk/klangk/caddy.py | 17 + src/klangk/klangk/forge_proxy.py | 1194 ++++++++++ src/klangk/klangk/main.py | 3 + src/klangk/klangk/settings.py | 43 + src/klangk/klangkd-tests/tests/test_api.py | 149 ++ .../klangkd-tests/tests/test_api_package.py | 8 +- .../klangkd-tests/tests/test_forge_proxy.py | 1943 +++++++++++++++++ .../tests/test_git_credential.py | 173 ++ 13 files changed, 4077 insertions(+), 5 deletions(-) create mode 100644 src/klangk/klangk/api/forge_proxy.py create mode 100644 src/klangk/klangk/forge_proxy.py create mode 100644 src/klangk/klangkd-tests/tests/test_forge_proxy.py diff --git a/docs/changes.md b/docs/changes.md index 35aca54e4..e45ac1053 100644 --- a/docs/changes.md +++ b/docs/changes.md @@ -851,6 +851,13 @@ sync` report a clear permission-denied error. ### Added +- **Forge proxy for git-forge OAuth tokens.** With `forge_proxy_hosts` + set, klangkd keeps a forge's OAuth tokens on the host and workspaces + reach the forge through `/forge-proxy//` with their workspace + token, so no forge token enters a container. Operations are enabled + explicitly in `forge_proxy_features` and repositories in + `forge_proxy_allowed_repos`; both refuse everything when empty. + - **Quiet Pi startup with on-demand `/header` (#3537).** The per-user `~/.pi/agent/settings.json` provisioned at first workspace login now sets `"quietStartup": true`, so the startup header diff --git a/features/git-credential/tools/git-credential-klangk b/features/git-credential/tools/git-credential-klangk index a87f299c4..588ba5262 100755 --- a/features/git-credential/tools/git-credential-klangk +++ b/features/git-credential/tools/git-credential-klangk @@ -433,6 +433,8 @@ def _absorb_line(cred, wwwauth, line): return True if key == "wwwauth[]": wwwauth.append(value) + elif key == "capability[]": + cred.setdefault("capability", []).append(value) else: cred[key] = value return True @@ -1192,6 +1194,108 @@ def _pat_dialog_get(cred, browser_id): return username, password, None +def _egress_netloc(): + """host:port of the egress endpoint (the bridge URL's netloc).""" + return urllib.parse.urlsplit(BRIDGE_URL).netloc.lower() if BRIDGE_URL else "" + + +def is_forge_proxy_request(cred): + """True for git requests to the egress /forge-proxy/ route. + + git rewrites https:///... to the egress URL (url.insteadOf), + so the credential request's host is the egress endpoint itself. + """ + netloc = _egress_netloc() + return ( + bool(netloc) + and cred.get("host", "").lower() == netloc + and cred.get("path", "").startswith("forge-proxy/") + ) + + +def forge_proxy_output(cred): + """Bearer workspace-JWT answer for the forge proxy, or None. + + The proxy authenticates the workspace; the forge token stays on the + host. Needs git's ``authtype`` capability (git >= 2.46). + """ + if "authtype" not in cred.get("capability", []): + return None + token = _get_workspace_token() + if not token: + return None + return "\n".join( + [ + "capability[]=authtype", + "authtype=Bearer", + f"credential={token}", + "ephemeral=true", + ] + ) + + +def _post_egress_json(path, timeout=TIMEOUT): + """POST/GET to an egress path with the workspace JWT; parsed JSON.""" + req = urllib.request.Request( + f"{BRIDGE_URL}{path}", + data=b"{}" if path.endswith("/start") else None, + headers=_bridge_headers(), + method="POST" if path.endswith("/start") else "GET", + ) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + return json.loads(resp.read()) + except urllib.error.HTTPError as e: + try: + return json.loads(e.read() or b"null") or {"detail": f"HTTP {e.code}"} + except (json.JSONDecodeError, ValueError, OSError): + return {"detail": f"HTTP {e.code}"} + except (urllib.error.URLError, OSError, json.JSONDecodeError, ValueError) as e: + return {"detail": str(e)} + + +def _configure_forge_rewrite(host): + """Route https:/// through the egress forge proxy for git.""" + proxy = f"{BRIDGE_URL}/forge-proxy/{host}/" + for args in ( + [f"url.{proxy}.insteadOf", f"https://{host}/"], + # The path tells this helper a request is for the forge proxy. + [f"credential.{BRIDGE_URL}.useHttpPath", "true"], + ): + subprocess.run(["git", "config", "--global", *args], check=False) + + +def run_forge_auth(host, browser_id): + """Authorize this workspace for a proxied forge; the token stays on + the host. Returns a process exit code.""" + host = _normalize_host(host) + status = _post_egress_json(f"/forge-proxy/{host}/_klangk/status") + if status.get("connected"): + _configure_forge_rewrite(host) + sys.stdout.write(f"already authorized for {host} as {status.get('login')}\n") + return 0 + start = _post_egress_json(f"/forge-proxy/{host}/_klangk/oauth/start") + txn_id = start.get("txn_id") + if not txn_id: + sys.stderr.write(f"forge-auth: {start.get('detail', 'cannot start')}\n") + return 1 + sys.stderr.write("Approve the authorization in the browser tab...\n") + resp = post_bridge( + "auth_flow_start", + {"protocol": "https", "host": host}, + browser_id, + timeout=GET_TIMEOUT, + extra={"txn_id": txn_id}, + ) + answer = _unwrap_bridge_response(resp or {}) + if answer.get("status") != "connected": + sys.stderr.write(f"forge-auth: {answer.get('error') or answer.get('detail') or 'not authorized'}\n") + return 1 + _configure_forge_rewrite(host) + sys.stdout.write(f"authorized {host} as {answer.get('login')}; git and the API now go through the forge proxy\n") + return 0 + + def _run_get(cred, browser_id): """Serve a get operation: provider flow or PAT dialog, then the git-credential protocol output.""" @@ -1210,12 +1314,27 @@ def main(): browser_id = _get_browser_id() _log_startup(operation, browser_id) - if not _bridge_ready(browser_id): - sys.exit(1) + if operation == "forge-auth": + host = sys.argv[2] if len(sys.argv) > 2 else "" + if not host or not _bridge_ready(browser_id): + sys.stderr.write("usage: git-credential-klangk forge-auth \n") + sys.exit(2) + sys.exit(run_forge_auth(host, browser_id)) cred = read_credential_input() _debug(f"cred input: {_redact(cred)}") + if is_forge_proxy_request(cred): + # Never a forge credential: the workspace JWT for the egress proxy. + if operation == "get": + out = forge_proxy_output(cred) + if out: + sys.stdout.write(out + "\n") + return + + if not _bridge_ready(browser_id): + sys.exit(1) + if operation == "get": _run_get(cred, browser_id) elif operation in ("store", "erase"): diff --git a/src/klangk/klangk/api/__init__.py b/src/klangk/klangk/api/__init__.py index 555bc6df3..f09e845c6 100644 --- a/src/klangk/klangk/api/__init__.py +++ b/src/klangk/klangk/api/__init__.py @@ -63,6 +63,7 @@ admin as _admin_routes, auth as _auth_routes, browser_delegate as _browser_routes, + forge_proxy as _forge_proxy_routes, llm_proxy as _llm_proxy_routes, resources as _resources_routes, workspaces as _workspace_routes, @@ -454,6 +455,8 @@ async def my_permissions( # LLM proxy routes live at /llm-proxy/ (outside /api/v1/) so they are # mounted on root_router, not on the api-prefixed router. #2072 root_router.include_router(_llm_proxy_routes.router) +# Forge proxy routes live at /forge-proxy/ (container egress, workspace JWT). +root_router.include_router(_forge_proxy_routes.router) __all__ = ( diff --git a/src/klangk/klangk/api/browser_delegate.py b/src/klangk/klangk/api/browser_delegate.py index 3228058db..bec8ec53e 100644 --- a/src/klangk/klangk/api/browser_delegate.py +++ b/src/klangk/klangk/api/browser_delegate.py @@ -12,6 +12,8 @@ ) from pydantic import BaseModel +from .. import forge_proxy as forge_proxy_mod +from . import forge_proxy as forge_proxy_routes from .common import get_app_dep from .common import ( require_workspace_token, @@ -83,6 +85,40 @@ def _resolve_bridge_target( return session, target_sock, body.model_dump(exclude={"browser_id"}) +async def _forge_relay(app, workspace_id, session, target_sock, payload): + """Apply the forge proxy's bridge policy; None leaves the plain relay. + + Proxied forges (forge_proxy.py): the container may only start a + klangkd-issued authorization; its code is exchanged here and never + relayed back, and nothing that could hand the container a forge + credential (its own authorize URL, a PAT prompt, a cache read) passes. + """ + forge = getattr(app.state, "forge_proxy", None) + policy = forge.bridge_policy(payload) if forge else "pass" + if policy == "refuse": + raise HTTPException( + status_code=403, + detail="This forge is proxied: run `git-credential-klangk forge-auth `", + ) + if policy != "txn": + return None + + async def dispatch(request, timeout): + return await session.dispatch_browser_request_to( + target_sock, request, timeout=timeout + ) + + client = forge_proxy_routes.http_client_for(app) + try: + return await forge.relay_authorization( + workspace_id, payload, dispatch, client + ) + except forge_proxy_mod.ForgeProxyError as exc: + raise HTTPException(status_code=exc.status, detail=exc.detail) + finally: + await client.aclose() + + @router.post("/browser-delegate") async def browser_delegate( body: BrowserDelegateRequest, @@ -99,6 +135,11 @@ async def browser_delegate( session, target_sock, payload = _resolve_bridge_target( body, app.state.container_registry, app.state.sockets, workspace_id ) + relayed = await _forge_relay( + app, workspace_id, session, target_sock, payload + ) + if relayed is not None: + return relayed # Credential get operations may wait for user interaction (PAT # dialog or OAuth device flow) — allow up to 15 minutes (matching # GitHub's device code expiry). The browser authorization flow @@ -140,6 +181,14 @@ async def browser_delegate_stream( session, target_sock, payload = _resolve_bridge_target( body, app.state.container_registry, app.state.sockets, workspace_id ) + # Proxied-forge credential flows are never streamed: an authorization + # code or token must not reach the container through this path either. + forge = getattr(app.state, "forge_proxy", None) + if forge and forge.bridge_policy(payload) != "pass": + raise HTTPException( + status_code=403, + detail="This forge is proxied: run `git-credential-klangk forge-auth `", + ) # Fetch the workspace so its settings.bridge_timeout override can apply. # One DB lookup per stream request — these are not high-frequency # (one per browser-delegated long-running action from the container). diff --git a/src/klangk/klangk/api/forge_proxy.py b/src/klangk/klangk/api/forge_proxy.py new file mode 100644 index 000000000..3eb40be01 --- /dev/null +++ b/src/klangk/klangk/api/forge_proxy.py @@ -0,0 +1,370 @@ +"""Forge proxy endpoints: allow-listed git and API access with host-held tokens. + +Mounted at ``/forge-proxy/`` (outside ``/api/v1/``) and reached by +containers through the egress caddy, like ``/llm-proxy/``. Every route +requires a **workspace JWT**; the forge token for that workspace is +injected here and never returned. See :mod:`klangk.forge_proxy`. +""" + +import logging +import ssl + +import certifi +import httpx +from fastapi import APIRouter, Depends, Request +from fastapi.responses import JSONResponse, StreamingResponse + +from .. import forge_proxy as fp +from .common import require_workspace_token + +logger = logging.getLogger(__name__) + +router = APIRouter( + prefix="/forge-proxy", + tags=["forge-proxy"], + dependencies=[Depends(require_workspace_token)], +) + +_TIMEOUT = httpx.Timeout(30.0, read=300.0) + + +def ssl_context(settings) -> ssl.SSLContext: + """Default trust plus the optional forge CA (host-side only).""" + ctx = ssl.create_default_context(cafile=certifi.where()) + if settings.forge_proxy_ca_cert: + ctx.load_verify_locations(cafile=settings.forge_proxy_ca_cert) + return ctx + + +def http_client_for(app) -> httpx.AsyncClient: + """An HTTP client for forge calls: no redirects, forge CA trust.""" + return httpx.AsyncClient( + timeout=_TIMEOUT, + verify=ssl_context(app.state.settings), + follow_redirects=False, + ) + + +def http_client(request: Request) -> httpx.AsyncClient: + return http_client_for(request.app) + + +async def _read_capped(request: Request, limit: int) -> bytes: + """The request body, refused once it exceeds ``limit`` bytes.""" + chunks, size = [], 0 + async for chunk in request.stream(): + size += len(chunk) + if size > limit: + raise fp.ForgeProxyError(413, "request body too large") + chunks.append(chunk) + return b"".join(chunks) + + +async def _capped_stream(request: Request, limit: int): + """Stream a git pack upstream without buffering; stop past ``limit``.""" + size = 0 + async for chunk in request.stream(): + size += len(chunk) + if size > limit: + raise fp.ForgeProxyError(413, "request body too large") + yield chunk + + +def _error(exc: fp.ForgeProxyError) -> JSONResponse: + return JSONResponse(status_code=exc.status, content={"detail": exc.detail}) + + +@router.post("/{host}/_klangk/oauth/start") +async def oauth_start( + host: str, + request: Request, + workspace_id: str = Depends(require_workspace_token), +): + """Begin a host-side authorization; returns only an opaque txn_id.""" + try: + txn_id = request.app.state.forge_proxy.start(workspace_id, host) + except fp.ForgeProxyError as exc: + return _error(exc) + return {"txn_id": txn_id} + + +@router.get("/{host}/_klangk/status") +async def oauth_status( + host: str, + request: Request, + workspace_id: str = Depends(require_workspace_token), +): + """Whether this workspace has a forge authorization (no secrets).""" + try: + return request.app.state.forge_proxy.status(workspace_id, host) + except fp.ForgeProxyError as exc: + return _error(exc) + + +@router.get("/{host}/{path:path}") +async def forward_get( + host: str, + path: str, + request: Request, + workspace_id: str = Depends(require_workspace_token), +): + """GET through the forge proxy (allow-listed reads, git refs).""" + return await forward(host, path, request, workspace_id) + + +@router.post("/{host}/{path:path}") +async def forward_post( + host: str, + path: str, + request: Request, + workspace_id: str = Depends(require_workspace_token), +): + """POST through the forge proxy (git packs, issue and comment create).""" + return await forward(host, path, request, workspace_id) + + +@router.put("/{host}/{path:path}") +async def forward_put( + host: str, + path: str, + request: Request, + workspace_id: str = Depends(require_workspace_token), +): + """PUT through the forge proxy (read-only collaborator add).""" + return await forward(host, path, request, workspace_id) + + +@router.delete("/{host}/{path:path}") +async def forward_delete( + host: str, + path: str, + request: Request, + workspace_id: str = Depends(require_workspace_token), +): + """DELETE through the forge proxy (passthrough mode only).""" + return await forward(host, path, request, workspace_id) + + +@router.patch("/{host}/{path:path}") +async def forward_patch( + host: str, + path: str, + request: Request, + workspace_id: str = Depends(require_workspace_token), +): + """PATCH through the forge proxy (issue state only).""" + return await forward(host, path, request, workspace_id) + + +async def forward(host: str, path: str, request: Request, workspace_id: str): + """Forward one allow-listed request with the workspace's forge token.""" + proxy = request.app.state.forge_proxy + try: + route = _route(proxy, host, request) + body = await _body(route, request) + client = http_client(request) + token = await _token(proxy, workspace_id, host, client) + await _check_actor(proxy, route, body, token, client) + except fp.ForgeProxyError as exc: + return _error(exc) + upstream = client.build_request( + request.method, + _upstream_url(proxy, host, route), + headers=_request_headers(request, token), + content=body, + ) + resp = await _send(client, upstream, request.method, route) + if isinstance(resp, JSONResponse): + return resp + logger.info( + "forge-proxy: workspace=%s host=%s %s %s %s/%s -> %s", + workspace_id, + host, + request.method, + route.kind, + route.owner, + route.repo, + resp.status_code, + ) + refused = await _refused(resp, client, proxy, workspace_id, host, token) + return ( + refused + or await _sanitized(resp, client, route) + or _stream(resp, client) + ) + + +async def _sanitized(resp, client, route: fp.Route): + """A local error in place of a forge error body on an API route. + + The forge's own error text never reaches the container, so a body that + echoes request credentials cannot leak them. Git routes keep the forge's + body: git clients print it as the failure reason. + """ + if resp.status_code < 400 or route.kind.startswith("git-"): + return None + await resp.aclose() + await client.aclose() + return JSONResponse( + status_code=resp.status_code, + content={"detail": f"forge refused the request ({resp.status_code})"}, + ) + + +def _route(proxy, host: str, request: Request) -> fp.Route: + """The allow-listed route for the raw (undecoded) request path.""" + proxy.provider(host) + raw = request.scope.get("raw_path", b"").decode("latin-1") + prefix = f"/forge-proxy/{host}" + if not raw.startswith(prefix + "/"): + raise fp.ForgeProxyError(400, "path not allowed") + route = fp.match_route( + request.method, + raw[len(prefix) :], + request.scope.get("query_string", b"").decode("latin-1"), + passthrough=proxy.passthrough(), + ) + proxy.check_route(route) + return route + + +async def _body(route: fp.Route, request: Request): + """A capped stream for git packs; a validated JSON body otherwise.""" + if route.kind == "git-receive": + return await _guarded_push(request) + if route.kind.startswith("git-"): + return _capped_stream(request, fp.MAX_GIT_BODY) + if route.kind.startswith("api-pass-"): + return await _read_capped(request, fp.MAX_PASS_BODY) + return fp.validate_json_body( + route.kind, await _read_capped(request, fp.MAX_JSON_BODY) + ) + + +async def _guarded_push(request: Request): + """Read a push's ref updates, refuse ref deletion, then stream the rest.""" + encoding = (request.headers.get("content-encoding") or "identity").lower() + if encoding != "identity": + raise fp.ForgeProxyError(415, "compressed push requests not allowed") + chunks = request.stream().__aiter__() + head = await _push_head(chunks) + return _replay(head, chunks, fp.MAX_GIT_BODY) + + +async def _push_head(chunks) -> bytes: + """Bytes up to and including the flush-pkt ending the ref updates.""" + buf = b"" + end = None + while end is None: + chunk = await anext(chunks, None) + if chunk is None: + raise fp.ForgeProxyError(400, "malformed push request") + buf += chunk + if len(buf) > fp.MAX_PUSH_COMMANDS: + raise fp.ForgeProxyError(413, "push commands too large") + end = fp.push_commands_end(buf) + fp.check_push_commands(buf, end) + return buf + + +async def _replay(head: bytes, chunks, limit: int): + """Stream the already-read head, then the rest of the body, capped.""" + yield head + size = len(head) + async for chunk in chunks: + size += len(chunk) + if size > limit: + raise fp.ForgeProxyError(413, "request body too large") + yield chunk + + +async def _token(proxy, workspace_id: str, host: str, client): + try: + return await proxy.access_token(workspace_id, host, client) + except BaseException: + await client.aclose() + raise + + +async def _check_actor(proxy, route: fp.Route, body, token, client) -> None: + """Account-dependent policy; closes the client when it refuses.""" + try: + proxy.check_actor(route, body, token.login) + except fp.ForgeProxyError: + await client.aclose() + raise + + +def _upstream_url(proxy, host: str, route: fp.Route) -> str: + url = f"{proxy.origin(host)}{route.path}" + return f"{url}?{route.query}" if route.query else url + + +def _request_headers(request: Request, token) -> dict: + headers = { + k: v + for k, v in request.headers.items() + if k.lower() in fp.FORWARD_REQUEST_HEADERS + } + headers["Authorization"] = f"Bearer {token.access_token}" + return headers + + +async def _send(client, upstream, method: str, route: fp.Route): + """The upstream response, or an error response (client closed).""" + try: + return await client.send(upstream, stream=True) + except fp.ForgeProxyError as exc: + await client.aclose() + return _error(exc) + except httpx.HTTPError as exc: + await client.aclose() + logger.warning( + "forge-proxy: %s %s failed: %s", + method, + route.kind, + type(exc).__name__, + ) + return JSONResponse( + status_code=502, content={"detail": "forge unreachable"} + ) + + +async def _refused(resp, client, proxy, workspace_id: str, host: str, token): + """An error response for an upstream redirect or 401, else None.""" + if not (300 <= resp.status_code < 400 or resp.status_code == 401): + return None + await resp.aclose() + await client.aclose() + if resp.status_code == 401: + proxy.forget_token(workspace_id, host, token) + return _error(fp.ForgeProxyError(401, fp._needs_auth(host))) + return JSONResponse( + status_code=502, content={"detail": "forge redirected the request"} + ) + + +def _stream(resp, client) -> StreamingResponse: + out_headers = { + k: v + for k, v in resp.headers.items() + if k.lower() in fp.FORWARD_RESPONSE_HEADERS + } + + async def body_iter(): + try: + async for chunk in resp.aiter_raw(): + yield chunk + except httpx.HTTPError as exc: + # Headers are already sent: end the body early (git and JSON + # clients see a truncated response) instead of a server error. + logger.warning( + "forge-proxy: upstream body failed: %s", type(exc).__name__ + ) + finally: + await resp.aclose() + await client.aclose() + + return StreamingResponse( + body_iter(), status_code=resp.status_code, headers=out_headers + ) diff --git a/src/klangk/klangk/caddy.py b/src/klangk/klangk/caddy.py index 263289226..103ab821b 100644 --- a/src/klangk/klangk/caddy.py +++ b/src/klangk/klangk/caddy.py @@ -945,6 +945,21 @@ def _build_llm_block(self, upstream: str, guard: str) -> str: " }\n" ) + def _build_forge_block(self, upstream: str, guard: str) -> str: + """The ``/forge-proxy/*`` location (forge_proxy.py). + + Same shape as the LLM block: container-source guard, the + site-level ``forward_auth`` workspace-token check, then the + klangkd backend, which injects the host-held forge token and + enforces the route allow-list. + """ + return ( + "\thandle /forge-proxy/* {\n" + f"{guard}" + f"\t\treverse_proxy {upstream}\n" + "\t}\n" + ) + def _egress_locations(self, upstream: str, container_srcs: str) -> str: """The container-egress locations, shared by headless and full modes. @@ -973,6 +988,7 @@ def _egress_locations(self, upstream: str, container_srcs: str) -> str: not_src_matcher = "" guard = " respond 403\n" llm = self._build_llm_block(upstream, guard) + forge = self._build_forge_block(upstream, guard) # #3473: the matcher must cover BOTH bridge endpoints. A bare # ``handle /api/v1/browser-delegate`` is an exact-path match, so the # streaming sibling ``/api/v1/browser-delegate/stream`` matched no @@ -1019,6 +1035,7 @@ def _egress_locations(self, upstream: str, container_srcs: str) -> str: + delegate_matcher + egress_ws + llm + + forge + delegate + catch_all ) diff --git a/src/klangk/klangk/forge_proxy.py b/src/klangk/klangk/forge_proxy.py new file mode 100644 index 000000000..106466dae --- /dev/null +++ b/src/klangk/klangk/forge_proxy.py @@ -0,0 +1,1194 @@ +"""Host-side custody of git-forge OAuth tokens for workspaces. + +A workspace never holds a forge token. klangkd runs the PKCE flow (the +verifier stays here), exchanges the authorization code when the browser +bridge returns it, keeps the access and refresh tokens per (workspace, +forge host), and injects them into an allow-listed reverse proxy that +containers reach at ``/forge-proxy//...`` with their workspace +JWT. The browser-delegate relay refuses container-built authorization +flows, PAT prompts, and credential-cache reads for proxied forges, so a +container cannot obtain a forge credential through the bridge either. + +Tokens and transactions live in memory (one klangkd process; a restart +means re-authorizing). ``forge_proxy_allowed_repos`` names the +repositories a workspace reaches (``owner/repo``, ``owner/*``, or ``*`` for +all; a ``:ro`` suffix for read-only and ``:issues`` for read-only plus +issue work); with no matching rule a repository is +refused. A push that deletes a ref is refused. +``forge_proxy_template_repos`` names the templates a workspace may +generate a repository from (into the signed-in account, or an owner in +``forge_proxy_template_owners``), and ``forge_proxy_collaborator_accounts`` +names the accounts a workspace may add, read-only, as collaborators on a +repository the signed-in account owns. +""" + +import asyncio +import base64 +import hashlib +import json +import logging +import re +import secrets +import time +import urllib.parse +from dataclasses import dataclass, field + +from . import settings as settings_mod + +logger = logging.getLogger(__name__) + +PROVIDERS_KEY = "KLANGKWS_FEATURE_OAUTH_PROVIDERS" +TXN_TTL_SECONDS = 600 +MAX_PENDING_TXNS_PER_WORKSPACE = 8 +MAX_PENDING_TXNS = 1000 +REFRESH_SKEW_SECONDS = 120 + +_NAME = r"[A-Za-z0-9][A-Za-z0-9_.-]{0,99}" +_REPO = rf"/(?P{_NAME})/(?P{_NAME})" +_NUM = r"(?P[1-9][0-9]{0,9})" + +# (method, compiled path pattern, route kind, allowed query keys). +_ROUTES = [ + ( + "GET", + re.compile(rf"^{_REPO}(?:\.git)?/info/refs$"), + "git-refs", + {"service"}, + ), + ( + "POST", + re.compile(rf"^{_REPO}(?:\.git)?/git-upload-pack$"), + "git-upload", + set(), + ), + ( + "POST", + re.compile(rf"^{_REPO}(?:\.git)?/git-receive-pack$"), + "git-receive", + set(), + ), + ("GET", re.compile(r"^/api/v1/user$"), "api-read", set()), + ("GET", re.compile(rf"^/api/v1/repos{_REPO}$"), "api-read", set()), + ( + "GET", + re.compile(rf"^/api/v1/repos{_REPO}/issues$"), + "api-read", + {"state", "page", "limit", "type"}, + ), + ( + "GET", + re.compile(rf"^/api/v1/repos{_REPO}/issues/{_NUM}$"), + "api-read", + set(), + ), + ( + "GET", + re.compile(rf"^/api/v1/repos{_REPO}/issues/{_NUM}/comments$"), + "api-read", + set(), + ), + ( + "POST", + re.compile(rf"^/api/v1/repos{_REPO}/issues$"), + "issue-create", + set(), + ), + ( + "POST", + re.compile(rf"^/api/v1/repos{_REPO}/issues/{_NUM}/comments$"), + "comment-create", + set(), + ), + ( + "PATCH", + re.compile(rf"^/api/v1/repos{_REPO}/issues/{_NUM}$"), + "issue-state", + set(), + ), + ( + "GET", + re.compile(rf"^/api/v1/repos{_REPO}/(?:labels|milestones)$"), + "api-read", + {"state", "page", "limit"}, + ), + ( + "GET", + re.compile(rf"^/api/v1/repos{_REPO}/(?:assignees|collaborators)$"), + "api-read", + {"page", "limit"}, + ), + ( + "POST", + re.compile(rf"^/api/v1/repos{_REPO}/issues/{_NUM}/labels$"), + "issue-labels", + set(), + ), + ( + "PUT", + re.compile(rf"^/api/v1/repos{_REPO}/issues/{_NUM}/labels$"), + "issue-labels", + set(), + ), + ( + "POST", + re.compile(rf"^/api/v1/repos{_REPO}/generate$"), + "repo-generate", + set(), + ), + ( + "PUT", + re.compile(rf"^/api/v1/repos{_REPO}/collaborators/(?P{_NAME})$"), + "collaborator-add", + set(), + ), +] + +_GIT_SERVICES = {"git-upload-pack", "git-receive-pack"} + +# Percent-encoding, backslashes, control characters and spaces, repeated +# slashes, and "." or ".." segments. +_BAD_PATH = re.compile(r"[%\\#?\x00-\x20\x7f]|//|/\.{1,2}(?:/|$)") + +# JSON write bodies per route kind: (allowed key -> value type, required keys). +_BODY_RULES = { + "issue-create": ( + { + "title": str, + "body": str, + "assignees": list, + "labels": list, + "milestone": int, + }, + {"title"}, + ), + "comment-create": ({"body": str}, {"body"}), + # Issue edits: title, body, open/close, assignees and milestone. + "issue-state": ( + { + "title": str, + "body": str, + "state": str, + "assignees": list, + "milestone": int, + }, + set(), + ), + "issue-labels": ({"labels": list}, {"labels"}), + "repo-generate": ( + { + "owner": str, + "name": str, + "description": str, + "private": bool, + "git_content": bool, + "default_branch": str, + }, + {"owner", "name"}, + ), + "collaborator-add": ({"permission": str}, {"permission"}), +} +# Exact-value checks after the type checks: (kind, key) -> allowed values. +_BODY_VALUES = { + ("issue-state", "state"): {"open", "closed"}, + ("collaborator-add", "permission"): {"read"}, +} +_NAME_ONLY = re.compile(rf"^{_NAME}$") +MAX_ASSIGNEES = 10 +MAX_LABELS = 20 +MAX_JSON_BODY = 64 * 1024 +# Upper bound for one git pack request streamed through the proxy. +MAX_GIT_BODY = 512 * 1024 * 1024 +# Upper bound for the ref-update commands read ahead of a push's pack. +MAX_PUSH_COMMANDS = 1024 * 1024 + +# Features an operator enables in ``forge_proxy_features``. Every route +# belongs to one; a route whose feature is not enabled is refused (403), +# so an empty setting (the default) allows no forge operation at all. +FEATURES = { + "read": "read the account, repositories, issues, comments, labels", + "git-read": "git clone and fetch", + "git-push": "git push (a push that deletes a ref is always refused)", + "issues": "open issues and comments", + "issue-edit": "edit, close, assign and label issues", + "site-create": "generate a repository from a listed template", + "collaborators": "add a listed account as a read-only collaborator", + "api-passthrough": "any other /api/v1/ call; the forge decides", +} +_KIND_FEATURE = { + "api-read": "read", + "git-refs": "git-read", + "git-upload": "git-read", + "git-receive": "git-push", + "issue-create": "issues", + "comment-create": "issues", + "issue-state": "issue-edit", + "issue-labels": "issue-edit", + "repo-generate": "site-create", + "collaborator-add": "collaborators", + "api-pass-read": "api-passthrough", + "api-pass-write": "api-passthrough", +} + + +def parse_features(raw: str) -> set[str]: + """Enabled features; ``*`` enables all. Unknown names are ignored.""" + names = {n.strip().lower() for n in (raw or "").split(",") if n.strip()} + return set(FEATURES) if "*" in names else names & set(FEATURES) + + +def route_feature(route: Route) -> str: + """The feature a route needs (a push advertisement needs git-push).""" + if "service=git-receive-pack" in route.query: + return "git-push" + return _KIND_FEATURE[route.kind] + + +def check_feature(route: Route, enabled: set[str]) -> None: + if route_feature(route) not in enabled: + raise ForgeProxyError( + 403, "operation not enabled through the forge proxy" + ) + + +# Route kinds a ``:issues`` repo rule allows on top of read-only access: +# open issues and comments, open/close an issue, and set its assignees. +ISSUE_KINDS = {"issue-create", "comment-create", "issue-state", "issue-labels"} + +# Route kinds that change the forge; refused for a read-only repo rule. +WRITE_KINDS = { + "git-receive", + "issue-create", + "comment-create", + "issue-state", + "issue-labels", + "api-pass-write", + "repo-generate", + "collaborator-add", +} + +# A pkt-line ref update: " " (oid: 40 or 64 hex). +_PUSH_COMMAND = re.compile( + r"^(?P[0-9a-f]{40}|[0-9a-f]{64}) " + r"(?P[0-9a-f]{40}|[0-9a-f]{64}) (?P\S+)$" +) +_SHALLOW = re.compile(r"^shallow [0-9a-f]{40,64}$") + +# Request headers forwarded upstream; everything else (cookies, the +# workspace JWT, forwarded-identity headers, Sudo) is dropped. +FORWARD_REQUEST_HEADERS = ( + "accept", + "content-type", + "content-encoding", + "git-protocol", + "user-agent", +) +# Response headers returned to the container. +FORWARD_RESPONSE_HEADERS = ( + "content-type", + "content-encoding", + "cache-control", + "expires", + "pragma", + "x-total-count", +) + + +class ForgeProxyError(Exception): + """A refusal with an HTTP status and a sanitized message.""" + + def __init__(self, status: int, detail: str): + super().__init__(detail) + self.status = status + self.detail = detail + + +@dataclass +class Txn: + """A pending authorization: the verifier never leaves klangkd.""" + + txn_id: str + workspace_id: str + host: str + verifier: str + state: str + created: float + in_flight: bool = False + + +@dataclass(eq=False) +class ForgeToken: + """Compared by identity: replacing or forgetting checks it is still + the stored object, so a stale refresh or 401 never clobbers a newer one.""" + + access_token: str + refresh_token: str + expires_at: float + login: str + + +@dataclass +class Route: + kind: str + owner: str + repo: str + path: str + query: str + user: str = "" + + +def s256_challenge(verifier: str) -> str: + """The S256 code_challenge for a verifier (RFC 7636 section 4.2).""" + digest = hashlib.sha256(verifier.encode("ascii")).digest() + return base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii") + + +def _host_of_url(url: str) -> str: + """Lowercased hostname (no port) of a URL, or ''.""" + try: + host = urllib.parse.urlsplit(url).hostname or "" + except ValueError: + return "" + return host.lower().rstrip(".") + + +def _ambiguous_netloc(url: str) -> bool: + """True when a URL's authority could be read differently by a browser + (percent-encoding, userinfo, non-ASCII, unparsable).""" + try: + netloc = urllib.parse.urlsplit(url).netloc + except ValueError: + return True + return "%" in netloc or "@" in netloc or not netloc.isascii() + + +def _norm_host(host: str) -> str: + """Lowercase, no port, no trailing dot.""" + return (host or "").strip().lower().split(":", 1)[0].rstrip(".") + + +def validate_raw_path(raw_path: str) -> str: + """Reject ambiguous or encoded paths before any routing. + + Percent-encoding, backslashes, repeated slashes and dot segments are + all refused, so the path the router matches is exactly the path + forwarded upstream. + """ + if not raw_path.startswith("/"): + raise ForgeProxyError(400, "path must be absolute") + if _BAD_PATH.search(raw_path): + raise ForgeProxyError(400, "path not allowed") + return raw_path + + +def _strict_query(query: str, allowed: set[str], kind: str) -> str: + """Rebuild the query from allowed keys; refuse unknown or repeated keys.""" + pairs = urllib.parse.parse_qsl(query, keep_blank_values=True) + try: + params = _no_duplicate_keys(pairs) + except ValueError: + raise ForgeProxyError(400, "repeated query parameter") + if set(params) - allowed: + raise ForgeProxyError(400, "query parameter not allowed") + if kind == "git-refs" and params.get("service") not in _GIT_SERVICES: + raise ForgeProxyError(400, "service not allowed") + return urllib.parse.urlencode(pairs) + + +# Passthrough mode: any forge API call, decided by the forge's own +# permissions and the token's scopes. These query keys still never pass. +PASS_METHODS = {"GET", "POST", "PUT", "PATCH", "DELETE"} +PASS_DENIED_QUERY = {"access_token", "token", "sudo"} +MAX_PASS_BODY = 25 * 1024 * 1024 +_PASS_REPO = re.compile(rf"^/api/v1/repos{_REPO}(?:/|$)") + + +def match_route( + method: str, raw_path: str, query: str, passthrough: bool = False +) -> Route: + """The allow-listed route for a request, or ForgeProxyError(403). + + With ``passthrough``, an ``/api/v1/`` call no allow-listed route + matches becomes an ``api-pass-read`` or ``api-pass-write`` route. + """ + path = validate_raw_path(raw_path) + try: + kind, allowed, groups = _find_route(method, path) + except ForgeProxyError: + if not (passthrough and path.startswith("/api/v1/")): + raise + return _pass_route(method, path, query) + return Route( + kind=kind, + owner=groups.get("owner", ""), + # Logged only; the forwarded path is the validated raw path. + repo=groups.get("repo", "").removesuffix(".git"), + path=path, + query=_strict_query(query, allowed, kind), + user=groups.get("user", ""), + ) + + +def _pass_route(method: str, path: str, query: str) -> Route: + if method not in PASS_METHODS: + raise ForgeProxyError(405, "method not allowed") + groups = _pass_groups(path) + return Route( + kind="api-pass-read" if method == "GET" else "api-pass-write", + owner=groups.get("owner", ""), + repo=groups.get("repo", ""), + path=path, + query=_pass_query(query), + ) + + +def _pass_groups(path: str) -> dict: + m = _PASS_REPO.match(path) + return m.groupdict() if m else {} + + +def _pass_query(query: str) -> str: + """The query unchanged, unless it carries credentials (400).""" + pairs = urllib.parse.parse_qsl(query, keep_blank_values=True) + if {k.lower() for k, _ in pairs} & PASS_DENIED_QUERY: + raise ForgeProxyError(400, "query parameter not allowed") + return urllib.parse.urlencode(pairs) + + +def _find_route(method: str, path: str): + """(kind, allowed query keys, path groups) of the matching route.""" + for route_method, pattern, kind, allowed in _ROUTES: + m = pattern.match(path) if method == route_method else None + if m: + return kind, allowed, m.groupdict() + raise ForgeProxyError(403, "operation not allowed through the forge proxy") + + +def validate_json_body(kind: str, body: bytes) -> bytes: + """Parse, check and re-serialize a write body for an API route.""" + rules = _BODY_RULES.get(kind) + if rules is None: + return body + if len(body) > MAX_JSON_BODY: + raise ForgeProxyError(413, "request body too large") + data = _json_object(body) + _check_fields(data, *rules) + _check_values(kind, data) + return json.dumps(data).encode() + + +def _json_object(body: bytes) -> dict: + try: + data = json.loads(body, object_pairs_hook=_no_duplicate_keys) + except ValueError: + raise ForgeProxyError(400, "request body must be a JSON object") + if not isinstance(data, dict): + raise ForgeProxyError(400, "request body must be a JSON object") + return data + + +def _check_fields(data: dict, types: dict, required: set) -> None: + keys = set(data) + if not (keys and keys <= set(types) and required <= keys): + raise ForgeProxyError(400, "request body fields not allowed") + _check_types(data, types) + + +def _check_types(data: dict, types: dict) -> None: + if not all(type(v) is types[k] for k, v in data.items()): + raise ForgeProxyError(400, "request body value has the wrong type") + + +def _check_values(kind: str, data: dict) -> None: + for key, value in data.items(): + _check_exact(kind, key, value) + _LIST_CHECKS.get(key, _no_check)(value) + if kind == "repo-generate": + _check_names(data["owner"], data["name"]) + + +def _check_exact(kind: str, key: str, value) -> None: + allowed = _BODY_VALUES.get((kind, key)) + if allowed is not None and value not in allowed: + raise ForgeProxyError(400, f"{key} value not allowed") + + +def _no_check(value) -> None: + """Values with no extra check beyond their type.""" + + +def _check_assignees(logins: list) -> None: + """Assignees: up to MAX_ASSIGNEES account names (strings).""" + ok = len(logins) <= MAX_ASSIGNEES and all( + isinstance(n, str) and _NAME_ONLY.match(n) for n in logins + ) + if not ok: + raise ForgeProxyError(400, "assignees value not allowed") + + +def _check_labels(labels: list) -> None: + """Labels: up to MAX_LABELS label ids (ints) or names (short strings).""" + ok = len(labels) <= MAX_LABELS and all(map(_label_ok, labels)) + if not ok: + raise ForgeProxyError(400, "labels value not allowed") + + +def _label_ok(label) -> bool: + if type(label) is int: + return label > 0 + return isinstance(label, str) and 0 < len(label) <= 100 + + +_LIST_CHECKS = {"assignees": _check_assignees, "labels": _check_labels} + + +def _check_names(*names: str) -> None: + if not all(_NAME_ONLY.match(n) for n in names): + raise ForgeProxyError(400, "repository owner or name not allowed") + + +def _no_duplicate_keys(pairs): + keys = [k for k, _ in pairs] + if len(keys) != len(set(keys)): + raise ValueError("duplicate key") + return dict(pairs) + + +# --- repository allow-list --------------------------------------------------- + + +@dataclass(frozen=True) +class RepoRule: + owner: str + repo: str # a repository name, or "*" for every repo of the owner + read_only: bool + issues: bool = False # read-only, but issues and comments may be opened + + +def parse_repo_rules(raw: str) -> list[RepoRule]: + """Rules from ``forge_proxy_allowed_repos``. + + Unset, empty or unusable entries allow nothing: a repository is + reachable only through a matching rule, and ``*`` matches every one. + """ + entries = [e.strip().lower() for e in (raw or "").split(",")] + return list(filter(None, map(_parse_rule, filter(None, entries)))) + + +_RULE = re.compile( + r"^(?:\*(?:/\*)?|(?P[a-z0-9][a-z0-9_.-]*)/" + r"(?P\*|[a-z0-9][a-z0-9_.-]*))(?P:ro|:issues)?$" +) + + +def _parse_rule(entry: str) -> RepoRule | None: + m = _RULE.match(entry) + if m is None: + return None + return RepoRule( + owner=m.group("owner") or "*", + repo=(m.group("repo") or "*").removesuffix(".git"), + read_only=m.group("mode") is not None, + issues=m.group("mode") == ":issues", + ) + + +def is_write(route: Route) -> bool: + """True for a route that changes the forge (a push advertisement too).""" + return ( + route.kind in WRITE_KINDS or "service=git-receive-pack" in route.query + ) + + +def check_repo(route: Route, rules: list[RepoRule]) -> None: + """Refuse a route outside the allow-list (403). + + Allow-listed account routes pass. A passthrough call whose path names + no repository (search, numeric ids, organizations, repository + creation) passes only with an unrestricted ``*`` rule. + """ + if not route.owner: + _check_unscoped(route, rules) + return + _check_rule(route, _matching_rule(route, rules)) + + +def _check_unscoped(route: Route, rules: list[RepoRule]) -> None: + star = next((r for r in rules if (r.owner, r.repo) == ("*", "*")), None) + if route.kind.startswith("api-pass-"): + _check_rule(route, star) + + +def _check_rule(route: Route, rule: RepoRule | None) -> None: + if rule is None: + raise ForgeProxyError( + 403, "repository not allowed through the forge proxy" + ) + if _read_only_refuses(route, rule): + raise ForgeProxyError( + 403, "repository is read-only through the forge proxy" + ) + + +def _read_only_refuses(route: Route, rule: RepoRule) -> bool: + """True when a read-only rule forbids this write (``:issues`` allows + opening issues and comments).""" + allowed = rule.issues and route.kind in ISSUE_KINDS + return rule.read_only and is_write(route) and not allowed + + +def _matching_rule(route: Route, rules: list[RepoRule]) -> RepoRule | None: + """The exact ``owner/repo`` rule, else ``owner/*``, else ``*``, else None.""" + index = {(r.owner, r.repo): r for r in reversed(rules)} + owner, repo = route.owner.lower(), route.repo.lower() + keys = ((owner, repo), (owner, "*"), ("*", "*")) + return next(filter(None, map(index.get, keys)), None) + + +# --- site creation: templates and collaborators ------------------------------ + + +def parse_names(raw: str) -> set[str]: + """Lowercased entries of a comma-separated setting (``a/b`` or ``a``).""" + return {e.strip().lower() for e in (raw or "").split(",") if e.strip()} + + +def check_template(route: Route, templates: set[str]) -> None: + """Refuse generating from a repository not listed as a template (403).""" + if f"{route.owner}/{route.repo}".lower() not in templates: + raise ForgeProxyError( + 403, "template not allowed through the forge proxy" + ) + + +def check_new_repo( + body: dict, login: str, owners: set[str], rules: list[RepoRule] +) -> None: + """The generated repository's owner and the repo allow-list (403).""" + owner = body["owner"].lower() + if owner != (login or "").lower() and owner not in owners: + raise ForgeProxyError( + 403, "repository owner not allowed through the forge proxy" + ) + target = Route("repo-generate", body["owner"], body["name"], "", "") + check_repo(target, rules) + + +def check_collaborator(route: Route, login: str, accounts: set[str]) -> None: + """Only listed accounts, only on a repository the signed-in user owns.""" + if route.owner.lower() != (login or "").lower(): + raise ForgeProxyError( + 403, "collaborators only on your own repositories" + ) + if route.user.lower() not in accounts: + raise ForgeProxyError( + 403, "collaborator not allowed through the forge proxy" + ) + + +# --- push inspection --------------------------------------------------------- + + +def push_commands_end(buf: bytes) -> int | None: + """Offset just past the flush-pkt ending a push's commands, or None. + + None means more bytes are needed. A malformed length is a 400. + """ + pos = 0 + while pos + 4 <= len(buf): + size = _pkt_len(buf[pos : pos + 4]) + if size == 0: + return pos + 4 + pos += size + return None + + +def _pkt_len(head: bytes) -> int: + try: + size = int(head, 16) + except ValueError: + raise ForgeProxyError(400, "malformed push request") + if size != 0 and size < 4: + raise ForgeProxyError(400, "malformed push request") + return size + + +def check_push_commands(buf: bytes, end: int) -> None: + """Refuse a push whose ref updates delete a ref (403).""" + pos = 0 + while pos < end - 4: + size = _pkt_len(buf[pos : pos + 4]) + _check_command(buf[pos + 4 : pos + size]) + pos += size + + +def _check_command(payload: bytes) -> None: + line = payload.split(b"\0", 1)[0].decode("latin-1").rstrip("\n") + if _SHALLOW.match(line): + return + m = _PUSH_COMMAND.match(line) + if m is None: + raise ForgeProxyError(400, "push command not allowed") + if set(m.group("new")) == {"0"}: + raise ForgeProxyError( + 403, "deleting a ref is not allowed through the forge proxy" + ) + + +@dataclass +class ForgeProxy: + """Per-process token custody and policy for proxied forges.""" + + app: object + _txns: dict = field(default_factory=dict) + _tokens: dict = field(default_factory=dict) + _locks: dict = field(default_factory=dict) + + # --- configuration --------------------------------------------------- + + def _settings(self): + return self.app.state.settings + + def proxied_hosts(self) -> set[str]: + raw = self._settings().forge_proxy_hosts or "" + return {_norm_host(h) for h in raw.split(",") if h.strip()} + + def provider(self, host: str) -> dict: + """The authorization_code_pkce provider entry for a proxied host.""" + host = _norm_host(host) + if host not in self.proxied_hosts(): + raise ForgeProxyError(404, "forge is not proxied") + for entry in self._provider_entries(): + if _provider_matches(entry, host): + return entry + raise ForgeProxyError(404, "forge provider is not configured") + + def _provider_entries(self) -> list[dict]: + raw = settings_mod.resolve_dynamic_config( + PROVIDERS_KEY, features_config=self._settings().features_config + ) + try: + entries = json.loads(raw) + except TypeError, ValueError: + return [] + if not isinstance(entries, list): + return [] + return [e for e in entries if isinstance(e, dict)] + + def origin(self, host: str) -> str: + return f"https://{_norm_host(host)}" + + def repo_rules(self) -> list[RepoRule]: + return parse_repo_rules(self._settings().forge_proxy_allowed_repos) + + def features(self) -> set[str]: + return parse_features(self._settings().forge_proxy_features) + + def passthrough(self) -> bool: + return "api-passthrough" in self.features() + + def check_route(self, route: Route) -> None: + """Refuse a route whose feature is off or that is outside + ``forge_proxy_allowed_repos`` (a template generate is checked + against ``forge_proxy_template_repos``).""" + check_feature(route, self.features()) + if route.kind == "repo-generate": + check_template(route, self.template_repos()) + return + check_repo(route, self.repo_rules()) + + def template_repos(self) -> set[str]: + return parse_names(self._settings().forge_proxy_template_repos) + + def check_actor(self, route: Route, body, login: str) -> None: + """Checks that need the signed-in account (known once a token is).""" + settings = self._settings() + if route.kind == "repo-generate": + owners = parse_names(settings.forge_proxy_template_owners) + check_new_repo(json.loads(body), login, owners, self.repo_rules()) + elif route.kind == "collaborator-add": + accounts = parse_names(settings.forge_proxy_collaborator_accounts) + check_collaborator(route, login, accounts) + + # --- authorization transactions -------------------------------------- + + def _prune(self, now: float) -> None: + for txn_id in [ + t + for t, txn in self._txns.items() + if now - txn.created > TXN_TTL_SECONDS + ]: + del self._txns[txn_id] + + def start(self, workspace_id: str, host: str) -> str: + """Create a transaction; returns only its opaque id.""" + self.provider(host) + now = time.time() + self._prune(now) + mine = sum( + 1 for x in self._txns.values() if x.workspace_id == workspace_id + ) + if ( + mine >= MAX_PENDING_TXNS_PER_WORKSPACE + or len(self._txns) >= MAX_PENDING_TXNS + ): + raise ForgeProxyError(429, "too many pending authorizations") + txn = Txn( + txn_id=secrets.token_urlsafe(32), + workspace_id=workspace_id, + host=_norm_host(host), + verifier=secrets.token_urlsafe(64), + state=secrets.token_urlsafe(32), + created=now, + ) + self._txns[txn.txn_id] = txn + return txn.txn_id + + def claim(self, workspace_id: str, txn_id: str) -> Txn: + """Take a pending transaction for its workspace, at most once.""" + self._prune(time.time()) + txn = self._txns.get(txn_id or "") + if txn is None or txn.workspace_id != workspace_id or txn.in_flight: + raise ForgeProxyError(403, "unknown authorization transaction") + txn.in_flight = True + return txn + + def authorize_url(self, txn: Txn) -> str: + entry = self.provider(txn.host) + params = { + "response_type": "code", + "client_id": entry["client_id"], + "redirect_uri": entry["redirect_uri"], + "code_challenge": s256_challenge(txn.verifier), + "code_challenge_method": "S256", + "state": txn.state, + } + if entry.get("scope"): + params["scope"] = entry["scope"] + url = entry["authorize_url"] + sep = "&" if "?" in url else "?" + return f"{url}{sep}{urllib.parse.urlencode(params)}" + + async def complete(self, txn: Txn, code: str, state: str, http) -> str: + """Exchange the code (verifier from the txn); returns the login.""" + try: + if not code or not secrets.compare_digest(state or "", txn.state): + raise ForgeProxyError(403, "authorization state mismatch") + if time.time() - txn.created > TXN_TTL_SECONDS: + raise ForgeProxyError( + 403, "authorization expired; start again" + ) + entry = self.provider(txn.host) + resp = await _call( + http.post, + entry["token_url"], + data={ + "grant_type": "authorization_code", + "code": code, + "code_verifier": txn.verifier, + "client_id": entry["client_id"], + "redirect_uri": entry["redirect_uri"], + }, + headers={"Accept": "application/json"}, + follow_redirects=False, + ) + tokens = _token_json(resp) + login = await self._login(txn.host, tokens["access_token"], http) + self._tokens[(txn.workspace_id, txn.host)] = ForgeToken( + access_token=tokens["access_token"], + refresh_token=tokens.get("refresh_token", ""), + expires_at=_expiry(tokens), + login=login, + ) + logger.info( + "forge-proxy: workspace %s authorized %s as %s (scope: %s)", + txn.workspace_id, + txn.host, + login, + granted_scope(tokens), + ) + return login + finally: + self._txns.pop(txn.txn_id, None) + + async def _login(self, host: str, access_token: str, http) -> str: + resp = await _call( + http.get, + f"{self.origin(host)}/api/v1/user", + headers={ + "Authorization": f"Bearer {access_token}", + "Accept": "application/json", + }, + follow_redirects=False, + ) + login = _login_of(resp) + if not login: + raise ForgeProxyError(502, "forge did not confirm the account") + return login + + # --- tokens ---------------------------------------------------------- + + def status(self, workspace_id: str, host: str) -> dict: + self.provider(host) + tok = self._tokens.get((workspace_id, _norm_host(host))) + return { + "connected": tok is not None, + "login": tok.login if tok else None, + } + + async def access_token( + self, workspace_id: str, host: str, http + ) -> ForgeToken: + """The workspace's current token, refreshed host-side (serialized).""" + key = (workspace_id, _norm_host(host)) + tok = self._tokens.get(key) + if tok is None: + raise ForgeProxyError(401, _needs_auth(host)) + if not _expiring(tok): + return tok + async with self._locks.setdefault(key, asyncio.Lock()): + if self._tokens.get(key) is tok: + return await self._refresh(key, tok, http) + # Another request refreshed (or a new authorization replaced + # it) while this one waited. + return self._stored(key) + + def _stored(self, key) -> ForgeToken: + current = self._tokens.get(key) + if current is None: + raise ForgeProxyError(401, _needs_auth(key[1])) + return current + + def forget_token( + self, workspace_id: str, host: str, tok: ForgeToken + ) -> None: + """Drop the token only if it is still the stored one.""" + key = (workspace_id, _norm_host(host)) + if self._tokens.get(key) is tok: + del self._tokens[key] + + async def _refresh(self, key, tok: ForgeToken, http) -> ForgeToken: + if not tok.refresh_token: + self.forget_token(key[0], key[1], tok) + raise ForgeProxyError(401, _needs_auth(key[1])) + tokens = await self._refresh_grant(key, tok, http) + fresh = ForgeToken( + access_token=tokens["access_token"], + refresh_token=tokens.get("refresh_token") or tok.refresh_token, + expires_at=_expiry(tokens), + login=tok.login, + ) + if self._tokens.get(key) is tok: + self._tokens[key] = fresh + return fresh + # Replaced or forgotten while refreshing: serve what is stored now. + return self._stored(key) + + async def _refresh_grant(self, key, tok: ForgeToken, http) -> dict: + entry = self.provider(key[1]) + try: + resp = await _call( + http.post, + entry["token_url"], + data={ + "grant_type": "refresh_token", + "refresh_token": tok.refresh_token, + "client_id": entry["client_id"], + }, + headers={"Accept": "application/json"}, + follow_redirects=False, + ) + return _token_json(resp) + except ForgeProxyError: + self.forget_token(key[0], key[1], tok) + raise ForgeProxyError(401, _needs_auth(key[1])) + + # --- browser-delegate relay policy ----------------------------------- + + def bridge_policy(self, payload: dict) -> str: + """How the relay treats a git_credential bridge request. + + ``"txn"``: a klangkd-issued authorization (handled by + :meth:`relay_authorization`); ``"refuse"``: a request that could + put a proxied forge's credential in the container (a + container-built authorize URL, a PAT prompt, a cache read); + ``"pass"``: unrelated to proxied forges. + """ + if payload.get("action") != "git_credential": + return "pass" + if ( + payload.get("txn_id") + and payload.get("operation") == "auth_flow_start" + ): + return "txn" + return "refuse" if self._may_leak(payload) else "pass" + + def _may_leak(self, payload: dict) -> bool: + hosts = self.proxied_hosts() + if not hosts: + return False + # Any container-built authorize URL is refused while a forge is + # proxied: hostname parsing that differs from the browser's + # (percent-encoding, IDNA) or a redirect chain could otherwise land + # on the proxied forge's authorize page. + if ( + payload.get("operation") == "auth_flow_start" + or _norm_host(str(payload.get("host", ""))) in hosts + ): + return True + return any( + _leaky_uri(payload.get(k), hosts) + for k in ("verification_uri", "verification_uri_complete") + ) + + async def relay_authorization(self, workspace_id, payload, dispatch, http): + """Run a klangkd-issued authorization through the browser tab. + + The tab gets the klangkd-built authorize URL; its answer (the + code) is consumed here and never relayed to the container. + """ + txn = self.claim(workspace_id, str(payload.get("txn_id", ""))) + try: + request = { + "action": "git_credential", + "operation": "auth_flow_start", + "protocol": "https", + "host": txn.host, + "authorize_url": self.authorize_url(txn), + "state": txn.state, + } + except ForgeProxyError: + self._txns.pop(txn.txn_id, None) + raise + try: + result = await dispatch(request, 900.0) + except BaseException: + self._txns.pop(txn.txn_id, None) + raise + answer = _unwrap(result) + if answer.get("error"): + self._txns.pop(txn.txn_id, None) + cancelled = answer["error"] == "cancelled" + return _wrap( + {"error": "cancelled" if cancelled else "authorization failed"} + ) + login = await self.complete( + txn, + str(answer.get("code", "")), + str(answer.get("state", "")), + http, + ) + return _wrap({"status": "connected", "host": txn.host, "login": login}) + + +def _provider_matches(entry: dict, host: str) -> bool: + return ( + _norm_host(str(entry.get("host", ""))) == host + and entry.get("flow") == "authorization_code_pkce" + and bool(entry.get("client_id")) + and _provider_urls_ok(entry) + ) + + +def _provider_urls_ok(entry: dict) -> bool: + https_only = all( + str(entry.get(k, "")).startswith("https://") + for k in ("authorize_url", "token_url") + ) + # A local dev deployment registers http://localhost:/. + redirect_ok = str(entry.get("redirect_uri", "")).startswith( + ("https://", "http://localhost:", "http://127.0.0.1:") + ) + return https_only and redirect_ok + + +def _leaky_uri(value, hosts: set[str]) -> bool: + """A verification URI that is, or might be read as, a proxied forge.""" + uri = str(value or "") + return bool(uri) and (_host_of_url(uri) in hosts or _ambiguous_netloc(uri)) + + +def granted_scope(tokens: dict) -> str: + """The scope a token response reports (logged; never the token).""" + return str(tokens.get("scope") or "not reported") + + +def _login_of(resp) -> str: + data = _json(resp) if resp.status_code == 200 and _is_json(resp) else None + login = data.get("login") if isinstance(data, dict) else None + return login if isinstance(login, str) else "" + + +def _expiring(tok: ForgeToken) -> bool: + return bool(tok.expires_at) and ( + tok.expires_at - time.time() <= REFRESH_SKEW_SECONDS + ) + + +def _needs_auth(host: str) -> str: + return ( + f"forge not authorized for this workspace; run " + f"`git-credential-klangk forge-auth {_norm_host(host)}`" + ) + + +async def _call(method, *args, **kwargs): + """An HTTP call to the forge; transport errors become a 502.""" + try: + return await method(*args, **kwargs) + except Exception as exc: # httpx.HTTPError and friends + logger.warning( + "forge-proxy: forge call failed: %s", type(exc).__name__ + ) + raise ForgeProxyError(502, "forge unreachable") + + +def _json(resp): + try: + return resp.json() + except ValueError: + return None + + +def _is_json(resp) -> bool: + return "json" in (resp.headers.get("content-type") or "") + + +def _token_json(resp) -> dict: + if resp.status_code != 200 or not _is_json(resp): + raise ForgeProxyError(502, "token endpoint refused the request") + data = _json(resp) + if not isinstance(data, dict) or not isinstance( + data.get("access_token"), str + ): + raise ForgeProxyError(502, "token endpoint refused the request") + return data + + +def _expiry(tokens: dict) -> float: + expires_in = tokens.get("expires_in") + if isinstance(expires_in, (int, float)) and expires_in > 0: + return time.time() + float(expires_in) + return 0.0 + + +def _unwrap(result) -> dict: + """The tab's answer inside the relay envelope {"status","result"}.""" + if not isinstance(result, dict): + return {"error": "no answer from the browser"} + if result.get("error"): + return {"error": result["error"]} + inner = _decode(result.get("result")) + return inner if isinstance(inner, dict) else {"error": "no answer"} + + +def _decode(inner): + if not isinstance(inner, str): + return inner + try: + return json.loads(inner) + except ValueError: + return {"error": "unreadable answer from the browser"} + + +def _wrap(answer: dict) -> dict: + return {"status": "ok", "result": json.dumps(answer)} diff --git a/src/klangk/klangk/main.py b/src/klangk/klangk/main.py index 355dc7eef..059ccc7c3 100644 --- a/src/klangk/klangk/main.py +++ b/src/klangk/klangk/main.py @@ -88,6 +88,7 @@ inactivity, model, caddy as caddy_mod, + forge_proxy as forge_proxy_mod, oidc, features, nix, @@ -271,6 +272,8 @@ def build_app(settings: KlangkSettings) -> FastAPI: app.state.proxy_watchdog = caddy_mod.CaddyWatchdog(app) # #2070: In-process LLM router backed by litellm.Router (subsystem). app.state.llm_router = LLMRouter(app) + # Host-side custody of git-forge OAuth tokens (forge_proxy.py). + app.state.forge_proxy = forge_proxy_mod.ForgeProxy(app) # #1480: Terminal(app_state) groups the ~25 tmux-session # management functions that share a Podman dependency. Reaches podman, # the registry, and settings through the single app_state reference. diff --git a/src/klangk/klangk/settings.py b/src/klangk/klangk/settings.py index 05ff5294d..c60f5a938 100644 --- a/src/klangk/klangk/settings.py +++ b/src/klangk/klangk/settings.py @@ -1461,6 +1461,49 @@ def settings_customise_sources( # applies to requests and registrations after it; already-attached # container envs keep the stale ID but the endpoint 403s regardless. browser_delegate_enabled: bool = True + # forge_proxy_hosts: comma-separated git-forge hosts whose OAuth + # tokens klangkd keeps host-side (forge_proxy.py). For these hosts the + # browser-delegate relay refuses container-built authorization flows, + # PAT prompts and credential-cache reads, and containers reach the + # forge through the allow-listed ``/forge-proxy//`` egress route + # with their workspace JWT. Empty (default) disables the proxy. + forge_proxy_hosts: str = "" + # forge_proxy_allowed_repos: comma-separated repositories workspaces + # reach through the forge proxy: ``owner/repo``, ``owner/*``, or ``*`` + # for every repository. A ``:ro`` suffix allows reads only; ``:issues`` + # allows reads plus issue work (opening, commenting, editing, closing, + # assigning and labelling, as forge_proxy_features enables) and refuses + # pushes and other writes. A repository no rule matches is refused with + # 403, so an empty value (the default) allows none. + forge_proxy_allowed_repos: str = "" + # forge_proxy_features: comma-separated operations the forge proxy + # performs; every other operation is refused with 403, so an empty + # value (the default) allows none. ``read`` (account, repositories, + # issues, comments, labels), ``git-read`` (clone and fetch), + # ``git-push``, ``issues`` (open issues and comments), ``issue-edit`` + # (edit, close, assign and label), ``site-create`` (generate from a + # template in forge_proxy_template_repos), ``collaborators``, + # ``api-passthrough`` (any other /api/v1/ call; the forge's own + # permissions and the token's scopes decide), or ``*`` for all. + # forge_proxy_allowed_repos applies to every repository path. + forge_proxy_features: str = "" + # forge_proxy_template_repos: comma-separated ``owner/repo`` template + # repositories a workspace may generate a new repository from through + # the forge proxy. The new repository is created in the signed-in + # forge account (or an owner in forge_proxy_template_owners) and must + # also pass forge_proxy_allowed_repos. Empty (default) allows none. + forge_proxy_template_repos: str = "" + # forge_proxy_template_owners: comma-separated organizations a + # workspace may generate a repository into, besides its own account. + forge_proxy_template_owners: str = "" + # forge_proxy_collaborator_accounts: comma-separated forge accounts a + # workspace may add as read-only collaborators to a repository the + # signed-in account owns (for example a build service account). + # Empty (default) allows none. + forge_proxy_collaborator_accounts: str = "" + # forge_proxy_ca_cert: optional PEM file added to the default trust + # store for klangkd's calls to proxied forges (a private CA). + forge_proxy_ca_cert: str | None = None health_check_interval: float | None = None health_check_startup_grace: float | None = None health_check_timeout: float | None = None diff --git a/src/klangk/klangkd-tests/tests/test_api.py b/src/klangk/klangkd-tests/tests/test_api.py index d91abd18d..64c4dd443 100644 --- a/src/klangk/klangkd-tests/tests/test_api.py +++ b/src/klangk/klangkd-tests/tests/test_api.py @@ -18868,3 +18868,152 @@ async def test_bound_refresh_via_endpoint(self, client, user): resp.json()["access_token"], _SECRET, algorithms=[_ALGORITHM] ) assert payload["cnf"]["jkt"] == dpop_mod.jwk_thumbprint(jwk) + + +# --- relay integration (real app fixtures) ----------------------------------- + +import json # noqa: E402 + +import test_forge_proxy as _forge_tests # noqa: E402 + +HOST = _forge_tests.HOST +providers = _forge_tests.providers +upstream = _forge_tests.upstream +from klangk import forge_proxy as fp # noqa: E402 + + +@pytest.mark.asyncio +class TestBridgeRelay: + """The browser-delegate relay enforces the proxied-forge policy.""" + + def _setup(self, app, registry, monkeypatch, workspace_id, answer): + monkeypatch.setattr(app.state.settings, "forge_proxy_hosts", HOST) + app.state.forge_proxy = fp.ForgeProxy(app) + sock = MagicMock() + registry.register_browser(f"bid-{workspace_id}", workspace_id, sock) + session = AsyncMock() + session.browser_subscribers = {sock} + session.dispatch_browser_request_to = AsyncMock(return_value=answer) + session.dispatch_browser_request_stream_to = MagicMock() + return session + + def _headers(self, app, workspace_id): + token = app.state.auth.create_workspace_token(workspace_id) + return {"Authorization": f"Bearer {token}"} + + async def test_refuses_pat_prompt_for_proxied_forge( + self, client, app, user, registry, sockets, providers, monkeypatch + ): + session = self._setup(app, registry, monkeypatch, "ws-r1", {}) + try: + with patch.object(sockets, "get_session", return_value=session): + resp = await client.post( + "/api/v1/browser-delegate", + json={ + "action": "git_credential", + "operation": "get", + "host": HOST, + "browser_id": "bid-ws-r1", + }, + headers=self._headers(app, "ws-r1"), + ) + assert resp.status_code == 403 + session.dispatch_browser_request_to.assert_not_awaited() + finally: + registry.revoke_workspace_browsers("ws-r1") + del app.state.forge_proxy + + async def test_txn_code_consumed_by_klangkd( + self, + client, + app, + user, + registry, + sockets, + providers, + monkeypatch, + upstream, + ): + upstream.on( + "POST", "/login/oauth/access_token", body={"access_token": "at"} + ) + upstream.on("GET", "/api/v1/user", body={"login": "me"}) + session = self._setup(app, registry, monkeypatch, "ws-r2", None) + txn_id = app.state.forge_proxy.start("ws-r2", HOST) + state = app.state.forge_proxy._txns[txn_id].state + session.dispatch_browser_request_to.return_value = { + "status": "ok", + "result": json.dumps({"code": "c0de", "state": state}), + } + try: + with patch.object(sockets, "get_session", return_value=session): + resp = await client.post( + "/api/v1/browser-delegate", + json={ + "action": "git_credential", + "operation": "auth_flow_start", + "host": HOST, + "txn_id": txn_id, + "browser_id": "bid-ws-r2", + }, + headers=self._headers(app, "ws-r2"), + ) + assert resp.status_code == 200 + assert "c0de" not in resp.text + assert json.loads(resp.json()["result"])["status"] == "connected" + assert app.state.forge_proxy.status("ws-r2", HOST)["connected"] + finally: + registry.revoke_workspace_browsers("ws-r2") + del app.state.forge_proxy + + async def test_txn_unknown_is_403( + self, + client, + app, + user, + registry, + sockets, + providers, + monkeypatch, + upstream, + ): + session = self._setup(app, registry, monkeypatch, "ws-r3", {}) + try: + with patch.object(sockets, "get_session", return_value=session): + resp = await client.post( + "/api/v1/browser-delegate", + json={ + "action": "git_credential", + "operation": "auth_flow_start", + "txn_id": "nope", + "browser_id": "bid-ws-r3", + }, + headers=self._headers(app, "ws-r3"), + ) + assert resp.status_code == 403 + finally: + registry.revoke_workspace_browsers("ws-r3") + del app.state.forge_proxy + + async def test_stream_refuses_proxied_forge( + self, client, app, user, registry, sockets, providers, monkeypatch + ): + session = self._setup(app, registry, monkeypatch, "ws-r4", {}) + try: + with patch.object(sockets, "get_session", return_value=session): + resp = await client.post( + "/api/v1/browser-delegate/stream", + json={ + "action": "git_credential", + "operation": "auth_flow_start", + "host": HOST, + "authorize_url": f"https://{HOST}/x", + "browser_id": "bid-ws-r4", + }, + headers=self._headers(app, "ws-r4"), + ) + assert resp.status_code == 403 + session.dispatch_browser_request_stream_to.assert_not_called() + finally: + registry.revoke_workspace_browsers("ws-r4") + del app.state.forge_proxy diff --git a/src/klangk/klangkd-tests/tests/test_api_package.py b/src/klangk/klangkd-tests/tests/test_api_package.py index d46a55b9d..43faf7250 100644 --- a/src/klangk/klangkd-tests/tests/test_api_package.py +++ b/src/klangk/klangkd-tests/tests/test_api_package.py @@ -37,12 +37,13 @@ # to preserve the monolith's 90 exactly; the expired-password route # (#3177), /audit (#3154), the #3214 admin audit-events route, step-up # (#3196), /auth/bind (#3218), the OIDC code exchange (#3201), and the -# merged events stream (#3251) are the additions since. -EXPECTED_ROUTE_COUNT = 97 +# merged events stream (#3251) are the additions since, plus the 5 forge-proxy +# routes (oauth start, status, GET/POST/PATCH forward). +EXPECTED_ROUTE_COUNT = 104 # Per-domain submodules and the number of routes each owns. 91 sub-routes # + 3 routes defined directly on the main router (version, config, -# my-permissions) + 3 on the root router (health, audit, empty) == 97. +# my-permissions) + 3 on the root router (health, audit, empty) == 102. SUBMODULE_ROUTES = { "auth": 21, # 15 + 2 OIDC login/callback + expired-password (#3177) + step-up (#3196) + bind (#3218) + exchange (#3201) "workspaces": 27, @@ -50,6 +51,7 @@ "browser_delegate": 2, "admin": 29, # + the merged events stream (#3251) "llm_proxy": 2, + "forge_proxy": 7, # oauth start + status + GET/POST/PUT/PATCH/DELETE forward } # One representative path from every domain (and the cross-cutting diff --git a/src/klangk/klangkd-tests/tests/test_forge_proxy.py b/src/klangk/klangkd-tests/tests/test_forge_proxy.py new file mode 100644 index 000000000..f2d0aa3c2 --- /dev/null +++ b/src/klangk/klangkd-tests/tests/test_forge_proxy.py @@ -0,0 +1,1943 @@ +"""Forge proxy: host-held forge tokens, allow-listed git/API proxy, bridge policy.""" + +import json +import time +import types + +import httpx +import pytest +from fastapi import FastAPI +from httpx import ASGITransport, AsyncClient + +from klangk import forge_proxy as fp +from klangk.api import forge_proxy as routes +from klangk.auth import Auth +from _helpers import make_settings + +HOST = "forge.test" +PROVIDER = { + "host": HOST, + "flow": "authorization_code_pkce", + "client_id": "cid", + "authorize_url": f"https://{HOST}/login/oauth/authorize", + "token_url": f"https://{HOST}/login/oauth/access_token", + "redirect_uri": "http://localhost:8997/", +} + + +@pytest.fixture +def providers(monkeypatch): + def set_(entries): + monkeypatch.setenv(fp.PROVIDERS_KEY, json.dumps(entries)) + + set_([PROVIDER]) + return set_ + + +def _app(env=None): + settings = make_settings( + env={ + "KLANGKD_FORGE_PROXY_HOSTS": HOST, + "KLANGKD_FORGE_PROXY_ALLOWED_REPOS": "*", + "KLANGKD_FORGE_PROXY_FEATURES": "read,git-read,git-push,issues,issue-edit,site-create,collaborators", + **(env or {}), + } + ) + state = types.SimpleNamespace( + state=types.SimpleNamespace(settings=settings) + ) + app = FastAPI() + app.state.settings = settings + app.state.auth = Auth(state) + app.state.forge_proxy = fp.ForgeProxy(app) + app.include_router(routes.router) + return app + + +def _ws(app, workspace_id="ws1"): + token = app.state.auth.create_workspace_token(workspace_id) + return {"Authorization": f"Bearer {token}"} + + +def _client(app): + return AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) + + +class Upstream: + """A fake forge: records requests, answers by (method, path).""" + + def __init__(self): + self.requests = [] + self.answers = {} + + def on(self, method, path, status=200, body=None, headers=None): + self.answers[(method, path)] = (status, body, headers or {}) + + def handler(self, request): + self.requests.append(request) + status, body, headers = self.answers.get( + (request.method, request.url.path), (404, {"message": "nope"}, {}) + ) + if isinstance(body, (dict, list)): + data = json.dumps(body).encode() + headers = {"Content-Type": "application/json", **headers} + else: + data = body or b"" + # A real byte stream, as a server sends: the proxy streams it raw. + return httpx.Response( + status, headers=headers, stream=httpx.ByteStream(data) + ) + + def client(self): + return httpx.AsyncClient(transport=httpx.MockTransport(self.handler)) + + +@pytest.fixture +def upstream(monkeypatch): + up = Upstream() + monkeypatch.setattr(routes, "http_client_for", lambda app: up.client()) + return up + + +def _authorized(app, workspace_id="ws1", expires_at=0.0, refresh="r1"): + app.state.forge_proxy._tokens[(workspace_id, HOST)] = fp.ForgeToken( + access_token="forge-at", + refresh_token=refresh, + expires_at=expires_at, + login="me", + ) + + +# --- pure policy ------------------------------------------------------------- + + +class TestPaths: + @pytest.mark.parametrize( + "path", + [ + "/a/%2e%2e/b", + "/a\\b", + "//a", + "/a/./b", + "/a/../b", + "/a/\x01", + "x/y", + "/a/\x7f", + ], + ) + def test_ambiguous_paths_rejected(self, path): + with pytest.raises(fp.ForgeProxyError) as e: + fp.validate_raw_path(path) + assert e.value.status == 400 + + def test_clean_path_ok(self): + assert ( + fp.validate_raw_path("/o/r.git/info/refs") == "/o/r.git/info/refs" + ) + + +class TestRoutes: + def test_git_refs(self): + r = fp.match_route( + "GET", "/o/r.git/info/refs", "service=git-upload-pack" + ) + assert (r.kind, r.owner, r.repo, r.query) == ( + "git-refs", + "o", + "r", + "service=git-upload-pack", + ) + + def test_git_refs_bad_service(self): + with pytest.raises(fp.ForgeProxyError) as e: + fp.match_route("GET", "/o/r.git/info/refs", "service=other") + assert e.value.status == 400 + + def test_receive_pack(self): + assert ( + fp.match_route("POST", "/o/r.git/git-receive-pack", "").kind + == "git-receive" + ) + + def test_issue_routes(self): + assert ( + fp.match_route("POST", "/api/v1/repos/o/r/issues", "").kind + == "issue-create" + ) + assert ( + fp.match_route( + "POST", "/api/v1/repos/o/r/issues/3/comments", "" + ).kind + == "comment-create" + ) + assert ( + fp.match_route("PATCH", "/api/v1/repos/o/r/issues/3", "").kind + == "issue-state" + ) + assert fp.match_route("GET", "/api/v1/user", "").kind == "api-read" + + def test_allowed_query_rebuilt(self): + r = fp.match_route( + "GET", "/api/v1/repos/o/r/issues", "state=open&limit=2" + ) + assert r.query == "state=open&limit=2" + + @pytest.mark.parametrize( + "query", ["access_token=x", "sudo=admin", "state=a&state=b"] + ) + def test_bad_query(self, query): + with pytest.raises(fp.ForgeProxyError): + fp.match_route("GET", "/api/v1/repos/o/r/issues", query) + + @pytest.mark.parametrize( + "method,path", + [ + ("GET", "/api/v1/admin/users"), + ("DELETE", "/api/v1/repos/o/r"), + ("GET", "/api/v1/user/applications/oauth2"), + ("PATCH", "/api/v1/repos/o/r"), + ("POST", "/api/v1/repos/o/r/issues/0/comments"), + ], + ) + def test_not_allowed(self, method, path): + with pytest.raises(fp.ForgeProxyError) as e: + fp.match_route(method, path, "") + assert e.value.status == 403 + + def test_empty_repo_name(self): + with pytest.raises(fp.ForgeProxyError) as e: + fp.match_route( + "GET", "/o/.git/info/refs", "service=git-upload-pack" + ) + assert e.value.status == 403 + + +class TestBodies: + def test_issue_create(self): + out = fp.validate_json_body( + "issue-create", b'{"title":"t","body":"b"}' + ) + assert json.loads(out) == {"title": "t", "body": "b"} + + def test_git_body_untouched(self): + assert fp.validate_json_body("git-upload", b"\x00pack") == b"\x00pack" + + @pytest.mark.parametrize( + "kind,body,status", + [ + ("issue-create", b'{"title":"t","assignees":"x"}', 400), + ("issue-create", b'{"title":"t","ref":"main"}', 400), + ("issue-labels", b'{"labels":[0]}', 400), + ("issue-labels", b'{"labels":[true]}', 400), + ("issue-labels", b'{"labels":[""]}', 400), + ("issue-create", b'{"body":"b"}', 400), + ("issue-create", b"not json", 400), + ("issue-create", b"[1]", 400), + ("issue-create", b'{"title":"a","title":"b"}', 400), + ("issue-create", b'{"title":1}', 400), + ("issue-state", b'{"state":"locked"}', 400), + ( + "comment-create", + b'{"body":"' + b"x" * fp.MAX_JSON_BODY + b'"}', + 413, + ), + ], + ) + def test_rejected(self, kind, body, status): + with pytest.raises(fp.ForgeProxyError) as e: + fp.validate_json_body(kind, body) + assert e.value.status == status + + +class TestProviderConfig: + def test_not_proxied(self, providers): + with pytest.raises(fp.ForgeProxyError) as e: + _app().state.forge_proxy.provider("other.test") + assert e.value.status == 404 + + @pytest.mark.parametrize( + "entries", + [ + "not json", + {"host": HOST}, + ["x", {"host": "else.test"}], + [dict(PROVIDER, flow="device_code")], + [dict(PROVIDER, token_url="http://forge.test/t")], + [dict(PROVIDER, redirect_uri="http://evil/")], + [dict(PROVIDER, client_id="")], + ], + ) + def test_unusable_entries(self, monkeypatch, entries): + monkeypatch.setenv( + fp.PROVIDERS_KEY, + entries if isinstance(entries, str) else json.dumps(entries), + ) + with pytest.raises(fp.ForgeProxyError): + _app().state.forge_proxy.provider(HOST) + + def test_found_with_port_and_case(self, providers): + assert ( + _app().state.forge_proxy.provider("FORGE.test:443")["client_id"] + == "cid" + ) + + def test_empty_hosts(self, providers): + app = _app({"KLANGKD_FORGE_PROXY_HOSTS": ""}) + assert app.state.forge_proxy.proxied_hosts() == set() + + def test_host_of_url(self): + assert fp._host_of_url("https://user@FORGE.TEST.:443/x") == HOST + assert fp._host_of_url("https://other.test/") == "other.test" + assert fp._host_of_url("http://[::1") == "" + + def test_providers_unset_or_not_a_list(self, monkeypatch): + monkeypatch.delenv(fp.PROVIDERS_KEY, raising=False) + f = _app().state.forge_proxy + assert f._provider_entries() == [] + monkeypatch.setenv(fp.PROVIDERS_KEY, "{}") + assert f._provider_entries() == [] + + def test_no_proxied_hosts_passes_everything(self, providers): + f = _app({"KLANGKD_FORGE_PROXY_HOSTS": ""}).state.forge_proxy + assert ( + f.bridge_policy( + { + "action": "git_credential", + "operation": "get", + "host": "x", + "verification_uri": "https://%66orge.test/", + } + ) + == "pass" + ) + + +class TestTransactions: + def test_start_claim_once(self, providers): + f = _app().state.forge_proxy + txn_id = f.start("ws1", HOST) + with pytest.raises(fp.ForgeProxyError): + f.claim("ws2", txn_id) + txn = f.claim("ws1", txn_id) + url = f.authorize_url(txn) + assert "code_challenge_method=S256" in url and txn.verifier not in url + with pytest.raises(fp.ForgeProxyError): + f.claim("ws1", txn_id) + + def test_expired_txn_pruned(self, providers, monkeypatch): + f = _app().state.forge_proxy + txn_id = f.start("ws1", HOST) + later = time.time() + fp.TXN_TTL_SECONDS + 1 + monkeypatch.setattr(fp.time, "time", lambda: later) + with pytest.raises(fp.ForgeProxyError): + f.claim("ws1", txn_id) + + def test_scope_and_query_in_authorize_url(self, providers): + providers( + [ + dict( + PROVIDER, + scope="read:user", + authorize_url=PROVIDER["authorize_url"] + "?x=1", + ) + ] + ) + f = _app().state.forge_proxy + url = f.authorize_url(f.claim("ws1", f.start("ws1", HOST))) + assert "scope=read%3Auser" in url and "?x=1&" in url + + +@pytest.mark.asyncio +class TestCompleteAndRefresh: + async def test_complete_stores_token_host_side(self, providers): + up = Upstream() + up.on( + "POST", + "/login/oauth/access_token", + body={ + "access_token": "at", + "refresh_token": "rt", + "expires_in": 3600, + }, + ) + up.on("GET", "/api/v1/user", body={"login": "me"}) + f = _app().state.forge_proxy + txn = f.claim("ws1", f.start("ws1", HOST)) + async with up.client() as http: + assert await f.complete(txn, "code", txn.state, http) == "me" + sent = up.requests[0].content.decode() + assert "code_verifier=" + txn.verifier in sent + assert f.status("ws1", HOST) == {"connected": True, "login": "me"} + assert txn.txn_id not in f._txns + + async def test_state_mismatch(self, providers): + f = _app().state.forge_proxy + txn = f.claim("ws1", f.start("ws1", HOST)) + with pytest.raises(fp.ForgeProxyError): + await f.complete(txn, "code", "wrong", None) + assert txn.txn_id not in f._txns + + @pytest.mark.parametrize( + "token_answer,user_answer", + [ + ((400, {"error": "bad"}), None), + ((200, b"html"), None), + ((200, {"no": "token"}), None), + ((200, {"access_token": "at"}), (403, {})), + ((200, {"access_token": "at"}), (200, b"html")), + ((200, {"access_token": "at"}), (200, {"login": ""})), + ], + ) + async def test_exchange_failures( + self, providers, token_answer, user_answer + ): + up = Upstream() + up.on("POST", "/login/oauth/access_token", *token_answer) + if user_answer: + up.on("GET", "/api/v1/user", *user_answer) + f = _app().state.forge_proxy + txn = f.claim("ws1", f.start("ws1", HOST)) + async with up.client() as http: + with pytest.raises(fp.ForgeProxyError) as e: + await f.complete(txn, "code", txn.state, http) + assert e.value.status == 502 + assert f.status("ws1", HOST)["connected"] is False + + async def test_access_token_missing(self, providers): + with pytest.raises(fp.ForgeProxyError) as e: + await _app().state.forge_proxy.access_token("ws1", HOST, None) + assert e.value.status == 401 + + async def test_refresh_rotates(self, providers): + app = _app() + _authorized(app, expires_at=time.time() + 10) + up = Upstream() + up.on( + "POST", + "/login/oauth/access_token", + body={ + "access_token": "at2", + "refresh_token": "r2", + "expires_in": 60, + }, + ) + async with up.client() as http: + assert ( + await app.state.forge_proxy.access_token("ws1", HOST, http) + ).access_token == "at2" + tok = app.state.forge_proxy._tokens[("ws1", HOST)] + assert tok.refresh_token == "r2" and tok.login == "me" + + async def test_refresh_keeps_old_refresh_token(self, providers): + app = _app() + _authorized(app, expires_at=time.time() + 10) + up = Upstream() + up.on( + "POST", "/login/oauth/access_token", body={"access_token": "at2"} + ) + async with up.client() as http: + await app.state.forge_proxy.access_token("ws1", HOST, http) + assert ( + app.state.forge_proxy._tokens[("ws1", HOST)].refresh_token == "r1" + ) + + async def test_refresh_rejected_forgets(self, providers): + app = _app() + _authorized(app, expires_at=time.time() + 10) + up = Upstream() + up.on( + "POST", + "/login/oauth/access_token", + 400, + {"error": "invalid_grant"}, + ) + async with up.client() as http: + with pytest.raises(fp.ForgeProxyError) as e: + await app.state.forge_proxy.access_token("ws1", HOST, http) + assert e.value.status == 401 + assert ("ws1", HOST) not in app.state.forge_proxy._tokens + + async def test_no_refresh_token_forgets(self, providers): + app = _app() + _authorized(app, expires_at=time.time() + 10, refresh="") + with pytest.raises(fp.ForgeProxyError): + await app.state.forge_proxy.access_token("ws1", HOST, None) + assert ("ws1", HOST) not in app.state.forge_proxy._tokens + + +class TestBridgePolicy: + def test_policy(self, providers): + f = _app().state.forge_proxy + auth_url = f"https://{HOST}/login/oauth/authorize?client_id=x" + cases = { + "pass": [ + {"action": "fetch"}, + { + "action": "git_credential", + "operation": "get", + "host": "github.com", + }, + { + "action": "git_credential", + "operation": "get", + "host": "github.com", + "verification_uri": "https://github.com/login/device", + }, + ], + "txn": [ + { + "action": "git_credential", + "operation": "auth_flow_start", + "txn_id": "t", + } + ], + "refuse": [ + {"action": "git_credential", "operation": "get", "host": HOST}, + { + "action": "git_credential", + "operation": "peek", + "host": "FORGE.TEST", + }, + { + "action": "git_credential", + "operation": "auth_flow_start", + "host": "github.com", + "authorize_url": auth_url, + }, + # Percent-encoded host: browsers decode it, urlsplit does not. + { + "action": "git_credential", + "operation": "auth_flow_start", + "host": "other.test", + "authorize_url": "https://%66orge.test/login/oauth/authorize", + }, + { + "action": "git_credential", + "operation": "auth_flow_start", + "host": "github.com", + "authorize_url": "https://github.com/x", + }, + { + "action": "git_credential", + "operation": "get", + "host": "x", + "verification_uri": f"https://{HOST}/device", + }, + { + "action": "git_credential", + "operation": "get", + "host": "x", + "verification_uri": "https://%66orge.test/verify", + }, + { + "action": "git_credential", + "operation": "get", + "host": "x", + "verification_uri_complete": "https://a@github.com/x", + }, + { + "action": "git_credential", + "operation": "get", + "host": "x", + "verification_uri": "https://fürge.test/x", + }, + { + "action": "git_credential", + "operation": "get", + "host": "x", + "verification_uri": "https://[::1/x", + }, + ], + } + for expected, payloads in cases.items(): + for payload in payloads: + assert f.bridge_policy(payload) == expected, payload + + +@pytest.mark.asyncio +class TestRelayAuthorization: + async def _relay(self, f, answer, txn_id=None): + txn_id = txn_id or f.start("ws1", HOST) + sent = [] + + async def dispatch(request, timeout): + sent.append(request) + return answer + + up = Upstream() + up.on("POST", "/login/oauth/access_token", body={"access_token": "at"}) + up.on("GET", "/api/v1/user", body={"login": "me"}) + async with up.client() as http: + out = await f.relay_authorization( + "ws1", + {"action": "git_credential", "txn_id": txn_id}, + dispatch, + http, + ) + return out, sent + + async def test_code_consumed_not_relayed(self, providers): + f = _app().state.forge_proxy + txn_id = f.start("ws1", HOST) + state = f._txns[txn_id].state + out, sent = await self._relay( + f, + { + "status": "ok", + "result": json.dumps({"code": "the-code", "state": state}), + }, + txn_id, + ) + assert "the-code" not in json.dumps(out) + assert json.loads(out["result"]) == { + "status": "connected", + "host": HOST, + "login": "me", + } + assert sent[0]["authorize_url"].startswith(PROVIDER["authorize_url"]) + + @pytest.mark.parametrize( + "answer", + [ + {"error": "cancelled"}, + {"status": "ok", "result": json.dumps({"error": "cancelled"})}, + "not a dict", + {"status": "ok", "result": "{not json"}, + {"status": "ok", "result": json.dumps([1])}, + ], + ) + async def test_browser_errors(self, providers, answer): + f = _app().state.forge_proxy + out, _ = await self._relay(f, answer) + assert "error" in json.loads(out["result"]) + assert f._txns == {} + + async def test_provider_gone_clears_txn(self, providers, monkeypatch): + f = _app().state.forge_proxy + txn_id = f.start("ws1", HOST) + monkeypatch.setenv(fp.PROVIDERS_KEY, "[]") + with pytest.raises(fp.ForgeProxyError): + await self._relay(f, {}, txn_id) + assert f._txns == {} + + +@pytest.mark.asyncio +class TestRouter: + async def test_requires_workspace_token(self, providers): + async with _client(_app()) as c: + assert ( + await c.get(f"/forge-proxy/{HOST}/api/v1/user") + ).status_code == 401 + + async def test_start_and_status(self, providers): + app = _app() + async with _client(app) as c: + r = await c.post( + f"/forge-proxy/{HOST}/_klangk/oauth/start", headers=_ws(app) + ) + assert set(r.json()) == {"txn_id"} + r = await c.get( + f"/forge-proxy/{HOST}/_klangk/status", headers=_ws(app) + ) + assert r.json() == {"connected": False, "login": None} + r = await c.post( + "/forge-proxy/other.test/_klangk/oauth/start", headers=_ws(app) + ) + assert r.status_code == 404 + r = await c.get( + "/forge-proxy/other.test/_klangk/status", headers=_ws(app) + ) + assert r.status_code == 404 + + async def test_forward_injects_token_and_filters_headers( + self, providers, upstream + ): + app = _app() + _authorized(app) + upstream.on( + "GET", + "/api/v1/user", + body={"login": "me"}, + headers={ + "Set-Cookie": "s=1", + "X-Total-Count": "1", + "Location": "https://x", + }, + ) + async with _client(app) as c: + r = await c.get( + f"/forge-proxy/{HOST}/api/v1/user", + headers={ + **_ws(app), + "Cookie": "c=1", + "Sudo": "admin", + "Accept": "application/json", + }, + ) + assert r.status_code == 200 and r.json() == {"login": "me"} + sent = upstream.requests[0] + assert sent.headers["authorization"] == "Bearer forge-at" + assert "cookie" not in sent.headers and "sudo" not in sent.headers + assert "set-cookie" not in r.headers and "location" not in r.headers + assert r.headers["x-total-count"] == "1" + + async def test_forward_with_query_and_body(self, providers, upstream): + app = _app() + _authorized(app) + upstream.on("POST", "/api/v1/repos/o/r/issues", 201, {"number": 7}) + upstream.on("GET", "/api/v1/repos/o/r/issues", body=[]) + async with _client(app) as c: + r = await c.post( + f"/forge-proxy/{HOST}/api/v1/repos/o/r/issues", + headers=_ws(app), + json={"title": "t", "body": "b"}, + ) + assert r.status_code == 201 + r = await c.get( + f"/forge-proxy/{HOST}/api/v1/repos/o/r/issues?state=open", + headers=_ws(app), + ) + assert r.status_code == 200 + assert upstream.requests[1].url.query == b"state=open" + + @pytest.mark.parametrize( + "method,path,status", + [ + ("GET", "/api/v1/admin/users", 403), + ("GET", "/api/v1/repos/o/r/issues?access_token=x", 400), + ("POST", "/api/v1/repos/o/r/issues", 400), + ], + ) + async def test_refusals(self, providers, upstream, method, path, status): + app = _app() + _authorized(app) + async with _client(app) as c: + r = await c.request( + method, + f"/forge-proxy/{HOST}{path}", + headers=_ws(app), + content=b"{}", + ) + assert r.status_code == status + assert upstream.requests == [] + + async def test_unproxied_host(self, providers, upstream): + app = _app() + async with _client(app) as c: + r = await c.get( + "/forge-proxy/other.test/api/v1/user", headers=_ws(app) + ) + assert r.status_code == 404 + + async def test_needs_auth(self, providers, upstream): + app = _app() + async with _client(app) as c: + r = await c.get( + f"/forge-proxy/{HOST}/api/v1/user", headers=_ws(app) + ) + assert r.status_code == 401 and "forge-auth" in r.json()["detail"] + + async def test_upstream_401_forgets(self, providers, upstream): + app = _app() + _authorized(app) + upstream.on("GET", "/api/v1/user", 401, {}) + async with _client(app) as c: + r = await c.get( + f"/forge-proxy/{HOST}/api/v1/user", headers=_ws(app) + ) + assert r.status_code == 401 + assert ("ws1", HOST) not in app.state.forge_proxy._tokens + + async def test_upstream_redirect_not_followed(self, providers, upstream): + app = _app() + _authorized(app) + upstream.on( + "GET", "/api/v1/user", 302, b"", {"Location": "https://evil/"} + ) + async with _client(app) as c: + r = await c.get( + f"/forge-proxy/{HOST}/api/v1/user", headers=_ws(app) + ) + assert r.status_code == 502 and "location" not in r.headers + + async def test_upstream_unreachable(self, providers, monkeypatch): + def boom(request): + raise httpx.ConnectError("down") + + monkeypatch.setattr( + routes, + "http_client_for", + lambda app: httpx.AsyncClient(transport=httpx.MockTransport(boom)), + ) + app = _app() + _authorized(app) + async with _client(app) as c: + r = await c.get( + f"/forge-proxy/{HOST}/api/v1/user", headers=_ws(app) + ) + assert r.status_code == 502 + + async def test_upstream_body_failure_truncates( + self, providers, monkeypatch + ): + class Broken(httpx.AsyncByteStream): + async def __aiter__(self): + yield b"part" + raise httpx.ReadError("reset") + + monkeypatch.setattr( + routes, + "http_client_for", + lambda app: httpx.AsyncClient( + transport=httpx.MockTransport( + lambda request: httpx.Response(200, stream=Broken()) + ) + ), + ) + app = _app() + _authorized(app) + async with _client(app) as c: + r = await c.get( + f"/forge-proxy/{HOST}/api/v1/user", headers=_ws(app) + ) + assert r.status_code == 200 and r.content == b"part" + + async def test_token_refresh_failure_closes_client( + self, providers, upstream + ): + app = _app() + _authorized(app, expires_at=time.time() + 5, refresh="") + async with _client(app) as c: + r = await c.get( + f"/forge-proxy/{HOST}/api/v1/user", headers=_ws(app) + ) + assert r.status_code == 401 + + async def test_raw_path_outside_prefix(self, providers, upstream): + """A raw path that does not carry the routed prefix (a proxy + normalized it differently) is refused, never re-rooted.""" + app = _app() + _authorized(app) + req = types.SimpleNamespace( + app=app, + method="GET", + scope={"raw_path": b"/elsewhere", "query_string": b""}, + ) + resp = await routes.forward(HOST, "api/v1/user", req, "ws1") + assert resp.status_code == 400 + + +class TestSslContext: + def test_default(self): + assert routes.ssl_context(make_settings(env={})) is not None + + def test_custom_ca(self, tmp_path): + import certifi + + pem = tmp_path / "ca.pem" + pem.write_text(open(certifi.where()).read()) + settings = make_settings(env={"KLANGKD_FORGE_PROXY_CA_CERT": str(pem)}) + assert routes.ssl_context(settings) is not None + + @pytest.mark.asyncio + async def test_http_client(self): + app = _app() + client = routes.http_client_for(app) + assert client.follow_redirects is False + await client.aclose() + + +@pytest.mark.asyncio +async def test_patch_forward(providers, upstream): + app = _app() + _authorized(app) + upstream.on( + "PATCH", "/api/v1/repos/o/r/issues/2", 201, {"state": "closed"} + ) + async with _client(app) as c: + r = await c.patch( + f"/forge-proxy/{HOST}/api/v1/repos/o/r/issues/2", + headers=_ws(app), + json={"state": "closed"}, + ) + assert r.status_code == 201 + + +def test_unwrap_dict_result(): + assert fp._unwrap({"status": "ok", "result": {"code": "c"}}) == { + "code": "c" + } + + +def test_container_flows_pass_when_no_forge_is_proxied(providers): + f = _app({"KLANGKD_FORGE_PROXY_HOSTS": ""}).state.forge_proxy + assert ( + f.bridge_policy( + { + "action": "git_credential", + "operation": "auth_flow_start", + "host": "github.com", + "authorize_url": "https://github.com/x", + } + ) + == "pass" + ) + + +class TestCapsAndCleanup: + def test_pending_txn_cap(self, providers): + f = _app().state.forge_proxy + for _ in range(fp.MAX_PENDING_TXNS_PER_WORKSPACE): + f.start("ws1", HOST) + with pytest.raises(fp.ForgeProxyError) as e: + f.start("ws1", HOST) + assert e.value.status == 429 + f.start("ws2", HOST) + + @pytest.mark.asyncio + async def test_txn_expired_at_completion(self, providers, monkeypatch): + f = _app().state.forge_proxy + txn = f.claim("ws1", f.start("ws1", HOST)) + later = time.time() + fp.TXN_TTL_SECONDS + 1 + monkeypatch.setattr(fp.time, "time", lambda: later) + with pytest.raises(fp.ForgeProxyError) as e: + await f.complete(txn, "code", txn.state, None) + assert e.value.status == 403 + + @pytest.mark.asyncio + async def test_dispatch_exception_clears_txn(self, providers): + f = _app().state.forge_proxy + txn_id = f.start("ws1", HOST) + + async def boom(request, timeout): + raise TimeoutError + + with pytest.raises(TimeoutError): + await f.relay_authorization("ws1", {"txn_id": txn_id}, boom, None) + assert f._txns == {} + + @pytest.mark.asyncio + async def test_browser_error_sanitized(self, providers): + f = _app().state.forge_proxy + txn_id = f.start("ws1", HOST) + + async def dispatch(request, timeout): + return {"error": ""} + + out = await f.relay_authorization( + "ws1", {"txn_id": txn_id}, dispatch, None + ) + assert json.loads(out["result"]) == {"error": "authorization failed"} + + @pytest.mark.asyncio + async def test_transport_error_on_exchange(self, providers): + def boom(request): + raise httpx.ConnectError("down") + + f = _app().state.forge_proxy + txn = f.claim("ws1", f.start("ws1", HOST)) + async with httpx.AsyncClient( + transport=httpx.MockTransport(boom) + ) as http: + with pytest.raises(fp.ForgeProxyError) as e: + await f.complete(txn, "code", txn.state, http) + assert e.value.status == 502 + + @pytest.mark.asyncio + async def test_malformed_json_token_answer(self, providers): + up = Upstream() + up.on( + "POST", + "/login/oauth/access_token", + 200, + b"{not json", + {"Content-Type": "application/json"}, + ) + f = _app().state.forge_proxy + txn = f.claim("ws1", f.start("ws1", HOST)) + async with up.client() as http: + with pytest.raises(fp.ForgeProxyError): + await f.complete(txn, "code", txn.state, http) + + @pytest.mark.asyncio + async def test_refresh_transport_error_forgets(self, providers): + def boom(request): + raise httpx.ConnectError("down") + + app = _app() + _authorized(app, expires_at=time.time() + 10) + async with httpx.AsyncClient( + transport=httpx.MockTransport(boom) + ) as http: + with pytest.raises(fp.ForgeProxyError) as e: + await app.state.forge_proxy.access_token("ws1", HOST, http) + assert e.value.status == 401 + + +@pytest.mark.asyncio +class TestRefreshConcurrency: + async def test_concurrent_requests_refresh_once(self, providers): + import asyncio + + app = _app() + _authorized(app, expires_at=time.time() + 10) + calls = [] + + async def handler(request): + calls.append(1) + await asyncio.sleep(0.01) + return httpx.Response( + 200, + json={"access_token": f"at{len(calls)}", "expires_in": 3600}, + ) + + async with httpx.AsyncClient( + transport=httpx.MockTransport(handler) + ) as http: + toks = await asyncio.gather( + *[ + app.state.forge_proxy.access_token("ws1", HOST, http) + for _ in range(5) + ] + ) + assert len(calls) == 1 + assert {t.access_token for t in toks} == {"at1"} + + async def test_waiter_sees_forgotten_token(self, providers): + import asyncio + + app = _app() + _authorized(app, expires_at=time.time() + 10) + f = app.state.forge_proxy + tok = f._tokens[("ws1", HOST)] + lock = f._locks.setdefault(("ws1", HOST), asyncio.Lock()) + await lock.acquire() + task = asyncio.ensure_future(f.access_token("ws1", HOST, None)) + await asyncio.sleep(0) + f.forget_token("ws1", HOST, tok) + lock.release() + with pytest.raises(fp.ForgeProxyError) as e: + await task + assert e.value.status == 401 + + async def test_stale_refresh_does_not_clobber_new_authorization( + self, providers + ): + app = _app() + _authorized(app, expires_at=time.time() + 10) + f = app.state.forge_proxy + old = f._tokens[("ws1", HOST)] + newer = fp.ForgeToken("new-at", "new-rt", 0.0, "me") + f._tokens[("ws1", HOST)] = newer + up = Upstream() + up.on( + "POST", "/login/oauth/access_token", body={"access_token": "stale"} + ) + async with up.client() as http: + served = await f._refresh(("ws1", HOST), old, http) + assert served is newer + assert f._tokens[("ws1", HOST)] is newer + f.forget_token("ws1", HOST, old) + assert f._tokens[("ws1", HOST)] is newer + + +@pytest.mark.asyncio +class TestBodyLimits: + async def test_json_body_too_large(self, providers, upstream): + app = _app() + _authorized(app) + async with _client(app) as c: + r = await c.post( + f"/forge-proxy/{HOST}/api/v1/repos/o/r/issues", + headers=_ws(app), + content=b"x" * (fp.MAX_JSON_BODY + 1), + ) + assert r.status_code == 413 and upstream.requests == [] + + async def test_git_body_streamed_and_capped( + self, providers, upstream, monkeypatch + ): + app = _app() + _authorized(app) + upstream.on("POST", "/o/r.git/git-upload-pack", 200, b"PACK") + async with _client(app) as c: + r = await c.post( + f"/forge-proxy/{HOST}/o/r.git/git-upload-pack", + headers=_ws(app), + content=b"0032want", + ) + assert r.status_code == 200 and r.content == b"PACK" + monkeypatch.setattr(fp, "MAX_GIT_BODY", 4) + r = await c.post( + f"/forge-proxy/{HOST}/o/r.git/git-upload-pack", + headers=_ws(app), + content=b"0032want", + ) + assert r.status_code == 413 + + +@pytest.mark.asyncio +async def test_refresh_after_forget_is_401(providers): + app = _app() + _authorized(app, expires_at=time.time() + 10) + f = app.state.forge_proxy + old = f._tokens.pop(("ws1", HOST)) + up = Upstream() + up.on("POST", "/login/oauth/access_token", body={"access_token": "late"}) + async with up.client() as http: + with pytest.raises(fp.ForgeProxyError) as e: + await f._refresh(("ws1", HOST), old, http) + assert e.value.status == 401 + + +# --- repository allow-list and push inspection -------------------------------- + +ZERO = "0" * 40 +OID = "a" * 40 + + +def _pkt(text: str) -> bytes: + data = text.encode() + return b"%04x" % (len(data) + 4) + data + + +def _push(*commands: str, pack: bytes = b"PACKDATA") -> bytes: + lines = [ + _pkt(c + ("\0report-status" if i == 0 else "")) + for i, c in enumerate(commands) + ] + return b"".join(lines) + b"0000" + pack + + +class TestRepoRules: + def test_unset_means_no_limit(self): + assert fp.parse_repo_rules("") == [] + assert fp.parse_repo_rules(" , ") == [] + + def test_parse(self): + rules = fp.parse_repo_rules( + "Owner/Repo.git, team/*:ro, bad, a/b/c, x/y:rw" + ) + assert rules == [ + fp.RepoRule("owner", "repo", False), + fp.RepoRule("team", "*", True), + ] + + def test_set_but_unusable_allows_nothing(self): + assert fp.parse_repo_rules("nonsense") == [] + route = fp.match_route("GET", "/api/v1/repos/o/r", "") + with pytest.raises(fp.ForgeProxyError) as e: + fp.check_repo(route, []) + assert e.value.status == 403 + + @pytest.mark.parametrize( + "method,path,query,allowed", + [ + ("GET", "/api/v1/repos/Owner/Repo/issues", "", True), + ("POST", "/owner/repo.git/git-receive-pack", "", True), + ("GET", "/api/v1/repos/team/any", "", True), + ( + "GET", + "/team/any.git/info/refs", + "service=git-upload-pack", + True, + ), + ( + "GET", + "/team/any.git/info/refs", + "service=git-receive-pack", + False, + ), + ("POST", "/team/any.git/git-receive-pack", "", False), + ("POST", "/api/v1/repos/team/any/issues", "", False), + ("GET", "/api/v1/repos/other/repo", "", False), + ("GET", "/api/v1/user", "", True), + ], + ) + def test_check(self, method, path, query, allowed): + rules = fp.parse_repo_rules("owner/repo, team/*:ro") + route = fp.match_route(method, path, query) + if allowed: + fp.check_repo(route, rules) + else: + with pytest.raises(fp.ForgeProxyError) as e: + fp.check_repo(route, rules) + assert e.value.status == 403 + + def test_parse_issues_mode(self): + rules = fp.parse_repo_rules("Team/Template:issues") + assert rules == [fp.RepoRule("team", "template", True, issues=True)] + + @pytest.mark.parametrize( + "method,path,query,allowed", + [ + ("GET", "/api/v1/repos/team/template", "", True), + ("GET", "/api/v1/repos/team/template/issues", "", True), + ("POST", "/api/v1/repos/team/template/issues", "", True), + ( + "POST", + "/api/v1/repos/team/template/issues/3/comments", + "", + True, + ), + ("PATCH", "/api/v1/repos/team/template/issues/3", "", True), + ("POST", "/team/template.git/git-receive-pack", "", False), + ( + "GET", + "/team/template.git/info/refs", + "service=git-receive-pack", + False, + ), + ( + "GET", + "/team/template.git/info/refs", + "service=git-upload-pack", + True, + ), + ( + "PUT", + "/api/v1/repos/team/template/collaborators/svc", + "", + False, + ), + ], + ) + def test_issues_mode(self, method, path, query, allowed): + rules = fp.parse_repo_rules("team/template:issues") + route = fp.match_route(method, path, query) + if allowed: + fp.check_repo(route, rules) + else: + with pytest.raises(fp.ForgeProxyError) as e: + fp.check_repo(route, rules) + assert e.value.status == 403 + + def test_exact_rule_wins_over_wildcard(self): + rules = fp.parse_repo_rules("team/*:ro, team/docs") + route = fp.match_route("POST", "/api/v1/repos/team/docs/issues", "") + fp.check_repo(route, rules) + + def test_unset_refuses_repo_routes(self): + with pytest.raises(fp.ForgeProxyError): + fp.check_repo(fp.match_route("GET", "/api/v1/repos/o/r", ""), []) + + +class TestPushInspection: + def test_create_and_update_pass(self): + body = _push( + f"{ZERO} {OID} refs/heads/new", f"{OID} {'b' * 40} refs/heads/main" + ) + end = fp.push_commands_end(body) + assert body[end:] == b"PACKDATA" + fp.check_push_commands(body, end) + + def test_shallow_line_passes(self): + body = _pkt(f"shallow {OID}") + _push(f"{ZERO} {OID} refs/heads/x") + fp.check_push_commands(body, fp.push_commands_end(body)) + + def test_delete_refused(self): + body = _push(f"{OID} {ZERO} refs/heads/main") + with pytest.raises(fp.ForgeProxyError) as e: + fp.check_push_commands(body, fp.push_commands_end(body)) + assert e.value.status == 403 + + def test_sha256_delete_refused(self): + body = _push(f"{'c' * 64} {'0' * 64} refs/heads/main") + with pytest.raises(fp.ForgeProxyError): + fp.check_push_commands(body, fp.push_commands_end(body)) + + @pytest.mark.parametrize( + "line", ["push-cert", f"{OID} {OID}", "garbage here now"] + ) + def test_unknown_command_refused(self, line): + body = _push(line) + with pytest.raises(fp.ForgeProxyError) as e: + fp.check_push_commands(body, fp.push_commands_end(body)) + assert e.value.status == 400 + + def test_incomplete_needs_more(self): + assert fp.push_commands_end(_pkt(f"{ZERO} {OID} refs/heads/x")) is None + assert fp.push_commands_end(b"00") is None + + @pytest.mark.parametrize("head", [b"zzzz", b"0002"]) + def test_malformed_length(self, head): + with pytest.raises(fp.ForgeProxyError) as e: + fp.push_commands_end(head + b"rest") + assert e.value.status == 400 + + +@pytest.mark.asyncio +class TestRouterPolicy: + def _app(self, repos="owner/repo, team/*:ro"): + return _app(env={"KLANGKD_FORGE_PROXY_ALLOWED_REPOS": repos}) + + async def test_disallowed_repo_api_and_git(self, providers, upstream): + app = self._app() + _authorized(app) + async with _client(app) as c: + for path in ( + "/api/v1/repos/other/repo/issues", + "/other/repo.git/info/refs?service=git-upload-pack", + ): + r = await c.get(f"/forge-proxy/{HOST}{path}", headers=_ws(app)) + assert r.status_code == 403 + assert upstream.requests == [] + + async def test_read_only_repo(self, providers, upstream): + app = self._app() + _authorized(app) + upstream.on("GET", "/team/docs.git/info/refs", 200, b"refs") + async with _client(app) as c: + r = await c.get( + f"/forge-proxy/{HOST}/team/docs.git/info/refs?service=git-upload-pack", + headers=_ws(app), + ) + assert r.status_code == 200 + r = await c.get( + f"/forge-proxy/{HOST}/team/docs.git/info/refs?service=git-receive-pack", + headers=_ws(app), + ) + assert r.status_code == 403 + assert len(upstream.requests) == 1 + + async def test_push_create_streams_whole_body(self, providers, upstream): + app = self._app() + _authorized(app) + upstream.on("POST", "/owner/repo.git/git-receive-pack", 200, b"ok") + body = _push(f"{ZERO} {OID} refs/heads/x") + async with _client(app) as c: + r = await c.post( + f"/forge-proxy/{HOST}/owner/repo.git/git-receive-pack", + headers=_ws(app), + content=body, + ) + assert r.status_code == 200 + assert upstream.requests[0].read() == body + + async def test_push_delete_refused(self, providers, upstream): + app = self._app() + _authorized(app) + async with _client(app) as c: + r = await c.post( + f"/forge-proxy/{HOST}/owner/repo.git/git-receive-pack", + headers=_ws(app), + content=_push(f"{OID} {ZERO} refs/heads/x"), + ) + assert r.status_code == 403 and upstream.requests == [] + + async def test_push_refusals(self, providers, upstream, monkeypatch): + app = self._app() + _authorized(app) + url = f"/forge-proxy/{HOST}/owner/repo.git/git-receive-pack" + async with _client(app) as c: + r = await c.post( + url, + headers={**_ws(app), "Content-Encoding": "gzip"}, + content=_push(f"{ZERO} {OID} refs/heads/x"), + ) + assert r.status_code == 415 + r = await c.post( + url, + headers=_ws(app), + content=_pkt(f"{ZERO} {OID} refs/heads/x"), + ) + assert r.status_code == 400 + monkeypatch.setattr(fp, "MAX_PUSH_COMMANDS", 8) + r = await c.post( + url, + headers=_ws(app), + content=_pkt(f"{ZERO} {OID} refs/heads/x"), + ) + assert r.status_code == 413 + assert upstream.requests == [] + + async def test_push_pack_capped(self, providers, upstream, monkeypatch): + app = self._app() + _authorized(app) + upstream.on("POST", "/owner/repo.git/git-receive-pack", 200, b"ok") + body = _push(f"{ZERO} {OID} refs/heads/x", pack=b"P" * 64) + + async def chunked(): + yield body[:60] + yield body[60:] + + monkeypatch.setattr(fp, "MAX_GIT_BODY", len(body) - 1) + async with _client(app) as c: + r = await c.post( + f"/forge-proxy/{HOST}/owner/repo.git/git-receive-pack", + headers=_ws(app), + content=chunked(), + ) + assert r.status_code == 413 + + +# --- site creation: template generate and collaborator add ------------------- + + +GEN = "/api/v1/repos/tmpl/site-template/generate" +GEN_BODY = {"owner": "me", "name": "new-site", "git_content": True} + + +class TestSiteCreationPolicy: + def test_routes(self): + r = fp.match_route("POST", GEN, "") + assert (r.kind, r.owner, r.repo) == ( + "repo-generate", + "tmpl", + "site-template", + ) + r = fp.match_route( + "PUT", "/api/v1/repos/me/new-site/collaborators/build_svc", "" + ) + assert (r.kind, r.owner, r.repo, r.user) == ( + "collaborator-add", + "me", + "new-site", + "build_svc", + ) + assert fp.is_write(r) + + def test_generate_body(self): + out = fp.validate_json_body( + "repo-generate", + json.dumps( + {**GEN_BODY, "private": True, "description": "d"} + ).encode(), + ) + assert json.loads(out)["private"] is True + + @pytest.mark.parametrize( + "kind,body", + [ + ("repo-generate", {"owner": "me"}), + ("repo-generate", {**GEN_BODY, "webhooks": True}), + ("repo-generate", {**GEN_BODY, "git_content": "yes"}), + ("repo-generate", {**GEN_BODY, "private": 1}), + ("repo-generate", {**GEN_BODY, "name": "../x"}), + ("repo-generate", {**GEN_BODY, "owner": "a b"}), + ("collaborator-add", {"permission": "write"}), + ("collaborator-add", {"permission": True}), + ("collaborator-add", {}), + ("issue-state", {"state": "merged"}), + ], + ) + def test_bad_bodies(self, kind, body): + with pytest.raises(fp.ForgeProxyError) as e: + fp.validate_json_body(kind, json.dumps(body).encode()) + assert e.value.status == 400 + + def test_parse_names(self): + assert fp.parse_names(" A/B , ,c ") == {"a/b", "c"} + assert fp.parse_names("") == set() + + def test_check_template(self): + route = fp.match_route("POST", GEN, "") + fp.check_template(route, {"tmpl/site-template"}) + with pytest.raises(fp.ForgeProxyError) as e: + fp.check_template(route, {"other/template"}) + assert e.value.status == 403 + + @pytest.mark.parametrize( + "owner,login,owners,rules,ok", + [ + ("me", "Me", set(), None, True), + ("team", "me", {"team"}, None, True), + ("team", "me", set(), None, False), + ("me", "", set(), None, False), + ("me", "me", set(), "me/*", True), + ("me", "me", set(), "me/*:ro", False), + ("me", "me", set(), "other/*", False), + ], + ) + def test_check_new_repo(self, owner, login, owners, rules, ok): + body = {"owner": owner, "name": "new-site"} + parsed = fp.parse_repo_rules(rules or "*") + if ok: + fp.check_new_repo(body, login, owners, parsed) + else: + with pytest.raises(fp.ForgeProxyError) as e: + fp.check_new_repo(body, login, owners, parsed) + assert e.value.status == 403 + + @pytest.mark.parametrize( + "path,login,ok", + [ + ("/api/v1/repos/me/r/collaborators/Build_Svc", "ME", True), + ("/api/v1/repos/other/r/collaborators/build_svc", "me", False), + ("/api/v1/repos/me/r/collaborators/someone", "me", False), + ("/api/v1/repos/me/r/collaborators/build_svc", "", False), + ], + ) + def test_check_collaborator(self, path, login, ok): + route = fp.match_route("PUT", path, "") + if ok: + fp.check_collaborator(route, login, {"build_svc"}) + else: + with pytest.raises(fp.ForgeProxyError) as e: + fp.check_collaborator(route, login, {"build_svc"}) + assert e.value.status == 403 + + +@pytest.mark.asyncio +class TestSiteCreationRouter: + def _app(self, **env): + base = { + "KLANGKD_FORGE_PROXY_ALLOWED_REPOS": "me/*, ops/registry", + "KLANGKD_FORGE_PROXY_TEMPLATE_REPOS": "tmpl/site-template", + "KLANGKD_FORGE_PROXY_COLLABORATOR_ACCOUNTS": "build_svc", + } + return _app(env={**base, **env}) + + async def test_generate_and_collaborator(self, providers, upstream): + app = self._app() + _authorized(app) + upstream.on("POST", GEN, 201, {"full_name": "me/new-site"}) + upstream.on( + "PUT", "/api/v1/repos/me/new-site/collaborators/build_svc", 204 + ) + async with _client(app) as c: + r = await c.post( + f"/forge-proxy/{HOST}{GEN}", headers=_ws(app), json=GEN_BODY + ) + assert r.status_code == 201 + r = await c.put( + f"/forge-proxy/{HOST}/api/v1/repos/me/new-site/collaborators/build_svc", + headers=_ws(app), + json={"permission": "read"}, + ) + assert r.status_code == 204 + assert json.loads(upstream.requests[0].read()) == GEN_BODY + assert ( + upstream.requests[1].headers["authorization"] == "Bearer forge-at" + ) + + @pytest.mark.parametrize( + "method,path,body,env", + [ + # Template not listed. + ("POST", "/api/v1/repos/tmpl/other/generate", GEN_BODY, {}), + # Generating into someone else's account. + ("POST", GEN, {**GEN_BODY, "owner": "victim"}, {}), + # Owner allowed, but the new repo is outside the allow-list. + ( + "POST", + GEN, + {**GEN_BODY, "owner": "team"}, + {"KLANGKD_FORGE_PROXY_TEMPLATE_OWNERS": "team"}, + ), + # Collaborator on a repository the user does not own. + ( + "PUT", + "/api/v1/repos/ops/registry/collaborators/build_svc", + {"permission": "read"}, + {}, + ), + # An account not in the collaborator list. + ( + "PUT", + "/api/v1/repos/me/new-site/collaborators/mallory", + {"permission": "read"}, + {}, + ), + ], + ) + async def test_refused(self, providers, upstream, method, path, body, env): + app = self._app(**env) + _authorized(app) + async with _client(app) as c: + r = await c.request( + method, + f"/forge-proxy/{HOST}{path}", + headers=_ws(app), + json=body, + ) + assert r.status_code == 403 + assert upstream.requests == [] + + async def test_write_permission_refused(self, providers, upstream): + app = self._app() + _authorized(app) + async with _client(app) as c: + r = await c.put( + f"/forge-proxy/{HOST}/api/v1/repos/me/new-site/collaborators/build_svc", + headers=_ws(app), + json={"permission": "admin"}, + ) + assert r.status_code == 400 and upstream.requests == [] + + async def test_generate_refused_when_no_templates( + self, providers, upstream + ): + app = self._app(KLANGKD_FORGE_PROXY_TEMPLATE_REPOS="") + _authorized(app) + async with _client(app) as c: + r = await c.post( + f"/forge-proxy/{HOST}{GEN}", headers=_ws(app), json=GEN_BODY + ) + assert r.status_code == 403 and upstream.requests == [] + + +class TestFailClosed: + def test_unset_allows_no_repository(self): + route = fp.match_route("GET", "/api/v1/repos/o/r", "") + with pytest.raises(fp.ForgeProxyError) as e: + fp.check_repo(route, fp.parse_repo_rules("")) + assert e.value.status == 403 + + def test_account_route_passes_without_rules(self): + fp.check_repo(fp.match_route("GET", "/api/v1/user", ""), []) + + @pytest.mark.parametrize("raw", ["*", "*/*"]) + def test_star_allows_everything(self, raw): + rules = fp.parse_repo_rules(raw) + assert rules == [fp.RepoRule("*", "*", False)] + fp.check_repo( + fp.match_route("POST", "/a/b.git/git-receive-pack", ""), rules + ) + + def test_star_modes_and_specific_rules_win(self): + rules = fp.parse_repo_rules("*:issues, me/*") + push = fp.match_route("POST", "/other/x.git/git-receive-pack", "") + with pytest.raises(fp.ForgeProxyError): + fp.check_repo(push, rules) + fp.check_repo( + fp.match_route("POST", "/api/v1/repos/other/x/issues", ""), rules + ) + fp.check_repo( + fp.match_route("POST", "/me/x.git/git-receive-pack", ""), rules + ) + assert fp.parse_repo_rules("*:ro") == [fp.RepoRule("*", "*", True)] + + @pytest.mark.asyncio + async def test_router_refuses_when_unset(self, providers, upstream): + app = _app(env={"KLANGKD_FORGE_PROXY_ALLOWED_REPOS": ""}) + _authorized(app) + upstream.on("GET", "/api/v1/user", body={"login": "me"}) + async with _client(app) as c: + r = await c.get( + f"/forge-proxy/{HOST}/api/v1/repos/o/r", headers=_ws(app) + ) + assert r.status_code == 403 + r = await c.get( + f"/forge-proxy/{HOST}/api/v1/user", headers=_ws(app) + ) + assert r.status_code == 200 + + +class TestIssueAssignees: + @pytest.mark.parametrize( + "kind,body", + [ + ("issue-state", {"assignees": ["dev.one"]}), + ("issue-state", {"state": "closed", "assignees": []}), + ("issue-create", {"title": "t", "assignees": ["a", "b"]}), + ], + ) + def test_allowed(self, kind, body): + out = fp.validate_json_body(kind, json.dumps(body).encode()) + assert json.loads(out) == body + + @pytest.mark.parametrize( + "kind,body", + [ + ("issue-state", {}), + ("issue-state", {"ref": "main"}), + ("issue-state", {"assignees": "someone"}), + ("issue-state", {"assignees": [1]}), + ("issue-state", {"assignees": ["a/b"]}), + ("issue-state", {"assignees": ["a"] * 11}), + ], + ) + def test_refused(self, kind, body): + with pytest.raises(fp.ForgeProxyError) as e: + fp.validate_json_body(kind, json.dumps(body).encode()) + assert e.value.status == 400 + + def test_issues_rule_allows_edit_but_not_push(self): + rules = fp.parse_repo_rules("team/*:issues") + fp.check_repo( + fp.match_route("PATCH", "/api/v1/repos/team/r/issues/3", ""), rules + ) + with pytest.raises(fp.ForgeProxyError): + fp.check_repo( + fp.match_route("POST", "/team/r.git/git-receive-pack", ""), + rules, + ) + + +class TestIssueSurface: + @pytest.mark.parametrize( + "kind,body", + [ + ("issue-state", {"title": "t", "body": "b", "milestone": 2}), + ( + "issue-create", + {"title": "t", "labels": [3, "bug"], "milestone": 1}, + ), + ("issue-labels", {"labels": ["bug", 7]}), + ], + ) + def test_allowed(self, kind, body): + assert ( + json.loads(fp.validate_json_body(kind, json.dumps(body).encode())) + == body + ) + + def test_too_many_labels(self): + with pytest.raises(fp.ForgeProxyError): + fp.validate_json_body( + "issue-labels", + json.dumps({"labels": list(range(1, 22))}).encode(), + ) + + @pytest.mark.parametrize( + "method,path,kind", + [ + ("GET", "/api/v1/repos/o/r/labels", "api-read"), + ("GET", "/api/v1/repos/o/r/milestones", "api-read"), + ("GET", "/api/v1/repos/o/r/assignees", "api-read"), + ("GET", "/api/v1/repos/o/r/collaborators", "api-read"), + ("POST", "/api/v1/repos/o/r/issues/4/labels", "issue-labels"), + ("PUT", "/api/v1/repos/o/r/issues/4/labels", "issue-labels"), + ], + ) + def test_routes(self, method, path, kind): + assert fp.match_route(method, path, "").kind == kind + + def test_issues_rule_allows_labels(self): + rules = fp.parse_repo_rules("team/*:issues") + fp.check_repo( + fp.match_route("PUT", "/api/v1/repos/team/r/issues/1/labels", ""), + rules, + ) + + +class TestPassthrough: + def test_off_by_default(self): + with pytest.raises(fp.ForgeProxyError) as e: + fp.match_route("GET", "/api/v1/orgs/x", "") + assert e.value.status == 403 + + @pytest.mark.parametrize( + "method,path,kind,owner", + [ + ("GET", "/api/v1/orgs/team/teams", "api-pass-read", ""), + ( + "DELETE", + "/api/v1/repos/o/r/issues/3/labels/2", + "api-pass-write", + "o", + ), + ("POST", "/api/v1/repos/o/r", "api-pass-write", "o"), + ], + ) + def test_routes(self, method, path, kind, owner): + r = fp.match_route(method, path, "page=2", passthrough=True) + assert (r.kind, r.owner, r.query) == (kind, owner, "page=2") + + def test_listed_routes_keep_their_kind(self): + r = fp.match_route("GET", "/api/v1/repos/o/r", "", passthrough=True) + assert r.kind == "api-read" + + @pytest.mark.parametrize( + "method,path,query,status", + [ + ("GET", "/api/v1/user", "access_token=x", 400), + ("GET", "/api/v1/orgs/x", "Sudo=admin", 400), + ("OPTIONS", "/api/v1/orgs/x", "", 405), + ("GET", "/o/r/raw/main/file", "", 403), + ], + ) + def test_refusals(self, method, path, query, status): + with pytest.raises(fp.ForgeProxyError) as e: + fp.match_route(method, path, query, passthrough=True) + assert e.value.status == status + + def test_repo_rules_still_apply(self): + rules = fp.parse_repo_rules("team/*:issues") + route = fp.match_route( + "DELETE", "/api/v1/repos/team/r/issues/3", "", passthrough=True + ) + with pytest.raises(fp.ForgeProxyError): + fp.check_repo(route, rules) + other = fp.match_route( + "GET", "/api/v1/repos/other/r/topics", "", passthrough=True + ) + with pytest.raises(fp.ForgeProxyError): + fp.check_repo(other, rules) + + @pytest.mark.asyncio + async def test_router_forwards_any_api_call(self, providers, upstream): + app = _app(env={"KLANGKD_FORGE_PROXY_FEATURES": "*"}) + _authorized(app) + upstream.on("DELETE", "/api/v1/repos/o/r/issues/3/labels/2", 204, b"") + upstream.on("GET", "/api/v1/orgs/team/teams", body=[]) + async with _client(app) as c: + r = await c.delete( + f"/forge-proxy/{HOST}/api/v1/repos/o/r/issues/3/labels/2", + headers=_ws(app), + ) + assert r.status_code == 204 + r = await c.get( + f"/forge-proxy/{HOST}/api/v1/orgs/team/teams", headers=_ws(app) + ) + assert r.status_code == 200 + assert ( + upstream.requests[0].headers["authorization"] == "Bearer forge-at" + ) + + @pytest.mark.asyncio + async def test_router_allowlist_mode_refuses(self, providers, upstream): + app = _app() + _authorized(app) + async with _client(app) as c: + r = await c.delete( + f"/forge-proxy/{HOST}/api/v1/repos/o/r/issues/3", + headers=_ws(app), + ) + assert r.status_code == 403 and upstream.requests == [] + + +class TestFeatures: + def test_parse(self): + assert fp.parse_features("") == set() + assert fp.parse_features(" Read , issues, bogus") == {"read", "issues"} + assert fp.parse_features("*") == set(fp.FEATURES) + + @pytest.mark.parametrize( + "method,path,query,feature", + [ + ( + "GET", + "/o/r.git/info/refs", + "service=git-upload-pack", + "git-read", + ), + ( + "GET", + "/o/r.git/info/refs", + "service=git-receive-pack", + "git-push", + ), + ("POST", "/o/r.git/git-receive-pack", "", "git-push"), + ("POST", "/api/v1/repos/o/r/issues", "", "issues"), + ("PATCH", "/api/v1/repos/o/r/issues/2", "", "issue-edit"), + ("GET", "/api/v1/repos/o/r", "", "read"), + ], + ) + def test_route_feature(self, method, path, query, feature): + assert fp.route_feature(fp.match_route(method, path, query)) == feature + + def test_locked_down_by_default(self): + route = fp.match_route("GET", "/api/v1/user", "") + with pytest.raises(fp.ForgeProxyError) as e: + fp.check_feature(route, fp.parse_features("")) + assert e.value.status == 403 + + @pytest.mark.asyncio + async def test_router_default_refuses_everything( + self, providers, upstream + ): + app = _app(env={"KLANGKD_FORGE_PROXY_FEATURES": ""}) + _authorized(app) + async with _client(app) as c: + for method, path in ( + ("GET", "/api/v1/user"), + ("GET", "/o/r.git/info/refs?service=git-upload-pack"), + ("POST", "/api/v1/repos/o/r/issues"), + ): + r = await c.request( + method, + f"/forge-proxy/{HOST}{path}", + headers=_ws(app), + content=b'{"title":"t"}', + ) + assert r.status_code == 403 + assert upstream.requests == [] + + @pytest.mark.asyncio + async def test_router_read_only_features(self, providers, upstream): + app = _app(env={"KLANGKD_FORGE_PROXY_FEATURES": "read,git-read"}) + _authorized(app) + upstream.on("GET", "/api/v1/user", body={"login": "me"}) + async with _client(app) as c: + assert ( + await c.get( + f"/forge-proxy/{HOST}/api/v1/user", headers=_ws(app) + ) + ).status_code == 200 + r = await c.post( + f"/forge-proxy/{HOST}/api/v1/repos/o/r/issues", + headers=_ws(app), + json={"title": "t"}, + ) + assert r.status_code == 403 + + +class TestCodexFindings: + @pytest.mark.parametrize( + "path", ["/api/v1/repos/o/r#", "/api/v1/repos/o/r?x"] + ) + def test_delimiters_in_path_rejected(self, path): + with pytest.raises(fp.ForgeProxyError) as e: + fp.match_route("DELETE", path, "", passthrough=True) + assert e.value.status == 400 + + @pytest.mark.parametrize( + "method,path", + [ + ("POST", "/api/v1/user/repos"), + ("GET", "/api/v1/repositories/42"), + ("GET", "/api/v1/repos/search"), + ("GET", "/api/v1/orgs/team/repos"), + ], + ) + def test_unscoped_passthrough_needs_star(self, method, path): + route = fp.match_route(method, path, "", passthrough=True) + with pytest.raises(fp.ForgeProxyError): + fp.check_repo(route, fp.parse_repo_rules("me/*")) + fp.check_repo(route, fp.parse_repo_rules("*")) + + def test_unscoped_passthrough_respects_star_mode(self): + route = fp.match_route( + "POST", "/api/v1/user/repos", "", passthrough=True + ) + with pytest.raises(fp.ForgeProxyError): + fp.check_repo(route, fp.parse_repo_rules("*:ro")) + + +@pytest.mark.asyncio +class TestErrorBodies: + async def test_api_error_body_replaced(self, providers, upstream): + app = _app() + _authorized(app) + upstream.on( + "GET", + "/api/v1/repos/o/r", + 500, + {"message": "token forge-at rejected"}, + ) + async with _client(app) as c: + r = await c.get( + f"/forge-proxy/{HOST}/api/v1/repos/o/r", headers=_ws(app) + ) + assert r.status_code == 500 + assert "forge-at" not in r.text and r.json()["detail"].endswith( + "(500)" + ) + + async def test_git_error_body_kept(self, providers, upstream): + app = _app() + _authorized(app) + upstream.on( + "POST", "/o/r.git/git-upload-pack", 403, b"denied by forge" + ) + async with _client(app) as c: + r = await c.post( + f"/forge-proxy/{HOST}/o/r.git/git-upload-pack", + headers=_ws(app), + content=b"0000", + ) + assert r.status_code == 403 and r.content == b"denied by forge" diff --git a/src/klangk/klangkd-tests/tests/test_git_credential.py b/src/klangk/klangkd-tests/tests/test_git_credential.py index 80d10f4ee..803252f36 100644 --- a/src/klangk/klangkd-tests/tests/test_git_credential.py +++ b/src/klangk/klangkd-tests/tests/test_git_credential.py @@ -2465,3 +2465,176 @@ def test_unreachable_refresh_keeps_the_cached_credential( ops = [r["operation"] for r in _BridgeHandler.requests] assert "erase" not in ops assert "auth_flow_start" not in ops + + +# --- forge proxy (token stays on the host) ---------------------------------- + + +class TestForgeProxyHelper: + EGRESS = "http://host.containers.internal:8995" + + def test_capability_lines_collected(self, monkeypatch): + import io + + monkeypatch.setattr( + sys, + "stdin", + io.StringIO( + "capability[]=authtype\ncapability[]=state\nhost=h\n\n" + ), + ) + cred = helper.read_credential_input() + assert cred["capability"] == ["authtype", "state"] + + def test_is_forge_proxy_request(self, monkeypatch): + monkeypatch.setattr(helper, "BRIDGE_URL", self.EGRESS) + hit = { + "host": "host.containers.internal:8995", + "path": "forge-proxy/f/o/r.git", + } + assert helper.is_forge_proxy_request(hit) + assert not helper.is_forge_proxy_request( + {**hit, "path": "llm-proxy/x"} + ) + assert not helper.is_forge_proxy_request({**hit, "host": "forge.test"}) + monkeypatch.setattr(helper, "BRIDGE_URL", "") + assert not helper.is_forge_proxy_request(hit) + + def test_forge_proxy_output(self, monkeypatch): + monkeypatch.setattr(helper, "_get_workspace_token", lambda: "ws-jwt") + out = helper.forge_proxy_output({"capability": ["authtype"]}) + assert "authtype=Bearer" in out and "credential=ws-jwt" in out + assert "password" not in out + assert helper.forge_proxy_output({}) is None + monkeypatch.setattr(helper, "_get_workspace_token", lambda: "") + assert helper.forge_proxy_output({"capability": ["authtype"]}) is None + + def test_forge_auth_already_connected(self, monkeypatch, capsys): + calls = [] + monkeypatch.setattr( + helper, + "_post_egress_json", + lambda path, timeout=30: {"connected": True, "login": "me"}, + ) + monkeypatch.setattr(helper, "_configure_forge_rewrite", calls.append) + assert helper.run_forge_auth("Forge.Test", "bid") == 0 + assert ( + calls == ["forge.test"] + and "already authorized" in capsys.readouterr().out + ) + + def test_forge_auth_runs_txn_through_bridge(self, monkeypatch, capsys): + answers = {"status": {"connected": False}, "start": {"txn_id": "t1"}} + monkeypatch.setattr( + helper, + "_post_egress_json", + lambda path, timeout=30: answers[ + "start" if path.endswith("/start") else "status" + ], + ) + sent = {} + + def fake_bridge(op, cred, bid, timeout=None, extra=None): + sent.update(op=op, extra=extra) + return { + "status": "ok", + "result": json.dumps({"status": "connected", "login": "me"}), + } + + monkeypatch.setattr(helper, "post_bridge", fake_bridge) + monkeypatch.setattr( + helper, "_configure_forge_rewrite", lambda host: None + ) + assert helper.run_forge_auth("forge.test", "bid") == 0 + assert sent == {"op": "auth_flow_start", "extra": {"txn_id": "t1"}} + assert "authorized forge.test as me" in capsys.readouterr().out + + def test_forge_auth_failures(self, monkeypatch): + monkeypatch.setattr( + helper, + "_post_egress_json", + lambda path, timeout=30: {"detail": "nope"}, + ) + assert helper.run_forge_auth("forge.test", "bid") == 1 + monkeypatch.setattr( + helper, + "_post_egress_json", + lambda path, timeout=30: ( + {"txn_id": "t"} if path.endswith("/start") else {} + ), + ) + monkeypatch.setattr(helper, "post_bridge", lambda *a, **k: None) + assert helper.run_forge_auth("forge.test", "bid") == 1 + + def test_post_egress_json(self, monkeypatch): + class H(BaseHTTPRequestHandler): + def _send(self, code, body): + self.send_response(code) + self.send_header("Content-Length", str(len(body))) + self.send_header("Connection", "close") + self.end_headers() + self.wfile.write(body) + + def do_GET(self): + self._send(200, b'{"connected": false}') + + def do_POST(self): + self._send(404, b'{"detail": "forge is not proxied"}') + + def log_message(self, *a): + pass + + server = HTTPServer(("127.0.0.1", 0), H) + Thread(target=server.serve_forever, daemon=True).start() + try: + monkeypatch.setattr( + helper, "BRIDGE_URL", f"http://127.0.0.1:{server.server_port}" + ) + monkeypatch.setattr(helper, "_get_workspace_token", lambda: "") + assert helper._post_egress_json( + "/forge-proxy/f/_klangk/status" + ) == {"connected": False} + assert helper._post_egress_json( + "/forge-proxy/f/_klangk/oauth/start" + ) == {"detail": "forge is not proxied"} + monkeypatch.setattr(helper, "BRIDGE_URL", "http://127.0.0.1:1") + assert "detail" in helper._post_egress_json( + "/forge-proxy/f/_klangk/status", timeout=2 + ) + finally: + server.shutdown() + + def test_get_for_proxy_answers_bearer_without_bridge( + self, fake_browser_id + ): + token = fake_browser_id / "klangk-workspace-token" + token.write_text("#!/bin/sh\necho ws-jwt\n") + result = run_helper( + "get", + "capability[]=authtype\nprotocol=http\nhost=host.containers.internal:8995\n" + "path=forge-proxy/forge.test/o/r.git/info/refs\n\n", + env_override={"KLANGKWS_BRIDGE_URL": self.EGRESS}, + extra_path=str(fake_browser_id), + ) + assert result.returncode == 0 + assert ( + "authtype=Bearer" in result.stdout + and "credential=ws-jwt" in result.stdout + ) + + def test_store_for_proxy_is_dropped(self, fake_browser_id): + result = run_helper( + "store", + "protocol=http\nhost=host.containers.internal:8995\npath=forge-proxy/f/x\n\n", + env_override={"KLANGKWS_BRIDGE_URL": self.EGRESS}, + extra_path=str(fake_browser_id), + ) + assert result.returncode == 0 and result.stdout == "" + + def test_forge_auth_usage(self, fake_browser_id): + result = run_helper( + "forge-auth", + env_override={"KLANGKWS_BRIDGE_URL": self.EGRESS}, + extra_path=str(fake_browser_id), + ) + assert result.returncode == 2 and "usage" in result.stderr