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