diff --git a/backend/app/api/deps.py b/backend/app/api/deps.py index ba495e7..ee01751 100644 --- a/backend/app/api/deps.py +++ b/backend/app/api/deps.py @@ -8,6 +8,7 @@ from jwt.exceptions import InvalidTokenError from pydantic import ValidationError from sqlmodel import Session +from app.cloud import config as cloud_config from app.core import security from app.core.config import settings from app.core.db import engine @@ -39,8 +40,11 @@ def decode_token(token: str) -> dict[str, Any]: rights — the difference is only in who is holding it, which the MCP endpoint checks separately. - This is also the seam a hosted deployment widens later: trusting an - additional issuer is a third branch here, not a change anywhere else. + The third branch is the seam a hosted deployment widens: a portal this + installation was enrolled with signs tokens with a key pinned at + enrolment, and they resolve to the local account that performed it. With + no enrolment the branch raises immediately, so an offline installation + pays nothing for the possibility. """ try: session: dict[str, Any] = jwt.decode( @@ -48,7 +52,11 @@ def decode_token(token: str) -> dict[str, Any]: ) return session except InvalidTokenError: + pass + try: return security.decode_oauth_token(token) + except InvalidTokenError: + return cloud_config.decode_portal_token(token) def user_from_token(session: Session, token: str) -> User | None: diff --git a/backend/app/api/main.py b/backend/app/api/main.py index 42c6820..1634249 100644 --- a/backend/app/api/main.py +++ b/backend/app/api/main.py @@ -3,6 +3,7 @@ from fastapi import APIRouter from app.api.routes import ( alerts, artifacts, + cloud, dashboards, flows, login, @@ -33,6 +34,9 @@ api_router.include_router(observability.router) api_router.include_router(runs.router) api_router.include_router(artifacts.router) api_router.include_router(workers.router) +# Remote access through a portal. Always mounted; with no enrolment the +# endpoints only ever report that there is none. +api_router.include_router(cloud.router) # Always mounted so the generated SDK stays the same shape; the endpoints # themselves refuse to work unless MCP is switched on. api_router.include_router(oauth.router) diff --git a/backend/app/api/routes/cloud.py b/backend/app/api/routes/cloud.py new file mode 100644 index 0000000..088664c --- /dev/null +++ b/backend/app/api/routes/cloud.py @@ -0,0 +1,145 @@ +"""Connecting this installation to a Fluksio portal, and cutting it loose. + +Entirely optional, and superuser-only to change: enrolling grants a remote +party the rights of the account that performed it, which is not a decision an +ordinary user of this installation gets to make on everyone else's behalf. +""" + +from __future__ import annotations + +import asyncio +import logging +from datetime import UTC, datetime +from typing import Any + +import httpx +from fastapi import APIRouter, Depends, HTTPException, Request +from pydantic import BaseModel, Field + +from app.api.deps import CurrentUser, get_current_active_superuser, get_current_user +from app.cloud import config as cloud_config +from app.models import Message + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/cloud", tags=["cloud"]) + + +class EnrollBody(BaseModel): + #: Where the portal's API lives, e.g. https://hub.fluksio.com + portal_url: str = Field(min_length=1, max_length=255) + claim_code: str = Field(min_length=1, max_length=32) + + +@router.get("/status", dependencies=[Depends(get_current_user)]) +def read_status(request: Request) -> dict[str, Any]: + """Whether this installation is enrolled, and whether the link is up. + + Readable by any signed-in user: everyone here has a right to know whether + the machine they are using can be reached from outside. + """ + connector = getattr(request.app.state, "cloud_connector", None) + if connector is None: + config = cloud_config.load() + return { + "enrolled": config is not None, + "connected": False, + "portal_url": config.portal_url if config else None, + "portal_account": config.portal_account if config else None, + "installation_id": config.installation_id if config else None, + "last_error": None, + "connected_since": None, + } + status: dict[str, Any] = connector.status() + return status + + +@router.post( + "/enroll", + dependencies=[Depends(get_current_active_superuser)], + response_model=Message, +) +async def enroll( + request: Request, current_user: CurrentUser, body: EnrollBody +) -> Message: + """Redeem a claim code and start dialling the portal. + + The account performing this is recorded as the one every portal session + will act as. There is no way to widen that later from the portal side. + """ + if cloud_config.exists(): + raise HTTPException( + status_code=409, + detail="This installation is already connected to a portal", + ) + + base = body.portal_url.rstrip("/") + try: + async with httpx.AsyncClient(timeout=15.0) as client: + response = await client.post( + f"{base}/api/v1/enroll/", + json={"claim_code": body.claim_code, "app_version": "0.1.0"}, + ) + except httpx.HTTPError as exc: + raise HTTPException( + status_code=502, detail=f"Could not reach the portal: {exc}" + ) from exc + if response.status_code == 404: + raise HTTPException( + status_code=400, detail="That claim code is unknown or has expired" + ) + if response.status_code != 200: + raise HTTPException( + status_code=502, + detail=f"The portal refused the claim ({response.status_code})", + ) + + data = response.json() + config = cloud_config.CloudConfig( + portal_url=base, + ws_url=data["ws_url"], + installation_id=data["installation_id"], + token=data["installation_token"], + issuer=data["issuer"], + # Pinned here, at the one moment the claim code proves who we are + # talking to. Nothing refreshes this. + jwks=data["jwks"], + local_user_id=str(current_user.id), + enrolled_at=datetime.now(UTC).isoformat(), + portal_account=current_user.email, + ) + cloud_config.save(config) + _start_connector(request.app) + return Message(message="Connected to the portal") + + +@router.delete( + "", dependencies=[Depends(get_current_active_superuser)], response_model=Message +) +def disconnect(request: Request) -> Message: + """Sever the connection from this side. + + Unilateral and immediate: the config is the only reason portal tokens + verify here, so deleting it ends remote access whatever the portal still + has on file. + """ + task = getattr(request.app.state, "cloud_task", None) + if task is not None: + task.cancel() + request.app.state.cloud_task = None + request.app.state.cloud_connector = None + cloud_config.delete() + return Message(message="Disconnected from the portal") + + +def _start_connector(app: Any) -> None: + from app.cloud.connector import CloudConnector + + existing = getattr(app.state, "cloud_task", None) + if existing is not None: + existing.cancel() + connector = CloudConnector(app) + app.state.cloud_connector = connector + app.state.cloud_task = asyncio.create_task( + connector.serve_forever(), name="cloud-connector" + ) diff --git a/backend/app/cloud/__init__.py b/backend/app/cloud/__init__.py new file mode 100644 index 0000000..efefeef --- /dev/null +++ b/backend/app/cloud/__init__.py @@ -0,0 +1,6 @@ +"""Optional remote access through a Fluksio portal. + +Nothing in here runs unless someone enrolled this installation: with no +``cloud.json`` the connector never starts and the portal's tokens are refused +like any other bad credential. Turning it off is deleting that one file. +""" diff --git a/backend/app/cloud/config.py b/backend/app/cloud/config.py new file mode 100644 index 0000000..a21a940 --- /dev/null +++ b/backend/app/cloud/config.py @@ -0,0 +1,128 @@ +"""What this installation knows about the portal it is enrolled with. + +One file on the data volume, beside the OAuth keypair and for the same reason: +it is a credential, it must survive a rebuild, and it must never be in the +flows repository. Deleting it is the local, unilateral way to sever the +connection — the portal's tokens stop verifying immediately, whatever the +portal still believes. + +The portal's public keys are stored here rather than fetched: they are pinned +at enrolment, when a person was holding a claim code they had just read off the +portal's own screen. A hub whose DNS or TLS is later hijacked cannot re-key an +installation that already enrolled; it can only fail to verify. +""" + +from __future__ import annotations + +import json +import logging +import os +from dataclasses import dataclass +from typing import Any + +import jwt +from jwt.exceptions import InvalidTokenError + +from app.core.config import settings + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class CloudConfig: + portal_url: str + ws_url: str + installation_id: str + token: str + issuer: str + jwks: dict[str, Any] + #: The local account every portal session acts as. Recorded at enrolment + #: from whoever performed it, so remote access can never exceed the rights + #: of the person who granted it. + local_user_id: str + enrolled_at: str + portal_account: str | None = None + + +_cache: tuple[float, CloudConfig | None] | None = None + + +def _read() -> CloudConfig | None: + """Load the config, re-reading only when the file has changed.""" + global _cache + path = settings.CLOUD_CONFIG_FILE + try: + mtime = path.stat().st_mtime + except OSError: + _cache = (0.0, None) + return None + if _cache is not None and _cache[0] == mtime: + return _cache[1] + try: + data = json.loads(path.read_text()) + config = CloudConfig(**data) + except (OSError, ValueError, TypeError): + logger.exception("Could not read %s; remote access stays off", path) + _cache = (mtime, None) + return None + _cache = (mtime, config) + return config + + +def load() -> CloudConfig | None: + return _read() + + +def exists() -> bool: + return settings.CLOUD_CONFIG_FILE.exists() + + +def save(config: CloudConfig) -> None: + path = settings.CLOUD_CONFIG_FILE + path.parent.mkdir(parents=True, exist_ok=True) + # Written through a temporary file so a crash mid-write cannot leave a + # half-parsed credential behind, and never group- or world-readable. + tmp = path.with_suffix(".tmp") + tmp.write_text(json.dumps(config.__dict__, indent=2)) + os.chmod(tmp, 0o600) + tmp.replace(path) + global _cache + _cache = None + + +def delete() -> None: + settings.CLOUD_CONFIG_FILE.unlink(missing_ok=True) + global _cache + _cache = None + + +def decode_portal_token(token: str) -> dict[str, Any]: + """Validate a token the portal minted for this installation. + + Raises ``InvalidTokenError`` for everything else, including the ordinary + case of not being enrolled at all — which is what makes this branch free + for an installation nobody connected. + """ + config = _read() + if config is None: + raise InvalidTokenError("this installation is not enrolled with a portal") + + try: + key = jwt.PyJWKSet.from_dict(config.jwks).keys[0] + except (IndexError, jwt.PyJWKError) as exc: + raise InvalidTokenError(f"the pinned portal key is unusable: {exc}") from exc + + claims: dict[str, Any] = jwt.decode( + token, + key, + algorithms=["RS256"], + # The audience is this installation's own id, so a token the portal + # minted for somebody else's machine fails here. + audience=config.installation_id, + issuer=config.issuer, + ) + if claims.get("scope") != "proxy": + raise InvalidTokenError("not a proxy token") + # Every portal session acts as the enrolling local user. Who they are on + # the portal is kept for the audit trail, not for authorization. + return {"sub": config.local_user_id, "portal_sub": claims.get("sub")} diff --git a/backend/app/cloud/connector.py b/backend/app/cloud/connector.py new file mode 100644 index 0000000..c2f7253 --- /dev/null +++ b/backend/app/cloud/connector.py @@ -0,0 +1,376 @@ +"""The link this installation opens to its portal. + +Same shape as the remote-worker agent, with the direction of the favour +reversed: there, a GPU box dials the engine so the engine can give it work; +here, the engine dials the portal so the portal can pass on what a browser +asked for. Both exist because the interesting machine is the one nobody can +route to. + +A proxied request is dispatched straight into this process's own ASGI app +rather than over localhost HTTP. That matters beyond saving a hop: HTTP +trigger nodes install themselves into ``app.routes`` while flows run, and an +in-process transport sees that table live, with no port, TLS or self-address +to discover. +""" + +from __future__ import annotations + +import asyncio +import base64 +import contextlib +import logging +import time +from datetime import timedelta +from typing import Any + +import httpx +from fastapi import FastAPI + +from app.cloud import config as cloud_config +from app.core.config import settings + +logger = logging.getLogger(__name__) + +PROTOCOL = 1 +HEARTBEAT_S = 20.0 +STATUS_S = 60.0 +MAX_BACKOFF_S = 30.0 +CHUNK_BYTES = 256 * 1024 +MAX_WS_MESSAGE = 1024 * 1024 +#: Only the versioned API is served over the tunnel. The MCP mount and the +#: OAuth endpoints live outside it and stay local-only. +ALLOWED_PREFIX = "/api/v1/" + + +class CloudConnector: + """Holds one websocket to the portal open, and answers over it.""" + + def __init__(self, app: FastAPI) -> None: + self._app = app + self._calls: dict[str, asyncio.Task[None]] = {} + self._streams: dict[str, asyncio.Task[None]] = {} + self._connected = False + self._connected_since: float | None = None + self._last_error: str | None = None + + # ------------------------------------------------------------------------- + # Status, for the local settings screen + # ------------------------------------------------------------------------- + + def status(self) -> dict[str, Any]: + config = cloud_config.load() + return { + "enrolled": config is not None, + "connected": self._connected, + "portal_url": config.portal_url if config else None, + "portal_account": config.portal_account if config else None, + "installation_id": config.installation_id if config else None, + "last_error": self._last_error, + "connected_since": self._connected_since, + } + + # ------------------------------------------------------------------------- + # The connection + # ------------------------------------------------------------------------- + + async def serve_forever(self) -> None: + backoff = 1.0 + while True: + config = cloud_config.load() + if config is None: + # Disconnected locally while we were running. + return + try: + await self._session(config) + backoff = 1.0 + except asyncio.CancelledError: + raise + except Exception as exc: + self._connected = False + self._connected_since = None + self._last_error = str(exc) + logger.warning("Portal link down: %s — retrying in %.0fs", exc, backoff) + await asyncio.sleep(backoff) + backoff = min(MAX_BACKOFF_S, backoff * 2) + + async def _session(self, config: cloud_config.CloudConfig) -> None: + import websockets + + url = f"{config.ws_url}?token={config.token}" + async with websockets.connect( + url, max_size=MAX_WS_MESSAGE, ping_interval=20 + ) as socket: + await socket.send( + _dump({"op": "hello", "protocol": PROTOCOL, "app_version": _version()}) + ) + welcome = _load(await socket.recv()) + if welcome.get("op") != "welcome": + raise RuntimeError( + str(welcome.get("reason") or "refused by the portal") + ) + + self._connected = True + self._connected_since = time.time() + self._last_error = None + logger.info( + "Attached to the portal as installation %s", config.installation_id + ) + + keepalive = asyncio.create_task(self._keepalive(socket, config)) + try: + async for raw in socket: + frame = _load(raw) + op = frame.get("op") + if op == "req": + self._start_call(socket, frame) + elif op == "cancel": + self._cancel(str(frame.get("id") or "")) + elif op == "ws_open": + self._start_stream(socket, frame) + elif op == "ws_close": + self._cancel_stream(str(frame.get("id") or "")) + finally: + keepalive.cancel() + self._connected = False + self._connected_since = None + for task in list(self._calls.values()) + list(self._streams.values()): + task.cancel() + self._calls.clear() + self._streams.clear() + + async def _keepalive(self, socket: Any, config: cloud_config.CloudConfig) -> None: + """Heartbeats, plus a health snapshot the portal can show while offline.""" + last_status = 0.0 + while True: + now = time.time() + if now - last_status >= STATUS_S or last_status == 0.0: + summary = await self._health_summary(config) + await socket.send( + _dump( + {"op": "status", "summary": summary, "app_version": _version()} + ) + ) + last_status = now + else: + await socket.send(_dump({"op": "heartbeat"})) + await asyncio.sleep(HEARTBEAT_S) + + async def _health_summary( + self, config: cloud_config.CloudConfig + ) -> dict[str, Any] | None: + """This installation's own health, as the dashboard reads it. + + Fetched through the same in-process transport as everything else, with + a short-lived local token: the observability API is authenticated, and + this process is entitled to mint one for the enrolling user. + """ + from app.core import security + + token = security.create_access_token(config.local_user_id, timedelta(minutes=5)) + try: + async with self._client() as client: + response = await client.get( + "/api/v1/observability/summary", + headers={"Authorization": f"Bearer {token}"}, + ) + if response.status_code == 200: + summary: dict[str, Any] = response.json() + return summary + except Exception: + logger.debug("Could not collect a health summary for the portal") + return None + + # ------------------------------------------------------------------------- + # Proxied requests + # ------------------------------------------------------------------------- + + def _client(self) -> httpx.AsyncClient: + return httpx.AsyncClient( + transport=httpx.ASGITransport(app=self._app), + base_url="http://installation.local", + timeout=httpx.Timeout(300.0, connect=5.0), + ) + + def _start_call(self, socket: Any, frame: dict[str, Any]) -> None: + call_id = str(frame.get("id") or "") + if not call_id: + return + task = asyncio.create_task(self._serve(socket, frame)) + self._calls[call_id] = task + task.add_done_callback(lambda _t: self._calls.pop(call_id, None)) + + async def _serve(self, socket: Any, frame: dict[str, Any]) -> None: + call_id = str(frame["id"]) + path = str(frame.get("path") or "") + try: + if not path.startswith(ALLOWED_PREFIX): + # Defence in depth: the hub only ever forwards API paths, and + # an installation that trusted anything else would be an SSRF + # gadget aimed at its own process. + await socket.send( + _dump({"op": "err", "id": call_id, "message": "path not allowed"}) + ) + return + body = frame.get("body") + content = base64.b64decode(body) if body else None + query = str(frame.get("query") or "") + + async with self._client() as client: + request = client.build_request( + str(frame.get("method") or "GET"), + f"{path}?{query}" if query else path, + headers=dict(frame.get("headers") or {}), + content=content, + ) + response = await client.send(request, stream=True) + try: + await socket.send( + _dump( + { + "op": "res", + "id": call_id, + "status": response.status_code, + "headers": dict(response.headers), + } + ) + ) + async for chunk in response.aiter_bytes(CHUNK_BYTES): + await socket.send( + _dump( + { + "op": "res_body", + "id": call_id, + "body": base64.b64encode(chunk).decode("ascii"), + "done": False, + } + ) + ) + await socket.send( + _dump({"op": "res_body", "id": call_id, "done": True}) + ) + finally: + await response.aclose() + except asyncio.CancelledError: + raise + except Exception as exc: + logger.warning("Proxied %s failed: %s", path, exc) + with contextlib.suppress(Exception): + await socket.send( + _dump({"op": "err", "id": call_id, "message": str(exc)}) + ) + + def _cancel(self, call_id: str) -> None: + task = self._calls.pop(call_id, None) + if task is not None: + task.cancel() + + # ------------------------------------------------------------------------- + # Bridged live stream + # ------------------------------------------------------------------------- + + def _start_stream(self, socket: Any, frame: dict[str, Any]) -> None: + stream_id = str(frame.get("id") or "") + if not stream_id: + return + task = asyncio.create_task(self._stream(socket, frame)) + self._streams[stream_id] = task + task.add_done_callback(lambda _t: self._streams.pop(stream_id, None)) + + async def _stream(self, socket: Any, frame: dict[str, Any]) -> None: + """Feed the engine's live events down the tunnel. + + The flows websocket is subscribed to directly rather than dialled: + httpx cannot speak websocket, and everything that endpoint does — check + the token, send a snapshot, forward the bus — is a few lines against + objects this process already holds. + """ + from sqlmodel import Session + + from app.api.deps import user_from_token + from app.core.db import engine + from app.flow.events import event_bus + + stream_id = str(frame["id"]) + path = str(frame.get("path") or "") + query = str(frame.get("query") or "") + token = _query_value(query, "token") + + if not path.startswith(ALLOWED_PREFIX) or not path.endswith("/flows/ws"): + await socket.send(_dump({"op": "ws_close", "id": stream_id, "code": 1008})) + return + with Session(engine) as session: + if user_from_token(session, token) is None: + await socket.send( + _dump({"op": "ws_close", "id": stream_id, "code": 1008}) + ) + return + + await socket.send(_dump({"op": "ws_open_ok", "id": stream_id})) + + controller = getattr(self._app.state, "flow_controller", None) + if controller is not None: + await socket.send( + _dump( + { + "op": "ws_msg", + "id": stream_id, + "text": _dump( + { + "type": "snapshot", + "values": controller.values(), + "nodes": [ + s.model_dump() for s in controller.node_statuses() + ], + "issues": [i.model_dump() for i in controller.issues], + "paused": controller.paused_flows(), + "logs": list(event_bus.recent_logs), + } + ), + } + ) + ) + + try: + async with event_bus.subscribe() as queue: + while True: + event = await queue.get() + await socket.send( + _dump({"op": "ws_msg", "id": stream_id, "text": _dump(event)}) + ) + except asyncio.CancelledError: + raise + except Exception as exc: + logger.debug("Live stream %s ended: %s", stream_id, exc) + with contextlib.suppress(Exception): + await socket.send( + _dump({"op": "ws_close", "id": stream_id, "code": 1011}) + ) + + def _cancel_stream(self, stream_id: str) -> None: + task = self._streams.pop(stream_id, None) + if task is not None: + task.cancel() + + +def _dump(payload: Any) -> str: + import json + + return json.dumps(payload, default=str) + + +def _load(raw: Any) -> dict[str, Any]: + import json + + data: dict[str, Any] = json.loads(raw) + return data + + +def _query_value(query: str, key: str) -> str: + from urllib.parse import parse_qs + + values = parse_qs(query).get(key) or [""] + return values[0] + + +def _version() -> str: + return settings.VERSION if hasattr(settings, "VERSION") else "0.1.0" diff --git a/backend/app/core/config.py b/backend/app/core/config.py index e653012..b093a86 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -53,6 +53,9 @@ class Settings(BaseSettings): PRIVATE_API_ENABLED: bool = False DOMAIN: str = "localhost" OAUTH_PRIVATE_KEY_FILE: Path = Path("flow-data/oauth-key.pem") + # Written only when someone enrols this installation with a portal. + # Its absence is what keeps remote access off. + CLOUD_CONFIG_FILE: Path = Path("flow-data/cloud.json") OAUTH_CODE_EXPIRE_SECONDS: int = 60 # Short, because an agent's token is a bearer secret held by a program # rather than a person, and it can refresh unattended. diff --git a/backend/app/main.py b/backend/app/main.py index 51cb3b8..3e22667 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -12,6 +12,7 @@ from starlette.middleware.cors import CORSMiddleware from app.api.main import api_router from app.api.routes.alerts import read_config as read_alerts_config +from app.cloud import config as cloud_config from app.core import security from app.core.config import settings from app.flow import logs, modules @@ -160,6 +161,20 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: ) await controller.start() run_service.start() + # Optional, and off unless someone enrolled this installation: the + # connector dials the portal, nothing dials in. + cloud_task: asyncio.Task[None] | None = None + app.state.cloud_connector = None + app.state.cloud_task = None + if cloud_config.exists(): + from app.cloud.connector import CloudConnector + + connector = CloudConnector(app) + app.state.cloud_connector = connector + cloud_task = asyncio.create_task( + connector.serve_forever(), name="cloud-connector" + ) + app.state.cloud_task = cloud_task try: # A mounted sub-app gets no lifespan of its own, so the MCP session # manager is entered here; without it every /mcp request fails. @@ -169,6 +184,10 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: watchdog_task.cancel() alerts_task.cancel() metrics_task.cancel() + # Re-read from app.state: enrolling at runtime replaces this. + running_cloud = getattr(app.state, "cloud_task", None) or cloud_task + if running_cloud is not None: + running_cloud.cancel() await run_in_threadpool(run_service.stop) await controller.stop() pool.stop() diff --git a/backend/tests/test_cloud.py b/backend/tests/test_cloud.py new file mode 100644 index 0000000..b402993 --- /dev/null +++ b/backend/tests/test_cloud.py @@ -0,0 +1,182 @@ +"""Remote access: off by default, and scoped to the account that enabled it. + +The property worth pinning down is the one everything else rests on — a +portal's token is worth nothing here until somebody at this installation +enrolled it, and even then it grants exactly the rights of the account that +did. +""" + +from __future__ import annotations + +import json +import uuid +from datetime import UTC, datetime, timedelta +from typing import Any + +import jwt +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from fastapi.testclient import TestClient +from sqlmodel import Session, select + +from app.api.deps import decode_token, user_from_token +from app.cloud import config as cloud_config +from app.core.config import settings +from app.models import User + +INSTALLATION_ID = "6f1c2d3e-4a5b-4c6d-8e9f-0a1b2c3d4e5f" +ISSUER = "https://hub.example.test" + + +@pytest.fixture +def portal_key() -> rsa.RSAPrivateKey: + return rsa.generate_private_key(public_exponent=65537, key_size=2048) + + +def _jwks(key: rsa.RSAPrivateKey) -> dict[str, Any]: + jwk = json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(key.public_key())) + jwk.update({"use": "sig", "alg": "RS256", "kid": "test-portal"}) + return {"keys": [jwk]} + + +def _portal_token( + key: rsa.RSAPrivateKey, + *, + subject: str = "portal-user-1", + audience: str = INSTALLATION_ID, + issuer: str = ISSUER, + scope: str = "proxy", +) -> str: + now = datetime.now(UTC) + pem = key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + return jwt.encode( + { + "sub": subject, + "iss": issuer, + "aud": audience, + "iat": now, + "exp": now + timedelta(hours=1), + "scope": scope, + }, + pem, + algorithm="RS256", + headers={"kid": "test-portal"}, + ) + + +@pytest.fixture +def enrolled( + tmp_path_factory: pytest.TempPathFactory, + portal_key: rsa.RSAPrivateKey, + db: Session, +) -> Any: + """Enrol this installation with a fake portal, then undo it.""" + local_user = db.exec( + select(User).where(User.email == settings.FIRST_SUPERUSER) + ).one() + original = settings.CLOUD_CONFIG_FILE + settings.CLOUD_CONFIG_FILE = ( + tmp_path_factory.mktemp(f"cloud-{uuid.uuid4().hex[:6]}") / "cloud.json" + ) + cloud_config.save( + cloud_config.CloudConfig( + portal_url=ISSUER, + ws_url=f"{ISSUER}/api/v1/tunnel/attach", + installation_id=INSTALLATION_ID, + token="installation-token", + issuer=ISSUER, + jwks=_jwks(portal_key), + local_user_id=str(local_user.id), + enrolled_at=datetime.now(UTC).isoformat(), + portal_account=settings.FIRST_SUPERUSER, + ) + ) + yield local_user + cloud_config.delete() + settings.CLOUD_CONFIG_FILE = original + + +def test_portal_token_is_refused_when_not_enrolled( + portal_key: rsa.RSAPrivateKey, tmp_path_factory: pytest.TempPathFactory +) -> None: + """An installation nobody connected trusts no portal at all.""" + original = settings.CLOUD_CONFIG_FILE + settings.CLOUD_CONFIG_FILE = tmp_path_factory.mktemp("empty") / "cloud.json" + try: + with pytest.raises(jwt.exceptions.InvalidTokenError): + decode_token(_portal_token(portal_key)) + finally: + settings.CLOUD_CONFIG_FILE = original + + +def test_portal_token_resolves_to_the_enrolling_user( + enrolled: User, portal_key: rsa.RSAPrivateKey, db: Session +) -> None: + claims = decode_token(_portal_token(portal_key)) + assert claims["sub"] == str(enrolled.id) + # Who they are on the portal is carried for the record, not for rights. + assert claims["portal_sub"] == "portal-user-1" + assert user_from_token(db, _portal_token(portal_key)) == enrolled + + +def test_token_for_another_installation_is_refused( + enrolled: User, # noqa: ARG001 (fixture installs the enrolment) + portal_key: rsa.RSAPrivateKey, +) -> None: + """The audience is this installation's id, so someone else's is worthless.""" + with pytest.raises(jwt.exceptions.InvalidTokenError): + decode_token(_portal_token(portal_key, audience=str(uuid.uuid4()))) + + +def test_token_from_an_unpinned_key_is_refused( + enrolled: User, # noqa: ARG001 (fixture installs the enrolment) +) -> None: + """A different portal, or a hijacked one, cannot sign for this installation.""" + impostor = rsa.generate_private_key(public_exponent=65537, key_size=2048) + with pytest.raises(jwt.exceptions.InvalidTokenError): + decode_token(_portal_token(impostor)) + + +def test_non_proxy_scope_is_refused( + enrolled: User, # noqa: ARG001 (fixture installs the enrolment) + portal_key: rsa.RSAPrivateKey, +) -> None: + with pytest.raises(jwt.exceptions.InvalidTokenError): + decode_token(_portal_token(portal_key, scope="session")) + + +def test_disconnecting_ends_remote_access( + enrolled: User, portal_key: rsa.RSAPrivateKey +) -> None: + """Deleting the config is the whole of the local revocation.""" + assert decode_token(_portal_token(portal_key))["sub"] == str(enrolled.id) + cloud_config.delete() + with pytest.raises(jwt.exceptions.InvalidTokenError): + decode_token(_portal_token(portal_key)) + + +def test_status_reports_not_enrolled( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + response = client.get( + f"{settings.API_V1_STR}/cloud/status", headers=superuser_token_headers + ) + assert response.status_code == 200 + assert response.json()["enrolled"] is False + + +def test_enrolling_needs_a_superuser( + client: TestClient, normal_user_token_headers: dict[str, str] +) -> None: + """Remote access is an installation-wide grant, not a personal setting.""" + response = client.post( + f"{settings.API_V1_STR}/cloud/enroll", + headers=normal_user_token_headers, + json={"portal_url": ISSUER, "claim_code": "ABCD-EFGH"}, + ) + assert response.status_code == 403