Optional remote access: dial out to a Fluksio portal
An installation can be enrolled with a portal by redeeming a claim code, after which it holds one authenticated websocket open and answers proxied API calls over it. Requests are dispatched into this process's own ASGI app, so the HTTP trigger routes flows install at runtime are visible to it, and the live flow stream is bridged straight off the event bus. decode_token grows the third branch its docstring anticipated: tokens signed by the enrolled portal resolve to the local account that performed the enrolment, verified against a JWKS pinned at that moment. With no enrolment the branch raises immediately, so an offline installation is unchanged and untouched. Disconnecting deletes one file, which is the entire local revocation. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XtBzdDyLsmDaF1W7DLYtYM
This commit is contained in:
Claude Opus 5
@@ -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.
|
||||
"""
|
||||
@@ -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")}
|
||||
@@ -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"
|
||||
Reference in New Issue
Block a user