nicolas-georger/deepseek-kubernetes-sandbox
1
0
Fork 0
mirror of https://github.com/sredevopsorg/deepseek-kubernetes-sandbox.git synced 2026-11-03 09:23:25 +00:00
Code Projects Releases Activity

feat(repo)!: repo reorganization (#16)

Browse Source 
* feat(repo)!: restructure around the dsh-harness image and one example per implementation

- the root Dockerfile is the primary artifact; versions.env is the single source of truth
- merge gateway-api/dsh-demo into examples/dsh-agents-as-a-service (manifests/, exposure variants)
- move the single-tenant template into examples/minimal-single-tenant
- publish the gateway as a nested package under the repository image
- mirror the DSH ref for release identities (first release dsh-v0.2.1-rc.2-r2)
- CI: lint/python/manifests/gateway-image, reusable multi-arch build, tag-driven release, weekly e2e
- docs: README, docs/{main-image,examples,gitflow,versioning}, example READMEs, AGENTS.md, governance
- SSO is opt-in again; gateway pull policy and cookie defaults match the documented demo

BREAKING CHANGE: example manifest paths, the gateway image name, and the removal
of deploy/ and gateway-api/ affect existing deployments; see the migration table
in README.md.

* docs(changelog): generate initial changelog from history

* fix(scripts): make set-version update Dockerfile and sample revision pins

* build(hadolint): document intentional lint exceptions for the main image

---------

Co-authored-by: SREDevOpsAgents <agents@sredevops.org>
This commit is contained in:
Nicolás Georger
2026-10-25 08:37:43 -03:00
committed by GitHub
parent 53e55fc851
commit 3b70f179ac
69 changed files with 3774 additions and 1643 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
examples/dsh-agents-as-a-service/gateway/.dockerignore Normal file
View File

@@ -0,0 +1,11 @@
# Runtime image only needs the app and its pinned runtime requirements.
__pycache__/
*.py[cod]
.pytest_cache/
.ruff_cache/
.venv/
test_gateway.py
requirements-dev.txt
pyproject.toml
README.md
coverage.xml

66
examples/dsh-agents-as-a-service/gateway/README.md
View File

@@ -4,7 +4,7 @@ A single-file, browser-usable control + data plane for
[DeepSeek Harness Web (DSH)](https://github.com/deepseek-ai/deepseek-harness)
sandboxes running behind the `agent-sandbox` CRDs.
It does three things:
It does four things:
1. **Users** — creates a `SandboxClaim` from a warm pool, mints a bearer token
(only its SHA-256 is stored as a claim annotation), reports derived state,
@@ -15,10 +15,15 @@ It does three things:
3. **Reverse proxy** — proxies HTTP (including SSE) and WebSocket traffic to
the sandbox's DSH Web server on `:3080`, including DSH's own cookie
handshake.
4. **Optional SSO** — when `AUTH_PROXY_HEADER` is set, trusts the identity
header an authenticating reverse proxy (for example oauth2-proxy) forwards,
auto-provisions that user, and exposes `POST /recreate` for a clean restart.
DSH is pinned to `dsh-v0.2.1-rc.2` in the example manifests. It binds `:3080`,
serves `/`, gates `/api` behind a `dsh-auth-*` cookie and requires a WebSocket
at `/api/remote.mux`.
The DSH version pin lives in [`versions.env`](../../../versions.env)
(`DSH_VERSION` plus the optional repository `RELEASE_REVISION`); the example
manifests and published images use the release identity built from it (currently
`dsh-v0.2.1-rc.2-r2`). DSH binds `:3080`, serves `/`, gates `/api` behind a
`dsh-auth-*` cookie and requires a WebSocket at `/api/remote.mux`.
## Flow
@@ -58,10 +63,11 @@ token. A failed handshake returns `503` with `Retry-After: 10`.
| Method | Path | Auth | Purpose |
| --- | --- | --- | --- |
| `GET` | `/healthz` | none | Liveness/readiness. Never touches the cluster. |
| `GET` | `/` | none | Landing page with signup + token login forms. |
| `GET` | `/` | none | Landing page with signup + token login forms. With SSO it auto-provisions the proxy-identified user and redirects to their workspace. |
| `POST` | `/users` | none | Signup. JSON body `{"user": "<dns-1123 label>"}` or form field `user`. JSON returns `201` when Ready, `202` when accepted but still Waking; form returns `303` to `/u/<user>/`. Both set the session cookie. `409` if the user exists, `400` if invalid, `413` if the body is too large. |
| `POST` | `/login` | none | Form (`user`, `token`) returns `303` to `/u/<user>/`; JSON returns `200`. Sets the session cookie. `401` on a bad token, `404` for an unknown user. |
| `POST` | `/logout` | none | Clears the session cookie and `303`s to `/`. |
| `POST` | `/recreate` | trusted proxy header | **SSO only.** Deletes the authenticated user's claim (cascading sandbox + PVC, so workspace data is wiped), waits out finalizers, provisions a fresh claim and `303`s to `/u/<user>/`. `401` without a valid identity header; cookie-authenticated calls must be same-origin. |
| `GET` | `/users/{user}` | bearer or cookie | Derived state: `Provisioning`, `Waking`, `Ready`, `Suspending`, `Suspended`. |
| `DELETE` | `/users/{user}` | bearer or cookie | Delete the claim (sandbox + PVC cascade). Cookie-auth mutations require a same-origin `Origin`/`Referer`. |
| `GET` | `/u/{user}` | — | `308` redirect to `/u/{user}/`. |
@@ -86,6 +92,37 @@ token. A failed handshake returns `503` with `Retry-After: 10`.
absolute paths are rewritten under `/u/<user>/`, and any `token` query
parameter is stripped.
### SSO (opt-in)
By default the gateway is the only authenticator: users sign up with a token or
the session cookie, and the gateway ignores all identity headers. Set
`AUTH_PROXY_HEADER` to enable SSO behind an authenticating reverse proxy such as
oauth2-proxy:
```bash
kubectl -n dsh-demo set env deploy/dsh-gateway AUTH_PROXY_HEADER=X-Forwarded-User
```
When set, the gateway treats a request carrying that header as authenticated as
that user, provided the value is a valid DNS-1123 label; the header is never
trusted when the variable is unset. The landing page then auto-provisions the
user (creates their claim idempotently, mints a rotated session token) and
redirects to `/u/<user>/`, and `POST /recreate` becomes available for a clean
restart. The base manifests leave it unset on purpose; the
[SSO exposure variant](../manifests/exposure/gateway-api-sso/) documents how to
enable it on the Deployment (`kubectl set env`).
> **Trust boundary — this is the security-critical part.** A trusted identity
> header is only safe when **nothing else can reach the gateway**. The gateway
> Service must stay ClusterIP and be reachable exclusively through the
> authenticating proxy; if a client can reach `dsh-gateway:8080` directly (a
> second Ingress/Gateway/Route, a port-forward on a shared host, a broad
> NetworkPolicy), it can send `X-Forwarded-User` itself and impersonate any user.
> Treat `AUTH_PROXY_HEADER` as "the proxy already authenticated this request",
> not as an authentication mechanism of its own. Disable SSO by removing the
> variable: the header is then ignored and bearer-token/session auth applies
> again.
### Proxy failure semantics
| Situation | Result |
@@ -123,6 +160,7 @@ can also be injected into `create_app` for tests.
| `MAX_BODY_MB` | `32` | Maximum buffered request body (and `413` threshold). |
| `HTTP_TIMEOUT` | `300` | Upstream connect/read timeout in seconds. Proxied streams (SSE, long responses) keep the connect/write budget but have **no read timeout**, so an idle event stream is not cut every five minutes. |
| `SWEEP_INTERVAL` | `15` | Seconds between idle sweeps. |
| `AUTH_PROXY_HEADER` | unset | Opt-in SSO: name of the trusted identity header set by an authenticating reverse proxy (e.g. `X-Forwarded-User`). Unset disables SSO and the header is never trusted. Enable only when the Service is reachable exclusively through that proxy (see [SSO](#sso-opt-in)). |
### Required RBAC
@@ -147,12 +185,21 @@ python -m ruff check .
uvicorn gateway:create_app --factory --host 0.0.0.0 --port 8080
```
Container image:
Container image, from the repository root:
```bash
docker build -t dsh-gateway examples/dsh-agents-as-a-service/gateway
make gateway # builds dsh-aaas-gateway:dev from this directory
```
The equivalent direct build is
`docker build -t dsh-aaas-gateway:dev examples/dsh-agents-as-a-service/gateway`.
For kind, retag it to the image reference in
`examples/dsh-agents-as-a-service/manifests/50-gateway.yaml` and `kind load` it.
Published images are nested below the main image as
`<GHCR_IMAGE>/dsh-agents-as-a-service/gateway:<identity>`, where the identity
comes from [`versions.env`](../../../versions.env) — see
[`docs/versioning.md`](../../../docs/versioning.md).
The image runs as UID `10000`, sets `PYTHONDONTWRITEBYTECODE=1` and
`HOME=/tmp` (read-only-rootfs compatible), listens on `8080`, and disables
uvicorn's access log so bearer tokens in URLs are not logged.
@@ -182,3 +229,8 @@ uvicorn's access log so bearer tokens in URLs are not logged.
re-runs the handshake once and replays the request.
* **Only DSH on `:3080` is proxied.** There is no dashboard/API-port split and
no platform API key injection.
* **SSO trusts a header.** With `AUTH_PROXY_HEADER` set, any client that can
reach the gateway Service directly can spoof the identity header. The gateway
cannot enforce the trust boundary itself: keep the Service ClusterIP and
reachable only through the authenticating proxy. There is no mTLS or signed
identity assertion.

147
examples/dsh-agents-as-a-service/gateway/gateway.py
View File

@@ -195,9 +195,7 @@ def derive_state(sandbox: dict | None) -> str:
def token_matches(claim: dict, token: str | None) -> bool:
"""Timing-safe check of a user token against the stored SHA-256."""
want = (claim.get("metadata", {}).get("annotations") or {}).get(
TOKEN_ANNOTATION, ""
)
want = (claim.get("metadata", {}).get("annotations") or {}).get(TOKEN_ANNOTATION, "")
if not token or not want:
return False
got = hashlib.sha256(token.encode()).hexdigest()
@@ -518,9 +516,7 @@ class Kubernetes:
async def get_pod(self, name: str) -> dict | None:
try:
pod = await self._call(
"core", "read_namespaced_pod", name, self.namespace
)
pod = await self._call("core", "read_namespaced_pod", name, self.namespace)
except ApiException as exc:
if exc.status == 404:
return None
@@ -530,9 +526,7 @@ class Kubernetes:
# so callers can keep using ``pod.get(...)``.
return k8s_client.ApiClient().sanitize_for_serialization(pod)
async def read_pod_logs(
self, name: str, container: str, tail_lines: int
) -> str:
async def read_pod_logs(self, name: str, container: str, tail_lines: int) -> str:
result = await self._call(
"core",
"read_namespaced_pod_log",
@@ -616,9 +610,7 @@ class Gateway:
return False # a session for another user never authorizes this one
return token_matches(claim, token)
async def _sso_session(
self, remote: str, headers, cookies, response: Response
) -> dict | None:
async def _sso_session(self, remote: str, headers, cookies, response: Response) -> dict | None:
"""Provision ``remote`` idempotently and mint/refresh a session token.
The authenticating proxy has already verified ``remote``, so it is
@@ -682,9 +674,7 @@ class Gateway:
while True:
async with self.state.lock:
if not self.state.maintenance.get(user):
self.state.in_flight[user] = (
self.state.in_flight.get(user, 0) + 1
)
self.state.in_flight[user] = self.state.in_flight.get(user, 0) + 1
self.state.last_activity[user] = time.time()
return True
if time.monotonic() > deadline:
@@ -693,9 +683,7 @@ class Gateway:
async def _release(self, user: str) -> None:
async with self.state.lock:
self.state.in_flight[user] = max(
self.state.in_flight.get(user, 1) - 1, 0
)
self.state.in_flight[user] = max(self.state.in_flight.get(user, 1) - 1, 0)
self.state.last_activity[user] = time.time()
# -- sandbox / handshake helpers --------------------------------------
@@ -724,9 +712,7 @@ class Gateway:
# Wake-on-connect: safe in ANY state, including mid-suspension.
self._clear_handshake(user)
try:
await self.k8s.patch_operating_mode(
sandbox["metadata"]["name"], "Running"
)
await self.k8s.patch_operating_mode(sandbox["metadata"]["name"], "Running")
except Exception as exc:
raise SandboxUnavailable("could not wake the sandbox") from exc
sandbox = await self.wait_ready(user, self.settings.wake_timeout)
@@ -740,10 +726,7 @@ class Gateway:
def _pod_name(self, sandbox: dict) -> str:
annotations = sandbox.get("metadata", {}).get("annotations") or {}
return (
annotations.get(POD_NAME_ANNOTATION)
or sandbox["metadata"]["name"]
)
return annotations.get(POD_NAME_ANNOTATION) or sandbox["metadata"]["name"]
@staticmethod
def _pod_ip(sandbox: dict) -> str | None:
@@ -787,16 +770,11 @@ class Gateway:
self.settings.dsh_log_tail_lines,
)
except ApiException as exc:
raise HandshakeError(
f"could not read logs for {pod_name}: {exc}"
) from exc
raise HandshakeError(f"could not read logs for {pod_name}: {exc}") from exc
launch_token = parse_dsh_launch_token(logs)
if not launch_token:
raise HandshakeError("DSH launch token not found in pod logs")
url = (
f"http://{fqdn}:{self.settings.dsh_port}/"
f"?token={quote(launch_token, safe='')}"
)
url = f"http://{fqdn}:{self.settings.dsh_port}/?token={quote(launch_token, safe='')}"
headers = {
"Host": host,
"X-Forwarded-Host": host,
@@ -812,9 +790,7 @@ class Gateway:
except httpx.HTTPError as exc:
raise HandshakeError(f"DSH handshake failed: {exc}") from exc
if response.status_code >= 400:
raise HandshakeError(
f"DSH handshake returned {response.status_code}"
)
raise HandshakeError(f"DSH handshake returned {response.status_code}")
cookie = extract_dsh_cookies(response.headers.get_list("set-cookie"))
if not cookie:
raise HandshakeError("DSH handshake set no dsh-auth cookie")
@@ -856,7 +832,7 @@ class Gateway:
else:
status = html.escape(state)
confirm = (
'onsubmit="return confirm(\'Delete this sandbox and start fresh? '
"onsubmit=\"return confirm('Delete this sandbox and start fresh? "
"Your workspace data will be wiped.')\""
)
body = (
@@ -887,9 +863,7 @@ class Gateway:
# session cookie minted by _sso_session; the real page is rendered
# once the sandbox state is known.
response = HTMLResponse()
claim = await self._sso_session(
remote, request.headers, request.cookies, response
)
claim = await self._sso_session(remote, request.headers, request.cookies, response)
sandbox = await self._sandbox_of(claim) if claim else None
page = HTMLResponse(self._sso_page(remote, derive_state(sandbox)))
if "set-cookie" in response.headers:
@@ -953,14 +927,10 @@ against the JSON API. The browser session is an HttpOnly cookie.</p>
@staticmethod
def _wants_form(request: Request) -> bool:
ctype = request.headers.get("content-type", "").lower()
return ctype.startswith(
("application/x-www-form-urlencoded", "multipart/form-data")
)
return ctype.startswith(("application/x-www-form-urlencoded", "multipart/form-data"))
async def _read_payload(self, request: Request) -> dict:
if request.headers.get("content-type", "").lower().startswith(
"application/json"
):
if request.headers.get("content-type", "").lower().startswith("application/json"):
data = await self._read_body(request)
try:
payload = json.loads(data)
@@ -997,9 +967,7 @@ against the JSON API. The browser session is an HttpOnly cookie.</p>
return self._error(413, "request body too large")
user = payload.get("user", "")
if not valid_user(user):
return self._error(
400, "body must be {'user': '<dns-1123 label>'}"
)
return self._error(400, "body must be {'user': '<dns-1123 label>'}")
if await self.k8s.get_claim(user) is not None:
return self._error(409, "user exists")
token = secrets.token_urlsafe(32)
@@ -1059,9 +1027,7 @@ against the JSON API. The browser session is an HttpOnly cookie.</p>
if claim is None:
return self._error(404, "not found")
token, cookie_user = self.credentials(request.headers, request.cookies)
if not self._authorized(
claim, token, cookie_user, user, self.remote_user(request.headers)
):
if not self._authorized(claim, token, cookie_user, user, self.remote_user(request.headers)):
return self._error(401, "unauthorized")
sandbox = await self._sandbox_of(claim)
return JSONResponse(
@@ -1080,9 +1046,7 @@ against the JSON API. The browser session is an HttpOnly cookie.</p>
if claim is None:
return self._error(404, "not found")
token, cookie_user = self.credentials(request.headers, request.cookies)
if not self._authorized(
claim, token, cookie_user, user, self.remote_user(request.headers)
):
if not self._authorized(claim, token, cookie_user, user, self.remote_user(request.headers)):
return self._error(401, "unauthorized")
if self._mutation_from_cookie(request.headers, "DELETE", cookie_user):
return self._error(403, "cross-origin request rejected")
@@ -1091,9 +1055,7 @@ against the JSON API. The browser session is an HttpOnly cookie.</p>
async with self.state.lock:
self.state.last_activity.pop(user, None)
self.state.in_flight.pop(user, None)
return JSONResponse(
{"user": user, "note": "claim deleted; sandbox and PVC cascade"}
)
return JSONResponse({"user": user, "note": "claim deleted; sandbox and PVC cascade"})
async def recreate(self, request: Request) -> Response:
"""Delete and re-provision the SSO user's sandbox for a clean start.
@@ -1122,9 +1084,7 @@ against the JSON API. The browser session is an HttpOnly cookie.</p>
break
await asyncio.sleep(1)
response = RedirectResponse(f"/u/{remote}/", status_code=303)
await self._sso_session(
remote, request.headers, request.cookies, response
)
await self._sso_session(remote, request.headers, request.cookies, response)
return response
# -- HTTP proxy --------------------------------------------------------
@@ -1141,9 +1101,7 @@ against the JSON API. The browser session is an HttpOnly cookie.</p>
return RedirectResponse("/", status_code=303)
return self._error(404, "unknown user")
token, cookie_user = self.credentials(request.headers, request.cookies)
if remote == user and not (
cookie_user == user and token_matches(claim, token)
):
if remote == user and not (cookie_user == user and token_matches(claim, token)):
# SSO user without a valid session cookie: bootstrap, then continue.
return RedirectResponse("/", status_code=303)
if not self._authorized(claim, token, cookie_user, user, remote):
@@ -1152,31 +1110,21 @@ against the JSON API. The browser session is an HttpOnly cookie.</p>
return self._error(403, "cross-origin request rejected")
if not await self._acquire(user):
return self._error(
503, "gateway busy, retry shortly", {"Retry-After": "5"}
)
return self._error(503, "gateway busy, retry shortly", {"Retry-After": "5"})
streaming = False
try:
sandbox, fqdn = await self._resolve_sandbox(user, claim)
host = request.headers.get("host", "")
cookie, _, was_cached = await self.dsh_cookie(
user, sandbox, fqdn, host
)
cookie, _, was_cached = await self.dsh_cookie(user, sandbox, fqdn, host)
body = await self._read_body(request)
upstream = await self._send_upstream(
request, user, fqdn, cookie, body
)
upstream = await self._send_upstream(request, user, fqdn, cookie, body)
if upstream.status_code == 401 and was_cached:
# The cached DSH cookie expired (pod restart, logout, ...):
# handshake once more and replay the (already buffered) request.
await upstream.aclose()
cookie, _, _ = await self.dsh_cookie(
user, sandbox, fqdn, host, force=True
)
upstream = await self._send_upstream(
request, user, fqdn, cookie, body
)
cookie, _, _ = await self.dsh_cookie(user, sandbox, fqdn, host, force=True)
upstream = await self._send_upstream(request, user, fqdn, cookie, body)
response = StreamingResponse(
self._relay(user, upstream),
@@ -1216,8 +1164,7 @@ against the JSON API. The browser session is an HttpOnly cookie.</p>
headers = {
key: value
for key, value in request.headers.items()
if key.lower() not in OWNED_REQUEST_HEADERS
and key.lower() != "content-length"
if key.lower() not in OWNED_REQUEST_HEADERS and key.lower() != "content-length"
}
host = request.headers.get("host", "")
headers["host"] = host
@@ -1229,9 +1176,7 @@ against the JSON API. The browser session is an HttpOnly cookie.</p>
else:
headers["x-forwarded-for"] = ""
port = host.rsplit(":", 1)[1] if ":" in host else ""
headers["x-forwarded-port"] = port or (
"443" if request.url.scheme == "https" else "80"
)
headers["x-forwarded-port"] = port or ("443" if request.url.scheme == "https" else "80")
last_error: Exception | None = None
for attempt in range(self.settings.upstream_attempts):
@@ -1341,9 +1286,7 @@ against the JSON API. The browser session is an HttpOnly cookie.</p>
if ":" in upstream_host and not upstream_host.startswith("["):
upstream_host = f"[{upstream_host}]" # IPv6 literal
authority = f"{upstream_host}:{self.settings.dsh_port}"
cookie, _, _ = await self.dsh_cookie(
user, sandbox, fqdn, authority
)
cookie, _, _ = await self.dsh_cookie(user, sandbox, fqdn, authority)
url = f"ws://{authority}/{path}"
query = filtered_query(websocket.url.query)
if query:
@@ -1362,9 +1305,7 @@ against the JSON API. The browser session is an HttpOnly cookie.</p>
headers["x-forwarded-for"] = websocket.client.host
subprotocols = [
part.strip()
for part in websocket.headers.get(
"sec-websocket-protocol", ""
).split(",")
for part in websocket.headers.get("sec-websocket-protocol", "").split(",")
if part.strip()
]
upstream_cm = self.ws_connect(
@@ -1380,16 +1321,12 @@ against the JSON API. The browser session is an HttpOnly cookie.</p>
return await self._ws_reject(websocket, 1011)
try:
await websocket.accept(
subprotocol=getattr(upstream, "subprotocol", None)
)
await websocket.accept(subprotocol=getattr(upstream, "subprotocol", None))
except RuntimeError:
pass
await self._pump(websocket, upstream)
with contextlib.suppress(Exception):
await websocket.close(
code=self._upstream_close_code(upstream)
)
await websocket.close(code=self._upstream_close_code(upstream))
finally:
if upstream_cm is not None:
with contextlib.suppress(Exception):
@@ -1429,9 +1366,7 @@ against the JSON API. The browser session is an HttpOnly cookie.</p>
asyncio.create_task(client_to_upstream()),
asyncio.create_task(upstream_to_client()),
]
done, pending = await asyncio.wait(
tasks, return_when=asyncio.FIRST_COMPLETED
)
done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
for task in pending:
task.cancel()
await asyncio.gather(*done, *pending, return_exceptions=True)
@@ -1458,17 +1393,13 @@ against the JSON API. The browser session is an HttpOnly cookie.</p>
continue
if self.state.in_flight.get(user, 0) > 0:
continue # requests in flight => the user is active
idle = time.time() - self.state.last_activity.setdefault(
user, time.time()
)
idle = time.time() - self.state.last_activity.setdefault(user, time.time())
if idle < self.settings.idle_timeout:
continue
self.state.maintenance[user] = True
try:
print(f"suspending {user} (idle {int(idle)}s)", flush=True)
await self.k8s.patch_operating_mode(
sandbox["metadata"]["name"], "Suspended"
)
await self.k8s.patch_operating_mode(sandbox["metadata"]["name"], "Suspended")
self._clear_handshake(user)
finally:
async with self.state.lock:
@@ -1519,12 +1450,8 @@ def create_app(
k8s = k8s if k8s is not None else Kubernetes(settings.namespace)
owns_http_client = http_client is None
if http_client is None:
http_client = httpx.AsyncClient(
follow_redirects=False, timeout=settings.http_timeout
)
gateway = Gateway(
settings, k8s, http_client, ws_connect or websockets.connect
)
http_client = httpx.AsyncClient(follow_redirects=False, timeout=settings.http_timeout)
gateway = Gateway(settings, k8s, http_client, ws_connect or websockets.connect)
routes = [
Route("/healthz", gateway.healthz),
Route("/", gateway.landing),

1
examples/dsh-agents-as-a-service/gateway/requirements-dev.txt
View File

@@ -1,4 +1,5 @@
-r requirements.txt
pytest==9.1.1
pytest-asyncio==1.4.0
pytest-cov==7.1.0
ruff==0.17.1

140
examples/dsh-agents-as-a-service/gateway/test_gateway.py
View File

@@ -45,9 +45,7 @@ FQDN = "dsh-alice-sbx.test-ns.svc.cluster.local"
def dsh_response(status=200, content=b"", headers=None):
"""A MockTransport response that is a real stream (``aiter_raw`` works)."""
return httpx.Response(
status, headers=headers, stream=httpx.ByteStream(content)
)
return httpx.Response(status, headers=headers, stream=httpx.ByteStream(content))
# --- fakes -----------------------------------------------------------------
@@ -99,9 +97,7 @@ class FakeK8s:
self.token_patches.append((user, token_hash))
claim = self.claims.get(user)
if claim is not None:
annotations = claim.setdefault("metadata", {}).setdefault(
"annotations", {}
)
annotations = claim.setdefault("metadata", {}).setdefault("annotations", {})
annotations["aaas.example.com/token-sha256"] = token_hash
async def get_pod(self, name):
@@ -118,9 +114,7 @@ class FakeUpstream:
self.requests = []
self.handshake_requests = []
self.handshake_status = 303
self.handshake_cookies = [
"dsh-auth-a=one; Path=/; HttpOnly; SameSite=Strict"
]
self.handshake_cookies = ["dsh-auth-a=one; Path=/; HttpOnly; SameSite=Strict"]
self.proxy_handler = None
self.proxy_requests = []
@@ -140,9 +134,7 @@ class FakeUpstream:
self.proxy_requests.append(request)
if self.proxy_handler is not None:
return self.proxy_handler(request)
return dsh_response(
200, b"hello from DSH", {"content-type": "text/plain"}
)
return dsh_response(200, b"hello from DSH", {"content-type": "text/plain"})
class FakeWS:
@@ -207,9 +199,7 @@ def add_user(k8s, user="alice", token=TOKEN, ready=True, pod_name=None, suffix="
"metadata": {
"name": claim_name(user),
"annotations": {
"aaas.example.com/token-sha256": hashlib.sha256(
token.encode()
).hexdigest()
"aaas.example.com/token-sha256": hashlib.sha256(token.encode()).hexdigest()
},
},
"status": {"sandbox": {"name": sandbox_name}},
@@ -224,10 +214,7 @@ def add_user(k8s, user="alice", token=TOKEN, ready=True, pod_name=None, suffix="
},
}
k8s.pods[pod_name] = {"metadata": {"name": pod_name, "uid": POD_UID}}
k8s.logs[pod_name] = (
"booting\n"
f"dsh web: http://10.0.0.7:3080/?token={TOKEN}-launch\n"
)
k8s.logs[pod_name] = f"booting\ndsh web: http://10.0.0.7:3080/?token={TOKEN}-launch\n"
return sandbox_name, pod_name
@@ -243,9 +230,7 @@ def env():
)
k8s = FakeK8s()
upstream = FakeUpstream()
http_client = httpx.AsyncClient(
transport=httpx.MockTransport(upstream), timeout=5
)
http_client = httpx.AsyncClient(transport=httpx.MockTransport(upstream), timeout=5)
ws_connect = FakeWSConnect()
app = create_app(
settings=settings,
@@ -275,9 +260,7 @@ def asgi_client(env, cookies=None):
def stale_handshake(env):
env.gateway.state.handshake[("alice", POD_UID, "gateway.test")] = (
"dsh-auth-a=stale"
)
env.gateway.state.handshake[("alice", POD_UID, "gateway.test")] = "dsh-auth-a=stale"
# --- pure helpers ----------------------------------------------------------
@@ -419,9 +402,7 @@ def test_token_matches_is_timing_safe_hash_check():
claim = {
"metadata": {
"annotations": {
"aaas.example.com/token-sha256": hashlib.sha256(
TOKEN.encode()
).hexdigest()
"aaas.example.com/token-sha256": hashlib.sha256(TOKEN.encode()).hexdigest()
}
}
}
@@ -486,14 +467,9 @@ async def test_kubernetes_create_claim_body(monkeypatch):
body = args[4]
assert body["metadata"]["name"] == "dsh-alice"
assert body["metadata"]["annotations"]["aaas.example.com/token-sha256"] == "hash"
assert (
body["metadata"]["annotations"]["aaas.example.com/managed-by"]
== "dsh-gateway"
)
assert body["metadata"]["annotations"]["aaas.example.com/managed-by"] == "dsh-gateway"
assert body["spec"]["warmPoolRef"]["name"] == "test-pool"
assert body["spec"]["additionalPodMetadata"]["labels"][
"sandbox.users.io/dsh-user"
] == "alice"
assert body["spec"]["additionalPodMetadata"]["labels"]["sandbox.users.io/dsh-user"] == "alice"
@pytest.mark.asyncio
@@ -625,9 +601,7 @@ async def test_landing_page_has_signup_and_login_forms(env):
@pytest.mark.asyncio
async def test_landing_page_shows_session_when_cookie_present(env):
async with asgi_client(
env, cookies={"dsh_gw_session": f"alice:{TOKEN}"}
) as client:
async with asgi_client(env, cookies={"dsh_gw_session": f"alice:{TOKEN}"}) as client:
response = await client.get("/")
assert "Signed in as" in response.text
assert "/u/alice/" in response.text
@@ -645,9 +619,7 @@ async def test_workspace_redirect_adds_trailing_slash(env):
@pytest.mark.asyncio
@pytest.mark.parametrize(
"bad_user", ["Alice", "-alice", "alice-", "", "a_b", 123, None, "a" * 60]
)
@pytest.mark.parametrize("bad_user", ["Alice", "-alice", "alice-", "", "a_b", 123, None, "a" * 60])
async def test_signup_rejects_invalid_dns1123_label(env, bad_user):
async with asgi_client(env) as client:
response = await client.post("/users", json={"user": bad_user})
@@ -781,9 +753,7 @@ async def test_login_json_and_failures(env):
@pytest.mark.asyncio
async def test_logout_clears_cookie(env):
async with asgi_client(
env, cookies={"dsh_gw_session": f"alice:{TOKEN}"}
) as client:
async with asgi_client(env, cookies={"dsh_gw_session": f"alice:{TOKEN}"}) as client:
response = await client.post("/logout")
assert response.status_code == 303
assert response.headers["location"] == "/"
@@ -798,12 +768,8 @@ async def test_logout_clears_cookie(env):
async def test_get_user_requires_valid_bearer_or_cookie(env):
add_user(env.k8s)
async with asgi_client(env) as client:
forbidden = await client.get(
"/users/alice", headers={"Authorization": "Bearer wrong"}
)
bearer = await client.get(
"/users/alice", headers={"Authorization": f"Bearer {TOKEN}"}
)
forbidden = await client.get("/users/alice", headers={"Authorization": "Bearer wrong"})
bearer = await client.get("/users/alice", headers={"Authorization": f"Bearer {TOKEN}"})
cookie = await client.get(
"/users/alice",
headers={"Cookie": f"dsh_gw_session=alice:{TOKEN}"},
@@ -839,9 +805,7 @@ async def test_invalid_user_in_path_is_404(env, path):
async def test_delete_user_requires_auth(env):
add_user(env.k8s)
async with asgi_client(env) as client:
response = await client.delete(
"/users/alice", headers={"Authorization": "Bearer wrong"}
)
response = await client.delete("/users/alice", headers={"Authorization": "Bearer wrong"})
assert response.status_code == 401
assert env.k8s.deleted == []
@@ -853,12 +817,8 @@ async def test_delete_with_cookie_requires_same_origin(env):
cookies = {"dsh_gw_session": f"alice:{TOKEN}"}
async with asgi_client(env, cookies=cookies) as client:
missing = await client.delete("/users/alice")
cross = await client.delete(
"/users/alice", headers={"Origin": "http://evil.test"}
)
good = await client.delete(
"/users/alice", headers={"Origin": "http://gateway.test"}
)
cross = await client.delete("/users/alice", headers={"Origin": "http://evil.test"})
good = await client.delete("/users/alice", headers={"Origin": "http://gateway.test"})
assert missing.status_code == 403
assert cross.status_code == 403
assert good.status_code == 200
@@ -870,9 +830,7 @@ async def test_delete_with_cookie_requires_same_origin(env):
async def test_delete_with_bearer_does_not_need_origin(env):
add_user(env.k8s)
async with asgi_client(env) as client:
response = await client.delete(
"/users/alice", headers={"Authorization": f"Bearer {TOKEN}"}
)
response = await client.delete("/users/alice", headers={"Authorization": f"Bearer {TOKEN}"})
assert response.status_code == 200
@@ -881,9 +839,7 @@ async def test_delete_missing_claim_is_404(env):
add_user(env.k8s)
env.k8s.claims.clear()
async with asgi_client(env) as client:
response = await client.delete(
"/users/alice", headers={"Authorization": f"Bearer {TOKEN}"}
)
response = await client.delete("/users/alice", headers={"Authorization": f"Bearer {TOKEN}"})
assert response.status_code == 404
@@ -958,9 +914,9 @@ async def test_missing_cookie_in_handshake_is_503(env):
@pytest.mark.asyncio
async def test_pod_annotation_overrides_sandbox_name(env):
sandbox_name, pod_name = add_user(env.k8s)
env.k8s.sandboxes[sandbox_name]["metadata"]["annotations"][
"agents.x-k8s.io/pod-name"
] = "custom-pod"
env.k8s.sandboxes[sandbox_name]["metadata"]["annotations"]["agents.x-k8s.io/pod-name"] = (
"custom-pod"
)
env.k8s.pods["custom-pod"] = {"metadata": {"uid": "other-uid"}}
env.k8s.logs["custom-pod"] = env.k8s.logs[pod_name]
async with asgi_client(env, cookies={"dsh_gw_session": f"alice:{TOKEN}"}) as client:
@@ -1082,9 +1038,7 @@ async def test_proxy_upstream_connection_error_is_502(env):
@pytest.mark.asyncio
async def test_proxy_body_limit_returns_413(env):
add_user(env.k8s)
env.gateway.settings = Settings(
**{**env.settings.__dict__, "max_body_bytes": 4}
)
env.gateway.settings = Settings(**{**env.settings.__dict__, "max_body_bytes": 4})
async with asgi_client(env) as client:
response = await client.post(
"/u/alice/api/echo",
@@ -1194,9 +1148,7 @@ async def test_proxied_requests_use_the_stream_timeout(env, monkeypatch):
"server": ("gateway.test", 80),
"path_params": {"user": "alice", "path": "events"},
}
response = await env.gateway._send_upstream(
Request(scope), "alice", FQDN, "dsh-auth-a=1", b""
)
response = await env.gateway._send_upstream(Request(scope), "alice", FQDN, "dsh-auth-a=1", b"")
await response.aclose()
assert captured["timeout"].read is None
@@ -1359,9 +1311,7 @@ async def test_sweeper_skips_claim_it_does_not_own(env):
assert "alice" not in env.gateway.state.last_activity
# ...and does not suspend it even when it looks long-idle.
env.gateway.state.last_activity["alice"] = (
time.time() - env.settings.idle_timeout - 10
)
env.gateway.state.last_activity["alice"] = time.time() - env.settings.idle_timeout - 10
await env.gateway.sweep_once()
assert env.k8s.patches == []
@@ -1370,9 +1320,7 @@ async def test_sweeper_skips_claim_it_does_not_own(env):
async def test_sweeper_suspends_idle_ready_sandbox(env):
env.k8s.claims["alice"] = ready_claim()
env.k8s.sandboxes["dsh-alice-sbx"] = ready_sandbox()
env.gateway.state.last_activity["alice"] = (
time.time() - env.settings.idle_timeout - 10
)
env.gateway.state.last_activity["alice"] = time.time() - env.settings.idle_timeout - 10
env.gateway.state.handshake[("alice", POD_UID, "gateway.test")] = "dsh-auth-a=x"
await env.gateway.sweep_once()
@@ -1478,9 +1426,7 @@ async def test_sso_landing_new_user_shows_provisioning(env):
async def test_sso_landing_ready_connects_and_offers_recreate_only(env):
enable_sso(env)
add_user(env.k8s)
async with asgi_client(
env, cookies={"dsh_gw_session": f"alice:{TOKEN}"}
) as client:
async with asgi_client(env, cookies={"dsh_gw_session": f"alice:{TOKEN}"}) as client:
response = await client.get("/", headers={"X-Forwarded-User": "alice"})
assert response.status_code == 200
assert "Signed in as <strong>alice</strong>" in response.text
@@ -1512,9 +1458,7 @@ async def test_recreate_deletes_and_reprovisions(env):
enable_sso(env)
add_user(env.k8s)
async with asgi_client(env) as client:
response = await client.post(
"/recreate", headers={"X-Forwarded-User": "alice"}
)
response = await client.post("/recreate", headers={"X-Forwarded-User": "alice"})
assert response.status_code == 303
assert response.headers["location"] == "/u/alice/"
assert env.k8s.deleted == ["alice"]
@@ -1526,9 +1470,7 @@ async def test_recreate_deletes_and_reprovisions(env):
async def test_recreate_requires_sso(env):
add_user(env.k8s)
async with asgi_client(env) as client:
response = await client.post(
"/recreate", headers={"X-Forwarded-User": "alice"}
)
response = await client.post("/recreate", headers={"X-Forwarded-User": "alice"})
assert response.status_code == 401
assert env.k8s.deleted == []
@@ -1537,9 +1479,7 @@ async def test_recreate_requires_sso(env):
async def test_recreate_rejects_cross_origin(env):
enable_sso(env)
add_user(env.k8s)
async with asgi_client(
env, cookies={"dsh_gw_session": f"alice:{TOKEN}"}
) as client:
async with asgi_client(env, cookies={"dsh_gw_session": f"alice:{TOKEN}"}) as client:
response = await client.post(
"/recreate",
headers={"X-Forwarded-User": "alice", "Origin": "http://evil.test"},
@@ -1553,9 +1493,7 @@ async def test_sso_get_user_authorizes_via_header(env):
enable_sso(env)
add_user(env.k8s)
async with asgi_client(env) as client:
response = await client.get(
"/users/alice", headers={"X-Forwarded-User": "alice"}
)
response = await client.get("/users/alice", headers={"X-Forwarded-User": "alice"})
assert response.status_code == 200
assert response.json()["user"] == "alice"
@@ -1565,9 +1503,7 @@ async def test_sso_proxy_redirects_to_bootstrap_without_session(env):
enable_sso(env)
add_user(env.k8s)
async with asgi_client(env) as client:
response = await client.get(
"/u/alice/", headers={"X-Forwarded-User": "alice"}
)
response = await client.get("/u/alice/", headers={"X-Forwarded-User": "alice"})
assert response.status_code == 303
assert response.headers["location"] == "/"
@@ -1576,9 +1512,7 @@ async def test_sso_proxy_redirects_to_bootstrap_without_session(env):
async def test_sso_proxy_redirects_to_bootstrap_when_claim_missing(env):
enable_sso(env)
async with asgi_client(env) as client:
response = await client.get(
"/u/alice/", headers={"X-Forwarded-User": "alice"}
)
response = await client.get("/u/alice/", headers={"X-Forwarded-User": "alice"})
assert response.status_code == 303
assert response.headers["location"] == "/"
@@ -1587,7 +1521,5 @@ async def test_sso_proxy_redirects_to_bootstrap_when_claim_missing(env):
async def test_spoofed_header_ignored_when_sso_disabled(env):
add_user(env.k8s)
async with asgi_client(env) as client:
response = await client.get(
"/u/alice/", headers={"X-Forwarded-User": "alice"}
)
response = await client.get("/u/alice/", headers={"X-Forwarded-User": "alice"})
assert response.status_code == 401