diff --git a/static/cx-core/CHANGELOG.md b/static/cx-core/CHANGELOG.md new file mode 100644 index 00000000..bd32ad45 --- /dev/null +++ b/static/cx-core/CHANGELOG.md @@ -0,0 +1,10 @@ +# cx-core (content-exchange core) - changelog + +One line per version, newest first. Every REST-facing change bumps `cx_core.VERSION` (reported by +`GET /v1/version`, and written into the `[cx_core]` section of iiab.ini by the cx_core Ansible +role), so this file is the human record of what each bump enables. Keep entries short: +`version - change (TICKET)`. + +- **0.3.0** - /v1/share gate: a token-scoped, two-lifetime share session (ephemeral connection vs durable session) over a scoped item set, reusing the export op per item, for granular wireless Share. (K2GO-376) +- **0.2.0** - export as a prepare/poll/fetch long operation, so a large Kolibri channel stages without timing out the client. (K2GO-507) +- **0.1.0** - initial content-exchange core: read (list + identity), export, import and delete over /v1 across Kiwix, Kolibri and Calibre. (K2GO-427) diff --git a/static/cx-core/README.md b/static/cx-core/README.md index baa76be4..4900f3b3 100644 --- a/static/cx-core/README.md +++ b/static/cx-core/README.md @@ -8,7 +8,9 @@ contracts (identity, manifest, adapter port, core REST API). The read, export, import and delete sides are built for all three adapters. Export runs as a liveness-paced long operation (prepare / poll / fetch), so a multi-gigabyte item does -not time out on the client (K2GO-507). +not time out on the client (K2GO-507). Share (K2GO-376) is the network twin: a token-gated +session over a scoped item set that reuses the export op per item (the gate contract only; +pairing and the network carrier are later app slices). ## What is here @@ -16,6 +18,8 @@ not time out on the client (K2GO-507). - `cx_core/ports.py`: the inner contract, the app-adapter port (read methods). - `cx_core/adapters/kiwix.py`: the first adapter, read side (ZIM header UUID + listing). - `cx_core/registry.py`: loads an adapter only when its app is installed. +- `cx_core/export_ops.py`: the export long-operation registry (op lifecycle + temp sweeps). +- `cx_core/share_ops.py`: the share gate (token + scoped item set + two lifetimes), K2GO-376. - `cx_core/router.py`: pure `/v1` route dispatch (unit-testable without a socket). - `cx_core/server.py` + `__main__.py`: a thin stdlib `http.server` wrapper, localhost only. @@ -68,6 +72,38 @@ Import / delete: - `POST /v1/export/plan`, `POST /v1/dedup/check`, `POST /v1/import/validate`, `POST /v1/import/item`, `DELETE /v1/content/{app}/{localRef}`. +Share (a gate over the export op, K2GO-376): Share is the network twin of Export. The +receiver pulls selected items from the sender's live cx-core, so the transfer can close +itself (the sender machine is on, unlike Clone). The gate has two lifetimes: the CONNECTION +(one open stream) reuses the export op's idle TTL + in_use pin per item, and the durable +SESSION (a token + a scoped item set + per-item delivered state) does not expire on the short +connection idle. The session closes on COMPLETE (the receiver has every item), CANCEL +(DELETE, by host or receiver; the sender's foreground service calls it in onDestroy), a long +SAFETY CAP (hours of no activity), or process death (in-memory + the export startup sweep). +Each item is a lazily created export fetch, so staging, the in_use pin and the on_close +cleanup are the export op's. An item counts as delivered only on a CLEAN close (the server +marks a full body), so a mid-stream drop stays pending and a resume re-fetches it. This slice +is loopback only: pairing (QR, hotspot, WLAN) is a later app slice. + +The session token travels in the `X-Share-Token` request header, never the URL (it is the +only auth, so it must not reach access logs). + +- `POST /v1/share {items: [{app, localRef}]}` -> `{sessionId, token, manifestUrl}`. Opens a + session scoped to the item set. +- `GET /v1/share/{sessionId}/manifest` -> the bundle manifest over the scope (per-item + identity, size, upstream hash when present, each with its `itemRef` and whether it is already + `delivered`), plus `missing` and `totalBytes`. +- `GET /v1/share/{sessionId}/fetch/{itemRef}` -> the chunked octet stream for one item. + `409 {state: staging}` while a staging item (Kolibri) stages (the receiver retries), + `409 {state: busy}` when the item already has an open stream (single-flight), `200` streams + once ready. +- `POST /v1/share/{sessionId}/complete` -> `{ok: true}`. The receiver's clean-success close. +- `DELETE /v1/share/{sessionId}` -> `{ok: true}`. Cancel (abort) by host or receiver. Both + closers revoke the token and every child temp; `404` on an unknown or already-closed session. + +Token checks return `401` when no token is sent, `403` on a mismatch, `404` on an unknown or +expired session. + ## Test ``` @@ -77,14 +113,16 @@ python -m unittest discover -s tests -t . The tests cover the pure logic with no device: ZIM header UUID parsing, filename decomposition, soft-key normalization, the manifest JSON shape, route dispatch, the export -long-operation registry (prepare / status / fetch / delete, the TTL and startup sweeps), -and the streaming over a real socket. +long-operation registry (prepare / status / fetch / delete, the TTL and startup sweeps), the +share gate (create / manifest / fetch / complete / delete, header-token enforcement, +single-flight, clean-close delivery, the safety cap), and the streaming over a real socket. ## Not built yet -The network and QR carriers, strict-mode content hashing, the cx-core live self-update -(D1d; the Ansible role baseline is in `tools/upstream-patches/overlays/roles/cx_core`), and -the dash-node <-> core listing consolidation (D1b). +The Share pairing and network carrier (QR, hotspot, WLAN: the app slices that build on the +`/v1/share` gate above), strict-mode content hashing, the cx-core live self-update (D1d; the +Ansible role baseline is in `tools/upstream-patches/overlays/roles/cx_core`), and the +dash-node <-> core listing consolidation (D1b). The registry resolves installed apps once at startup (a snapshot). Reacting to a module being installed or removed at runtime is a deferred design item: it must consume the diff --git a/static/cx-core/cx_core/__init__.py b/static/cx-core/cx_core/__init__.py index fe8a0fa7..e0213fe9 100644 --- a/static/cx-core/cx_core/__init__.py +++ b/static/cx-core/cx_core/__init__.py @@ -7,5 +7,5 @@ # The core build. Bumped per core change so a box can report which core it runs (GET # /v1/version), mirroring dash-node's package.json version and /system/version. The # cx_core Ansible role writes this into the [cx_core] section of iiab.ini. -VERSION = "0.2.0" +VERSION = "0.3.0" __version__ = VERSION diff --git a/static/cx-core/cx_core/ports.py b/static/cx-core/cx_core/ports.py index 25a49a9f..166b58a7 100644 --- a/static/cx-core/cx_core/ports.py +++ b/static/cx-core/cx_core/ports.py @@ -24,13 +24,19 @@ class ExportStream: error); the EXPORT OP LAYER sets both: on_open pins the op for the stream's life, on_close unpins and drops it. The server fires on_open at the top of the same try whose finally runs on_close, so the pin and its release are a matched pair even when the client disconnects. - The adapter only opens the stream; the op layer owns the lifecycle (K2GO-507).""" + The adapter only opens the stream; the op layer owns the lifecycle (K2GO-507). + + ``completed`` is set True by the server after the FULL body has been written (the clean + terminator), and stays False on a mid-stream client drop. The export op ignores it (it drops + the temp either way), but the share gate reads it so a share item counts as delivered only on + a clean close, never on a drop (K2GO-376).""" filename: str content_type: str chunks: Iterable[bytes] on_close: Optional[Callable[[], None]] = None on_open: Optional[Callable[[], None]] = None + completed: bool = False @dataclass diff --git a/static/cx-core/cx_core/router.py b/static/cx-core/cx_core/router.py index 23a28903..a44d6523 100644 --- a/static/cx-core/cx_core/router.py +++ b/static/cx-core/cx_core/router.py @@ -15,6 +15,7 @@ from .export_ops import ExportOps from .importing import is_within_dir from .registry import Registry +from .share_ops import ShareOps Response = Tuple[int, Any] @@ -165,10 +166,14 @@ def _export_prepare(ops: Optional[ExportOps], body: Any) -> Response: def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any = None, - ops: Optional[ExportOps] = None) -> Response: + ops: Optional[ExportOps] = None, share: Optional[ShareOps] = None, + token: Optional[str] = None) -> Response: """Route one request. ``query`` is a dict of first values; ``body`` is the parsed JSON - for a POST; ``ops`` is the export long-operation registry (the server supplies it, the - export-op routes need it).""" + for a POST; ``ops`` is the export long-operation registry and ``share`` is the share-gate + registry (the server supplies both; the routes that need them guard against None). ``token`` + is the share session token: the server reads it from the ``X-Share-Token`` request header + (never the URL, so it stays out of access logs) and passes it here, so the router stays + header-agnostic and unit-testable without a socket.""" trimmed = path.rstrip("/") or "/" segments = [unquote(s) for s in trimmed.split("/") if s != ""] @@ -179,6 +184,17 @@ def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any # POST /v1/export/plan : plan an export (sizes only, no payload moved). if segments == ["v1", "export", "plan"]: return _export_plan(registry, body) + # POST /v1/share : open a share session, return its sessionId + token + manifest pointer. + if segments == ["v1", "share"]: + if share is None: + return 500, {"error": "share unavailable"} + items = body.get("items") if isinstance(body, dict) else None + return share.create(items) + # POST /v1/share/{sessionId}/complete : the receiver signals a clean, done transfer. + if len(segments) == 4 and segments[0] == "v1" and segments[1] == "share" and segments[3] == "complete": + if share is None: + return 500, {"error": "share unavailable"} + return share.complete(segments[2], token) # POST /v1/dedup/check : receiver dedup verdicts, nothing moved. if segments == ["v1", "dedup", "check"]: return _dedup_check(registry, body) @@ -196,6 +212,11 @@ def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any if ops is None: return 500, {"error": "export ops unavailable"} return ops.delete(segments[2]) + # DELETE /v1/share/{sessionId} : cancel a share session (abort, by host or receiver). + if len(segments) == 3 and segments[0] == "v1" and segments[1] == "share": + if share is None: + return 500, {"error": "share unavailable"} + return share.delete(segments[2], token) # DELETE /v1/content/{app}/{localRef} : reclaim space by removing one item. if len(segments) == 4 and segments[0] == "v1" and segments[1] == "content": return _delete_item(registry, segments[2], segments[3]) @@ -225,6 +246,19 @@ def dispatch(registry: Registry, method: str, path: str, query: dict, body: Any return 500, {"error": "export ops unavailable"} return ops.fetch(segments[3]) + # GET /v1/share/{sessionId}/manifest : the bundle manifest over the session's item set. + if len(segments) == 4 and segments[0] == "v1" and segments[1] == "share" and segments[3] == "manifest": + if share is None: + return 500, {"error": "share unavailable"} + return share.manifest(segments[2], token) + + # GET /v1/share/{sessionId}/fetch/{itemRef} : stream one shared item (payload is an + # ExportStream the server streams). 409 while the item stages; the receiver retries. + if len(segments) == 5 and segments[0] == "v1" and segments[1] == "share" and segments[3] == "fetch": + if share is None: + return 500, {"error": "share unavailable"} + return share.fetch(segments[2], segments[4], token) + # GET /v1/storage : box content-volume disk summary + per-app content totals. if segments == ["v1", "storage"]: return _storage(registry) diff --git a/static/cx-core/cx_core/server.py b/static/cx-core/cx_core/server.py index 704db989..ab908fae 100644 --- a/static/cx-core/cx_core/server.py +++ b/static/cx-core/cx_core/server.py @@ -15,23 +15,30 @@ from .ports import ExportStream from .registry import Registry from .router import dispatch +from .share_ops import ShareOps _MAX_BODY = 1 * 1024 * 1024 # a plan request is small; cap the read defensively _BAD_JSON = object() # sentinel: body present but not valid JSON _TOO_LARGE = object() # sentinel: declared body exceeds _MAX_BODY -def make_handler(registry: Registry, ops: ExportOps): +def make_handler(registry: Registry, ops: ExportOps, share: ShareOps = None): class Handler(BaseHTTPRequestHandler): server_version = f"cx-core/{VERSION}" # HTTP/1.1 so the chunked export stream is valid (HTTP/1.0 would hand the client the # chunk-size lines as body). The JSON path sends Content-Length, so keep-alive holds. protocol_version = "HTTP/1.1" + def _share_token(self): + # The share session token travels in a header, never the URL query, so it does not + # land in access logs (it is the only auth for the gate). + return self.headers.get("X-Share-Token") + def do_GET(self) -> None: # noqa: N802 (http.server naming) parts = urlsplit(self.path) query = {k: v[0] for k, v in parse_qs(parts.query).items()} - status, payload = dispatch(registry, "GET", parts.path, query, ops=ops) + status, payload = dispatch(registry, "GET", parts.path, query, ops=ops, share=share, + token=self._share_token()) if isinstance(payload, ExportStream): self._stream(status, payload) else: @@ -48,13 +55,15 @@ def do_POST(self) -> None: # noqa: N802 (http.server naming) if body is _BAD_JSON: self._send(400, {"error": "invalid JSON body"}) return - status, payload = dispatch(registry, "POST", parts.path, {}, body, ops=ops) + status, payload = dispatch(registry, "POST", parts.path, {}, body, ops=ops, share=share, + token=self._share_token()) self._send(status, payload) def do_DELETE(self) -> None: # noqa: N802 (http.server naming) parts = urlsplit(self.path) query = {k: v[0] for k, v in parse_qs(parts.query).items()} - status, payload = dispatch(registry, "DELETE", parts.path, query, ops=ops) + status, payload = dispatch(registry, "DELETE", parts.path, query, ops=ops, share=share, + token=self._share_token()) self._send(status, payload) def _read_json_body(self): @@ -86,15 +95,18 @@ def _stream(self, status: int, stream: ExportStream) -> None: # Close after the stream: the body length is unknown up front, so a chunked # response plus an explicit close keeps the connection state simple. self.close_connection = True - self.send_response(status) - self.send_header("Content-Type", stream.content_type) - self.send_header("Content-Disposition", f'attachment; filename="{stream.filename}"') - self.send_header("Transfer-Encoding", "chunked") - self.send_header("Connection", "close") - self.end_headers() try: - # Pin the op for the stream's life inside the same try whose finally runs - # on_close, so the pin is always released (even on a client disconnect). + # Write the headers INSIDE the same try whose finally runs on_close, so on_close + # is guaranteed on every exit: a client that disconnects during the header write + # must still release the op pin and the share single-flight slot (K2GO-376), + # never strand them. on_open follows the headers, so a pre-header failure runs + # on_close without a matching on_open: the op cleanup (_finish) is safe either way. + self.send_response(status) + self.send_header("Content-Type", stream.content_type) + self.send_header("Content-Disposition", f'attachment; filename="{stream.filename}"') + self.send_header("Transfer-Encoding", "chunked") + self.send_header("Connection", "close") + self.end_headers() if stream.on_open: stream.on_open() for block in stream.chunks: @@ -104,6 +116,10 @@ def _stream(self, status: int, stream: ExportStream) -> None: self.wfile.write(block) self.wfile.write(b"\r\n") self.wfile.write(b"0\r\n\r\n") + # The full body (and its terminator) is written: a clean close. A mid-stream + # client drop raises above and leaves this False, so the share gate counts the + # item delivered only on a clean close (K2GO-376). + stream.completed = True finally: if stream.on_close: stream.on_close() @@ -118,10 +134,14 @@ def log_message(self, *_args) -> None: def serve(config: Config) -> None: registry = Registry.load(config) ops = ExportOps(registry, config.staging_dir) + share = ShareOps(registry, ops) # Reclaim export op temps left by a dead process before serving (K2GO-507 process-death - # cleanup owner): a fresh process holds no ops, so any op temp on disk is an orphan. + # cleanup owner): a fresh process holds no ops, so any op temp on disk is an orphan. The + # share sweep then drops any in-memory sessions (K2GO-376 closer 2); a share session's own + # on-disk footprint is its child export temps, already reclaimed by the export sweep above. ops.sweep_startup() - httpd = ThreadingHTTPServer((config.host, config.port), make_handler(registry, ops)) + share.sweep_startup() + httpd = ThreadingHTTPServer((config.host, config.port), make_handler(registry, ops, share)) print(f"cx-core {VERSION} listening on http://{config.host}:{config.port} " f"(apps: {', '.join(registry.apps()) or 'none'})") httpd.serve_forever() diff --git a/static/cx-core/cx_core/share_ops.py b/static/cx-core/cx_core/share_ops.py new file mode 100644 index 00000000..c2ba4bd9 --- /dev/null +++ b/static/cx-core/cx_core/share_ops.py @@ -0,0 +1,394 @@ +"""Share gate: the shared contract for granular wireless Share (K2GO-376). + +Share is the NETWORK TWIN of Export. The bundle, the manifest and the import path are the +same; only the carrier changes. The receiver PULLS from the sender's live cx-core (the +machine is ON), so the transfer can close itself: this is the whole point of Share over +Clone, whose standalone rsync daemon is mute on the host side and can only be stopped by +the user. See controller/docs/ADR-content-exchange.md, section "Pillar Share". + +Two lifetimes (the refinement that makes large, resumable transfers work; collapsing them +into one short idle TTL was the first-cut error: a multi-hour transfer that pauses would be +reclaimed mid-way): + + CONNECTION (ephemeral, Layer 1): ONE open stream. Its idle TTL and in_use pin belong to the + CHILD EXPORT OP (export_ops.py), reused per item: a dead socket's op is reclaimed after the + export idle TTL (~10 min), a live stream is pinned so its temp is never swept under it, and a + per-item resume is just a re-fetch (the export op re-stages). This gate adds no connection + TTL of its own; it reuses the export op's. + + SESSION (durable, Layer 2): a TOKEN plus a SCOPED item set plus the per-item delivered state. + It does NOT expire on the short connection idle. It lives until ONE of: + : COMPLETE : the receiver signals it has every item (clean success) -> release. + : CANCEL (DELETE, by host or receiver) : abort; the sender foreground service calls DELETE + in onDestroy (the app-side closer). + : a long SAFETY CAP (hours of no activity), so a truly abandoned session is reclaimed; this + is NOT the ~10-minute connection idle. An active transfer keeps the session fresh. + : process death : the session is in-memory (v1 decision), so a restart loses it and the + export op's startup sweep reclaims the child temps on boot. + +The session does not reinvent the transfer: each shared item is a lazily created export op, so +staging + the in_use pin + on_close cleanup are the hardened, device-validated K2GO-507 ones. +The session is the GATE over those child ops: it holds the token and the scope, ties the child +ops together, and revokes them all at once on close. + +Ownership (one source per fact, so this is not a second owner of anything the export op owns): + : the session owns the token, the scoped item set, the delivered state and the session's own + liveness (the safety cap). + : the child export op owns the staged temp, the stream and the connection liveness (idle TTL + + in_use pin). +The ``streaming`` set is the session's single truth for "which items have an open stream": it +both single-flights a per-item fetch (so two streams never share one child op's temp) and keeps +the safety cap from reclaiming a session mid-transfer. That is a nested scope of the export +op's own per-op pin, not a duplicated truth. + +delivered is set ONLY on a CLEAN item close (the server marks ExportStream.completed after the +full body), never on a mid-stream drop, so a dropped item stays pending and a resume re-fetches +it. The manifest surfaces per-item delivered, so a resuming receiver skips what it already has. +The clean close is signalled; it is never guessed. + +The token travels in the ``X-Share-Token`` request header, never the URL, so it stays out of +access logs (it is the only auth). It is opaque, bound to the session (revoking the session +revokes the token) and scoped to the item set. The only surface a token holder reaches is +/v1/share/* scoped to this session's items; the wider /v1 API is never widened. + +This slice is loopback only: pairing (QR, hotspot, WLAN) is a later presentation slice and is +outside the core, the way plugging in the Ethernet cable is not part of content exchange. +""" + +from __future__ import annotations + +import secrets +import threading +import time +from dataclasses import dataclass, field +from typing import Any, Dict, List, Tuple, Union + +from .registry import Registry + +Response = Tuple[int, Any] + +# How long a session may go with NO activity before the safety cap reclaims it. Hours, not the +# export op's ~10-minute connection idle: a paused transfer that resumes within the window +# survives, and a truly abandoned session is eventually reclaimed. The cap is a backstop; the +# normal closes are COMPLETE and CANCEL. +DEFAULT_SAFETY_CAP_SECONDS = 6 * 3600 + + +@dataclass +class _Session: + token: str + # The scoped item set, as (app, local_ref) pairs. The index is the item's itemRef in the + # manifest and the fetch URL: an opaque per-session handle, so a Calibre local_ref that + # contains slashes never lands in a URL path segment. + items: List[Tuple[str, str]] + # itemRef -> the child export opId, created lazily on first fetch of that item. + item_ops: Dict[str, str] = field(default_factory=dict) + # itemRefs delivered on a CLEAN stream close (full body). Surfaced in the manifest for resume. + delivered: set = field(default_factory=set) + # itemRefs with an open (or opening) stream: single-flight + "do not cap mid-transfer". + streaming: set = field(default_factory=set) + # Last activity, for the safety cap only (bumped on manifest / fetch / complete / close). + last_seen: float = field(default_factory=time.monotonic) + + +class ShareOps: + """Thread-safe registry of share sessions. One instance lives for the server's lifetime; + the handler threads call into it concurrently. It wraps the one ExportOps instance and + delegates every per-item transfer (the connection lifetime) to it.""" + + def __init__(self, registry: Registry, export_ops, + safety_cap_seconds: int = DEFAULT_SAFETY_CAP_SECONDS): + self._registry = registry + self._export = export_ops + self._cap = safety_cap_seconds + self._lock = threading.Lock() + self._sessions: Dict[str, _Session] = {} + + # ---- REST-facing operations (return (status, payload), or (200, ExportStream)) ---- + + def create(self, items: Any) -> Response: + """Open a share session over a scoped item set. Returns the sessionId, an opaque token + bound to the session, and the manifest pointer. Partial is first-class: the set is + recorded as requested and each item is resolved at manifest / fetch time, so an absent + item is reported, not fatal.""" + if not isinstance(items, list) or not items: + return 400, {"error": "body must be {items: [{app, localRef}]} with at least one item"} + scoped: List[Tuple[str, str]] = [] + seen = set() + for req in items: + app = req.get("app") if isinstance(req, dict) else None + local_ref = req.get("localRef") if isinstance(req, dict) else None + if not isinstance(app, str) or not isinstance(local_ref, str) or not local_ref: + return 400, {"error": "each item needs a string app and a non-empty localRef"} + key = (app, local_ref) + if key not in seen: + seen.add(key) + scoped.append(key) + session_id = secrets.token_hex(16) + token = secrets.token_urlsafe(32) + with self._lock: + self._sessions[session_id] = _Session(token=token, items=scoped) + # The pointer carries no token: the QR carries the address, the token and this pointer + # as separate fields (ADR), and the receiver sends the token as the X-Share-Token header. + return 200, { + "sessionId": session_id, + "token": token, + "manifestUrl": f"/v1/share/{session_id}/manifest", + } + + def manifest(self, session_id: str, token: Any) -> Response: + """The bundle manifest over the session's scoped set: per-item identity, size and + upstream hash when present, each tagged with its itemRef and a ``delivered`` flag. + Reuses the read-side item JSON, so it is the same manifest shape the disk export builds. + + ``delivered`` is ADVISORY: it means the SENDER streamed this item to completion once (a + clean stream close), NOT that the receiver persisted it. It is a progress / resume HINT. + The receiver's resume authority must be its OWN persisted state, so it never skips an item + it received but failed to save.""" + session = self._resolve(session_id, token) + if isinstance(session, tuple): + return session + found: List[dict] = [] + missing: List[dict] = [] + total = 0 + for index, (app, local_ref) in enumerate(session.items): + item_ref = str(index) + if not self._registry.has(app): + missing.append({"itemRef": item_ref, "app": app, "localRef": local_ref, + "reason": "unknown app"}) + continue + item = self._registry.get(app).get(local_ref) + if item is None: + missing.append({"itemRef": item_ref, "app": app, "localRef": local_ref, + "reason": "not found"}) + continue + entry = item.to_json() + entry["itemRef"] = item_ref + entry["delivered"] = item_ref in session.delivered + found.append(entry) + total += item.bytes + return 200, {"sessionId": session_id, "items": found, "missing": missing, + "totalBytes": total} + + def fetch(self, session_id: str, item_ref: str, token: Any): + """Stream one shared item. Lazily creates the child export op the first time an item is + fetched, then drives it: 409 {state: staging} while a staging app (Kolibri) stages (the + receiver retries this same URL), 409 {state: busy} when the item already has an open + stream (single-flight), (200, ExportStream) once ready. The stream's on_close releases + the single-flight slot and, on a clean close, records the item delivered.""" + session = self._resolve(session_id, token) + if isinstance(session, tuple): + return session + target = self._item(session, item_ref) + if target is None: + return 404, {"error": "unknown item"} + app, local_ref = target + # Single-flight per item: one open stream at a time, so two fetches never stream the same + # child op's temp (the first close removes it under the second). A second concurrent + # fetch of the same item gets 409 busy and retries. Claim the slot before touching the + # export op; the stream's on_close releases it on the 200 path, the finally on every + # other path. + with self._lock: + live = self._sessions.get(session_id) + if live is None: + return 404, {"error": "unknown session"} + if item_ref in live.streaming: + return 409, {"state": "busy", "itemRef": item_ref} + live.streaming.add(item_ref) + keep = False + try: + # A child op swept or already finished returns 404; re-prepare and retry once, so a + # resumed fetch re-stages instead of dead-ending on a stale opId. + for attempt in (1, 2): + op_id = self._ensure_op(session_id, item_ref, app, local_ref) + if isinstance(op_id, tuple): + return op_id # prepare failed: 404 absent, 507 no space + status, result = self._export.fetch(op_id) + if status == 404 and attempt == 1: + self._drop_op(session_id, item_ref, op_id) + continue + if status == 409: + return self._staging_response(op_id, item_ref) + if status != 200: + return status, result + self._wrap_close(session_id, item_ref, result) + keep = True # the stream now owns the slot; its on_close releases it + return 200, result + return 404, {"error": "item not available"} + finally: + if not keep: + self._release_streaming(session_id, item_ref) + + def complete(self, session_id: str, token: Any) -> Response: + """Clean-success close: the receiver has every item it wanted and releases the session. + This is the receiver-driven close (supersedes any drop-based auto-close). Revokes the + token and every child op + temp. 404 on an unknown or already-closed session.""" + return self._close(session_id, token) + + def delete(self, session_id: str, token: Any) -> Response: + """Cancel (abort) close, by the host or the receiver; also the app-side closer (the + sender's foreground service calls this in onDestroy). Revokes the token and every child + op + temp. 404 on an unknown or already-closed session, so a double close (receiver and + sender) is harmless: the second caller sees the gate already closed.""" + return self._close(session_id, token) + + # ---- lifecycle ---- + + def sweep_startup(self) -> None: + """Drop any in-memory sessions on boot. A fresh process holds none, so this is + defensive; the on-disk footprint of a session is its child export temps (cxexport-*), + reclaimed by the export op's own startup sweep, which the server runs on boot. Reusing + that single sweep avoids a second temp owner.""" + with self._lock: + self._sessions.clear() + + # ---- token + item resolution ---- + + def _resolve(self, session_id: str, token: Any) -> Union[_Session, Response]: + """Return the live session, bumping last_seen (keeps an active session off the safety + cap), or an error response: 404 on an unknown or expired session, 401 when no token is + sent, 403 on a token mismatch.""" + self._sweep_capped() + with self._lock: + session = self._sessions.get(session_id) + if session is None: + return 404, {"error": "unknown session"} + denied = self._check_token(session, token) + if denied is not None: + return denied + session.last_seen = time.monotonic() + return session + + def _close(self, session_id: str, token: Any) -> Response: + """Resolve + token-check + revoke. The single close primitive behind COMPLETE and + CANCEL: both revoke the token and the child ops; they differ only in the app's intent.""" + self._sweep_capped() + with self._lock: + session = self._sessions.get(session_id) + if session is None: + return 404, {"error": "unknown session"} + denied = self._check_token(session, token) + if denied is not None: + return denied + self._sessions.pop(session_id, None) + child_ops = list(session.item_ops.values()) + for op_id in child_ops: + self._export.delete(op_id) # drops each child op + removes its staged temp + return 200, {"ok": True} + + @staticmethod + def _check_token(session: _Session, token: Any): + """None when the token is valid, else the error response. compare_digest is + constant-time, so a wrong token leaks no timing signal.""" + if not isinstance(token, str) or token == "": + return 401, {"error": "token required"} + if not secrets.compare_digest(token, session.token): + return 403, {"error": "token mismatch"} + return None + + @staticmethod + def _item(session: _Session, item_ref: str): + """Resolve an itemRef (its index in the scoped set) to (app, local_ref), or None.""" + try: + index = int(item_ref) + except (TypeError, ValueError): + return None + if 0 <= index < len(session.items): + return session.items[index] + return None + + def _ensure_op(self, session_id: str, item_ref: str, app: str, local_ref: str): + """Return the child export opId for this item, preparing one if none exists. Returns an + error response tuple when prepare fails (absent item or no staging space).""" + with self._lock: + session = self._sessions.get(session_id) + if session is None: + return 404, {"error": "unknown session"} + op_id = session.item_ops.get(item_ref) + if op_id is not None: + return op_id + status, payload = self._export.prepare(app, local_ref) + if status != 200: + return status, payload + op_id = payload["opId"] + with self._lock: + session = self._sessions.get(session_id) + if session is None: + # The session closed (CANCEL / cap) while we prepared: drop the orphan op we + # just made so its temp is not leaked. + self._export.delete(op_id) + return 404, {"error": "unknown session"} + # Single-flight guarantees no concurrent fetch of this same item, so item_ops has no + # entry for it here; store ours as the one child op for the item. + session.item_ops[item_ref] = op_id + return op_id + + def _drop_op(self, session_id: str, item_ref: str, op_id: str) -> None: + """Forget a child op mapping whose op has vanished, so the next fetch re-prepares.""" + with self._lock: + session = self._sessions.get(session_id) + if session is not None and session.item_ops.get(item_ref) == op_id: + session.item_ops.pop(item_ref, None) + + def _staging_response(self, op_id: str, item_ref: str) -> Response: + """The item is not ready: report it through the one fetch endpoint, since the gate has + no separate status route (do not widen the API). 500 on a staging error so the receiver + stops retrying; 409 {state: staging} with the heartbeat otherwise so it retries and can + show progress.""" + st_status, st = self._export.status(op_id) + if st_status == 200 and st.get("state") == "error": + return 500, {"state": "error", "itemRef": item_ref, "error": st.get("error")} + progress = {"state": "staging", "itemRef": item_ref} + if st_status == 200: + progress["stagedBytes"] = st.get("stagedBytes", 0) + progress["totalBytes"] = st.get("totalBytes", 0) + return 409, progress + + def _wrap_close(self, session_id: str, item_ref: str, stream) -> None: + """Layer the single-flight release + delivered bookkeeping onto the export op's own + on_close. The server fires on_close in the same try's finally that opens the stream, so + it always runs, even on a client disconnect. on_open stays the export op's (its per-op + pin); only on_close needs the session bookkeeping.""" + inner_close = stream.on_close + + def on_close() -> None: + # The export op's close runs first (drops the child op and removes its temp), then + # the session records the outcome. A clean close (full body) marks the item + # delivered; a mid-stream drop leaves it pending so a resume re-fetches it. + if inner_close: + inner_close() + self._on_stream_close(session_id, item_ref, bool(getattr(stream, "completed", False))) + + stream.on_close = on_close + + def _on_stream_close(self, session_id: str, item_ref: str, completed: bool) -> None: + with self._lock: + session = self._sessions.get(session_id) + if session is None: + return + session.streaming.discard(item_ref) + session.item_ops.pop(item_ref, None) # the child op finished; a re-fetch re-prepares + session.last_seen = time.monotonic() + if completed: + session.delivered.add(item_ref) + + def _release_streaming(self, session_id: str, item_ref: str) -> None: + """Release a single-flight slot claimed by a fetch that did not open a stream (staging, + error, or absent).""" + with self._lock: + session = self._sessions.get(session_id) + if session is not None: + session.streaming.discard(item_ref) + + def _sweep_capped(self) -> None: + now = time.monotonic() + capped: List[_Session] = [] + with self._lock: + for session_id, session in list(self._sessions.items()): + # Never reclaim a session with an open stream: its transfer is live. Otherwise an + # abandoned session (no activity past the safety cap) is reclaimed. + if not session.streaming and now - session.last_seen > self._cap: + capped.append(self._sessions.pop(session_id)) + for session in capped: + for op_id in list(session.item_ops.values()): + self._export.delete(op_id) diff --git a/static/cx-core/tests/test_share_ops.py b/static/cx-core/tests/test_share_ops.py new file mode 100644 index 00000000..9a5fb881 --- /dev/null +++ b/static/cx-core/tests/test_share_ops.py @@ -0,0 +1,670 @@ +"""Share gate tests (K2GO-376): the shared contract for granular wireless Share. + +Everything runs on loopback, like the export op tests: a share session wraps the export op, +so these cover the gate (token in a header, the scope, the two lifetimes, single-flight, +clean-close delivery) and the per-item transfer reuse. The app forks build against this. +""" + +import http.client +import json +import os +import tempfile +import threading +import time +import unittest +from http.server import ThreadingHTTPServer + +from cx_core.export_ops import ExportOps +from cx_core.model import AUTHORITATIVE, Dedup, Identity, ManifestItem +from cx_core.ports import ContentAdapter, ExportPlan, ExportStream +from cx_core.registry import Registry +from cx_core.router import dispatch +from cx_core.server import make_handler +from cx_core.share_ops import ShareOps + + +def _mitem(app, local_ref, total): + return ManifestItem( + app=app, kind="zim", title=local_ref, bytes=total, + identity=Identity(local_ref, Dedup(line={}, build=None, version="", id_kind=AUTHORITATIVE))) + + +class _DirectFake(ContentAdapter): + """A direct app (Kiwix / Calibre): nothing to stage, ready at once. ``items`` maps a + local_ref to its byte size.""" + + app = "kiwix" + + def __init__(self, items): + self._items = dict(items) + + def is_installed(self): + return True + + def list(self): + return [_mitem(self.app, ref, total) for ref, total in self._items.items()] + + def prepare_export(self, local_ref): + if local_ref not in self._items: + return None + return ExportPlan(False, self._items[local_ref], local_ref, "application/octet-stream") + + def open_fetch(self, local_ref, staged_dir=None): + if local_ref not in self._items: + return None + return ExportStream(local_ref, "application/octet-stream", [b"x" * self._items[local_ref]]) + + +class _StagingFake(ContentAdapter): + """A staging app (Kolibri): stage_export writes into dest, optionally gated so a test can + observe the staging state before the op flips to ready.""" + + app = "kolibri" + + def __init__(self, items, gate=None, fail=False): + self._items = dict(items) + self._gate = gate + self._fail = fail + + def is_installed(self): + return True + + def list(self): + return [_mitem(self.app, ref, total) for ref, total in self._items.items()] + + def prepare_export(self, local_ref): + if local_ref not in self._items: + return None + return ExportPlan(True, self._items[local_ref], f"{local_ref}.tar", "application/x-tar") + + def stage_export(self, local_ref, dest, should_cancel=None): + if self._gate is not None: + self._gate.wait(5) + if self._fail: + return False + with open(os.path.join(dest, "part.bin"), "wb") as f: + f.write(b"x" * self._items[local_ref]) + return True + + def open_fetch(self, local_ref, staged_dir=None): + path = os.path.join(staged_dir or "", "part.bin") + data = b"" + if os.path.isfile(path): + with open(path, "rb") as f: + data = f.read() + return ExportStream(f"{local_ref}.tar", "application/x-tar", [data]) + + +def _share(adapters, staging, cap=3600, conn_ttl=600): + """Build a share gate over an export op. ``cap`` is the Layer-2 session safety cap; the + ``conn_ttl`` is the Layer-1 export-op (connection) idle TTL: two separate timeouts.""" + reg = Registry(adapters, import_root=staging, content_volume=staging) + ops = ExportOps(reg, staging_dir=staging, ttl_seconds=conn_ttl) + share = ShareOps(reg, ops, safety_cap_seconds=cap) + return share, ops, reg + + +def _create(share, items): + status, body = share.create(items) + assert status == 200, (status, body) + return body["sessionId"], body["token"] + + +def _drain(stream): + """Drive a stream as the server would on a CLEAN close: on_open, consume the chunks, mark + completed (the server sets this after the full body), on_close.""" + if stream.on_open: + stream.on_open() + try: + data = b"".join(stream.chunks) + stream.completed = True + return data + finally: + if stream.on_close: + stream.on_close() + + +def _drain_drop(stream): + """Drive a stream as a MID-STREAM client drop: open, then close without a full read, so + ``completed`` stays False and the item is not counted delivered.""" + if stream.on_open: + stream.on_open() + if stream.on_close: + stream.on_close() + + +def _fetch_ready(share, sid, item_ref, token, timeout=5): + end = time.time() + timeout + while time.time() < end: + status, result = share.fetch(sid, item_ref, token) + if status == 200: + return result + if status != 409: + raise AssertionError(f"unexpected fetch status {status}: {result}") + time.sleep(0.01) + raise AssertionError("item never became ready") + + +def _wait_ready(ops, op_id, timeout=5): + end = time.time() + timeout + while ops._ops[op_id].state != "ready" and time.time() < end: + time.sleep(0.01) + + +class CreateTest(unittest.TestCase): + def test_create_returns_session_token_and_pointer(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + status, body = share.create([{"app": "kiwix", "localRef": "a.zim"}]) + self.assertEqual(status, 200) + self.assertIn("sessionId", body) + self.assertTrue(body["token"]) + self.assertEqual(body["manifestUrl"], f"/v1/share/{body['sessionId']}/manifest") + # The pointer carries no token: the token travels as the X-Share-Token header. + self.assertNotIn(body["token"], body["manifestUrl"]) + + def test_distinct_sessions_get_distinct_tokens(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + a = share.create([{"app": "kiwix", "localRef": "a.zim"}])[1] + b = share.create([{"app": "kiwix", "localRef": "a.zim"}])[1] + self.assertNotEqual(a["sessionId"], b["sessionId"]) + self.assertNotEqual(a["token"], b["token"]) + + def test_empty_or_bad_body_400(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + self.assertEqual(share.create([])[0], 400) + self.assertEqual(share.create("nope")[0], 400) + self.assertEqual(share.create([{"app": "kiwix"}])[0], 400) + self.assertEqual(share.create([{"localRef": "a.zim"}])[0], 400) + self.assertEqual(share.create([{"app": "kiwix", "localRef": ""}])[0], 400) + + def test_duplicate_items_are_deduped(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [ + {"app": "kiwix", "localRef": "a.zim"}, + {"app": "kiwix", "localRef": "a.zim"}, + ]) + _, body = share.manifest(sid, token) + self.assertEqual(len(body["items"]), 1) + + +class ManifestTest(unittest.TestCase): + def test_manifest_lists_scoped_items_with_itemref_total_and_delivered(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10, "b.zim": 25})], d) + sid, token = _create(share, [ + {"app": "kiwix", "localRef": "a.zim"}, + {"app": "kiwix", "localRef": "b.zim"}, + ]) + status, body = share.manifest(sid, token) + self.assertEqual(status, 200) + self.assertEqual([i["itemRef"] for i in body["items"]], ["0", "1"]) + self.assertEqual([i["identity"]["localRef"] for i in body["items"]], ["a.zim", "b.zim"]) + self.assertEqual([i["delivered"] for i in body["items"]], [False, False]) + self.assertEqual(body["totalBytes"], 35) + self.assertEqual(body["missing"], []) + + def test_manifest_reports_missing_items_partial(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [ + {"app": "kiwix", "localRef": "a.zim"}, + {"app": "kiwix", "localRef": "gone.zim"}, + {"app": "nope", "localRef": "x"}, + ]) + _, body = share.manifest(sid, token) + self.assertEqual([i["identity"]["localRef"] for i in body["items"]], ["a.zim"]) + reasons = {m["reason"] for m in body["missing"]} + self.assertEqual(reasons, {"not found", "unknown app"}) + self.assertEqual(body["totalBytes"], 10) + + def test_manifest_token_enforcement(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + self.assertEqual(share.manifest(sid, None)[0], 401) # no token + self.assertEqual(share.manifest(sid, "")[0], 401) # empty token + self.assertEqual(share.manifest(sid, "wrong")[0], 403) # mismatch + self.assertEqual(share.manifest("deadbeef", token)[0], 404) # unknown session + self.assertEqual(share.manifest(sid, token)[0], 200) # correct + + +class FetchTest(unittest.TestCase): + def test_fetch_direct_item_streams_bytes(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10, "b.zim": 7})], d) + sid, token = _create(share, [ + {"app": "kiwix", "localRef": "a.zim"}, + {"app": "kiwix", "localRef": "b.zim"}, + ]) + status, stream = share.fetch(sid, "0", token) + self.assertEqual(status, 200) + self.assertEqual(len(_drain(stream)), 10) + + def test_fetch_token_enforcement(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + self.assertEqual(share.fetch(sid, "0", None)[0], 401) + self.assertEqual(share.fetch(sid, "0", "wrong")[0], 403) + self.assertEqual(share.fetch("deadbeef", "0", token)[0], 404) + + def test_fetch_unknown_itemref_404(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + self.assertEqual(share.fetch(sid, "9", token)[0], 404) # out of range + self.assertEqual(share.fetch(sid, "x", token)[0], 404) # not an index + + def test_fetch_absent_item_404(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [{"app": "kiwix", "localRef": "gone.zim"}]) + self.assertEqual(share.fetch(sid, "0", token)[0], 404) + + def test_fetch_staging_item_409_then_streams(self): + with tempfile.TemporaryDirectory() as d: + gate = threading.Event() # hold staging open + share, _ops, _reg = _share([_StagingFake({"chan": 120}, gate=gate)], d) + sid, token = _create(share, [{"app": "kolibri", "localRef": "chan"}]) + status, body = share.fetch(sid, "0", token) + self.assertEqual(status, 409) + self.assertEqual(body["state"], "staging") + gate.set() + stream = _fetch_ready(share, sid, "0", token) + self.assertEqual(len(_drain(stream)), 120) + + def test_fetch_staging_error_surfaces_500(self): + with tempfile.TemporaryDirectory() as d: + gate = threading.Event() + gate.set() # let staging run, but it fails + share, _ops, _reg = _share([_StagingFake({"chan": 50}, gate=gate, fail=True)], d) + sid, token = _create(share, [{"app": "kolibri", "localRef": "chan"}]) + end = time.time() + 5 + seen = None + while time.time() < end: + status, body = share.fetch(sid, "0", token) + if status == 500: + seen = body + break + time.sleep(0.01) + self.assertIsNotNone(seen, "staging error never surfaced") + self.assertEqual(seen["state"], "error") + + def test_single_flight_rejects_a_second_concurrent_fetch_of_one_item(self): + # Two open streams on one item would share the child op's temp, and the first close would + # remove it under the second. The gate single-flights: a second fetch of a busy item gets + # 409 busy until the first stream closes. + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + status1, stream1 = share.fetch(sid, "0", token) # claims the slot + self.assertEqual(status1, 200) + status2, body2 = share.fetch(sid, "0", token) # same item still open + self.assertEqual(status2, 409) + self.assertEqual(body2["state"], "busy") + _drain(stream1) # closes the first stream: releases + status3, stream3 = share.fetch(sid, "0", token) # free again + self.assertEqual(status3, 200) + _drain(stream3) + + +class CompleteTest(unittest.TestCase): + def test_complete_closes_cleanly(self): + # Closer: COMPLETE (the receiver has every item). Supersedes any drop-based auto-close. + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10, "b.zim": 7})], d) + sid, token = _create(share, [ + {"app": "kiwix", "localRef": "a.zim"}, + {"app": "kiwix", "localRef": "b.zim"}, + ]) + _drain(share.fetch(sid, "0", token)[1]) + _drain(share.fetch(sid, "1", token)[1]) + status, body = share.complete(sid, token) + self.assertEqual(status, 200) + self.assertTrue(body["ok"]) + self.assertEqual(share.manifest(sid, token)[0], 404) # released + + def test_complete_token_enforcement_and_double_complete(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + self.assertEqual(share.complete(sid, None)[0], 401) + self.assertEqual(share.complete(sid, "wrong")[0], 403) + self.assertEqual(share.complete("deadbeef", token)[0], 404) + self.assertEqual(share.complete(sid, token)[0], 200) + self.assertEqual(share.complete(sid, token)[0], 404) # already closed + + +class DeliveredTest(unittest.TestCase): + def test_delivered_only_on_clean_close_dropped_item_resumes(self): + # delivered is set on a clean close only. A dropped item stays pending and re-fetches + # (the resume model): the gate does not close on a drop. + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10, "b.zim": 7})], d) + sid, token = _create(share, [ + {"app": "kiwix", "localRef": "a.zim"}, + {"app": "kiwix", "localRef": "b.zim"}, + ]) + _drain(share.fetch(sid, "0", token)[1]) # clean -> delivered + _drain_drop(share.fetch(sid, "1", token)[1]) # drop -> not delivered + _, man = share.manifest(sid, token) + delivered = {i["itemRef"]: i["delivered"] for i in man["items"]} + self.assertTrue(delivered["0"]) + self.assertFalse(delivered["1"]) + self.assertEqual(share.manifest(sid, token)[0], 200) # a dropped item did not close it + # Resume: re-fetch the dropped item, this time cleanly. + status, stream = share.fetch(sid, "1", token) + self.assertEqual(status, 200) + _drain(stream) + _, man2 = share.manifest(sid, token) + self.assertTrue({i["itemRef"]: i["delivered"] for i in man2["items"]}["1"]) + + +class CancelTest(unittest.TestCase): + def test_delete_cancels_and_revokes_child_temp(self): + # Closer: CANCEL (DELETE). Fetch starts a staging child op (leaves a temp on disk); the + # cancel revokes the session, the token and the child temp. + with tempfile.TemporaryDirectory() as d: + gate = threading.Event() + gate.set() + share, ops, _reg = _share([_StagingFake({"c0": 20, "c1": 30}, gate=gate)], d) + sid, token = _create(share, [ + {"app": "kolibri", "localRef": "c0"}, + {"app": "kolibri", "localRef": "c1"}, + ]) + share.fetch(sid, "1", token) # 409 staging, but creates the child op + temp + op_id = share._sessions[sid].item_ops["1"] + _wait_ready(ops, op_id) + temp = ops._ops[op_id].temp + self.assertTrue(os.path.isdir(temp)) + status, body = share.delete(sid, token) + self.assertEqual(status, 200) + self.assertTrue(body["ok"]) + self.assertNotIn(sid, share._sessions) + self.assertFalse(os.path.exists(temp)) # child temp revoked + self.assertEqual(share.manifest(sid, token)[0], 404) + + def test_delete_token_enforcement(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + self.assertEqual(share.delete(sid, None)[0], 401) + self.assertEqual(share.delete(sid, "wrong")[0], 403) + self.assertIn(sid, share._sessions) # still open after the rejected closes + self.assertEqual(share.delete(sid, token)[0], 200) + + def test_double_delete_is_404_already_closed(self): + # The receiver and the sender onDestroy may both DELETE; the second sees 404. + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + self.assertEqual(share.delete(sid, token)[0], 200) + self.assertEqual(share.delete(sid, token)[0], 404) + + +class SafetyCapTest(unittest.TestCase): + def test_session_survives_past_the_connection_idle(self): + # The point of the two lifetimes: a paused transfer older than the connection idle TTL + # (10 min) but younger than the session safety cap (hours) must NOT be reclaimed. + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d, cap=3600, conn_ttl=60) + sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + share._sessions[sid].last_seen -= 120 # older than conn idle 60, younger than cap 3600 + share._sweep_capped() + self.assertIn(sid, share._sessions) + self.assertEqual(share.manifest(sid, token)[0], 200) + + def test_safety_cap_reclaims_abandoned_session_and_temp(self): + with tempfile.TemporaryDirectory() as d: + gate = threading.Event() + gate.set() + share, ops, _reg = _share([_StagingFake({"chan": 40}, gate=gate)], d, cap=3600) + sid, token = _create(share, [{"app": "kolibri", "localRef": "chan"}]) + share.fetch(sid, "0", token) # 409 staging, leaves a child op + temp + op_id = share._sessions[sid].item_ops["0"] + _wait_ready(ops, op_id) + temp = ops._ops[op_id].temp + share._sessions[sid].last_seen -= 4000 # abandoned past the cap + share._sweep_capped() + self.assertNotIn(sid, share._sessions) + self.assertFalse(os.path.exists(temp)) + + def test_safety_cap_never_reclaims_a_session_with_an_open_stream(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10, "b.zim": 7})], d, cap=3600) + sid, token = _create(share, [ + {"app": "kiwix", "localRef": "a.zim"}, + {"app": "kiwix", "localRef": "b.zim"}, + ]) + status, stream = share.fetch(sid, "0", token) # item 0 now streaming + self.assertEqual(status, 200) + share._sessions[sid].last_seen -= 4000 # past the cap + share._sweep_capped() + self.assertIn(sid, share._sessions) # streaming: survives + _drain(stream) # closes: clears streaming + share._sessions[sid].last_seen -= 4000 + share._sweep_capped() + self.assertNotIn(sid, share._sessions) # now abandoned: reclaimed + + def test_startup_sweep_clears_in_memory_sessions(self): + with tempfile.TemporaryDirectory() as d: + share, _ops, _reg = _share([_DirectFake({"a.zim": 10})], d) + sid, _token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + self.assertIn(sid, share._sessions) + share.sweep_startup() + self.assertEqual(share._sessions, {}) + + def test_boot_reclaims_a_leftover_share_session_and_its_temp(self): + # Process death: sessions live only in memory, so a fresh process holds none; the + # session's on-disk footprint is its child export temps, reclaimed by the export op's + # startup sweep. This mirrors exactly what serve() runs on boot. + with tempfile.TemporaryDirectory() as d: + gate = threading.Event() + gate.set() + share, ops, _reg = _share([_StagingFake({"chan": 40}, gate=gate)], d) + sid, token = _create(share, [{"app": "kolibri", "localRef": "chan"}]) + share.fetch(sid, "0", token) # leaves a staged cxexport-* temp on disk + op_id = share._sessions[sid].item_ops["0"] + _wait_ready(ops, op_id) + temp = ops._ops[op_id].temp + self.assertTrue(os.path.isdir(temp)) + # Simulate a restart: a new process, new registries on the same staging dir. + new_share, new_ops, _ = _share([_StagingFake({"chan": 40})], d) + new_ops.sweep_startup() # reclaims the orphan temp + new_share.sweep_startup() # holds no sessions + self.assertFalse(os.path.exists(temp)) + self.assertEqual(new_share._sessions, {}) + + +class ShareRouteTest(unittest.TestCase): + """The share routes forward to the ShareOps registry; the gate mechanics live above. Here + we check the routing, the token-via-param wiring (the server reads the X-Share-Token header + and passes it as dispatch's token), and the missing-share guard.""" + + def setUp(self): + self._d = tempfile.TemporaryDirectory() + self.share, self.ops, self.reg = _share([_DirectFake({"a.zim": 10, "b.zim": 7})], self._d.name) + + def tearDown(self): + self._d.cleanup() + + def test_post_create_manifest_fetch_complete(self): + status, body = dispatch(self.reg, "POST", "/v1/share", {}, + {"items": [{"app": "kiwix", "localRef": "a.zim"}, + {"app": "kiwix", "localRef": "b.zim"}]}, + ops=self.ops, share=self.share) + self.assertEqual(status, 200) + sid, token = body["sessionId"], body["token"] + + status, man = dispatch(self.reg, "GET", f"/v1/share/{sid}/manifest", {}, + ops=self.ops, share=self.share, token=token) + self.assertEqual(status, 200) + self.assertEqual(len(man["items"]), 2) + + status, stream = dispatch(self.reg, "GET", f"/v1/share/{sid}/fetch/0", {}, + ops=self.ops, share=self.share, token=token) + self.assertEqual(status, 200) + self.assertIsInstance(stream, ExportStream) + self.assertEqual(len(_drain(stream)), 10) + + status, out = dispatch(self.reg, "POST", f"/v1/share/{sid}/complete", {}, None, + ops=self.ops, share=self.share, token=token) + self.assertEqual(status, 200) + self.assertTrue(out["ok"]) + + def test_delete_route_cancels(self): + sid, token = _create(self.share, [{"app": "kiwix", "localRef": "a.zim"}]) + status, out = dispatch(self.reg, "DELETE", f"/v1/share/{sid}", {}, + ops=self.ops, share=self.share, token=token) + self.assertEqual(status, 200) + self.assertTrue(out["ok"]) + + def test_manifest_without_token_401(self): + sid, _token = _create(self.share, [{"app": "kiwix", "localRef": "a.zim"}]) + status, _ = dispatch(self.reg, "GET", f"/v1/share/{sid}/manifest", {}, + ops=self.ops, share=self.share, token=None) + self.assertEqual(status, 401) + + def test_share_routes_without_share_500(self): + for method, path, body in [ + ("POST", "/v1/share", {"items": [{"app": "kiwix", "localRef": "a.zim"}]}), + ("POST", "/v1/share/abc/complete", None), + ("GET", "/v1/share/abc/manifest", None), + ("GET", "/v1/share/abc/fetch/0", None), + ("DELETE", "/v1/share/abc", None), + ]: + status, _ = dispatch(self.reg, method, path, {}, body, ops=self.ops, share=None, + token="t") + self.assertEqual(status, 500, path) + + +class PreHeaderDropTest(unittest.TestCase): + """A client that disconnects DURING the response header write must still fire the stream's + on_close, so the single-flight slot is released and the safety cap can reclaim the session. + The server writes the headers inside the same try whose finally runs on_close; this checks + that guarantee through the real handler (white-box, because a real-socket drop timed exactly + at the header write is not deterministic).""" + + def test_pre_header_disconnect_releases_slot_so_cap_reclaims(self): + with tempfile.TemporaryDirectory() as d: + share, ops, reg = _share([_DirectFake({"a.zim": 10})], d, cap=3600) + sid, token = _create(share, [{"app": "kiwix", "localRef": "a.zim"}]) + status, stream = share.fetch(sid, "0", token) # claims the single-flight slot + self.assertEqual(status, 200) + self.assertIn("0", share._sessions[sid].streaming) + + # A handler whose header write fails, as if the client dropped during the headers. + handler_cls = make_handler(reg, ops, share) + + class _FailHeaders(handler_cls): + def send_response(self, *_a, **_k): + raise BrokenPipeError("client gone during headers") + + handler = _FailHeaders.__new__(_FailHeaders) # skip the socket __init__ + handler.close_connection = False + with self.assertRaises(BrokenPipeError): + handler._stream(200, stream) + + # on_close ran in the finally despite the header failure: the slot is released and the + # item is NOT delivered (the body never streamed). + self.assertNotIn("0", share._sessions[sid].streaming) + _, man = share.manifest(sid, token) + self.assertFalse(man["items"][0]["delivered"]) + # With the slot released, the cap reclaims the otherwise-abandoned session. + share._sessions[sid].last_seen -= 4000 + share._sweep_capped() + self.assertNotIn(sid, share._sessions) + + +class ShareHttpTest(unittest.TestCase): + """End-to-end over a real socket: POST create, GET manifest, GET fetch (chunked stream + decoded by a real client), POST complete, all with the token in the X-Share-Token header.""" + + def setUp(self): + self._d = tempfile.TemporaryDirectory() + self.payload = b"x" * 4096 + share, ops, reg = _share([_DirectFake({"a.zim": len(self.payload), "b.zim": 7})], self._d.name) + self.httpd = ThreadingHTTPServer(("127.0.0.1", 0), make_handler(reg, ops, share)) + self.port = self.httpd.server_address[1] + self._t = threading.Thread(target=self.httpd.serve_forever, daemon=True) + self._t.start() + + def tearDown(self): + self.httpd.shutdown() + self.httpd.server_close() + self._d.cleanup() + + def _conn(self): + return http.client.HTTPConnection("127.0.0.1", self.port, timeout=10) + + # Connection: close on each request so the server does not keep a worker thread blocked on + # the next request line, which otherwise logs a reset when the suite tears the server down. + def _req(self, method, path, body=None, token=None): + c = self._conn() + headers = {"Connection": "close"} + if body is not None: + headers["Content-Type"] = "application/json" + if token is not None: + headers["X-Share-Token"] = token + c.request(method, path, body=json.dumps(body) if body is not None else None, headers=headers) + resp = c.getresponse() + status = resp.status + out = resp.read() + c.close() + return status, (json.loads(out) if out else None) + + def test_round_trip_over_http(self): + status, created = self._req( + "POST", "/v1/share", + {"items": [{"app": "kiwix", "localRef": "a.zim"}, {"app": "kiwix", "localRef": "b.zim"}]}) + self.assertEqual(status, 200) + sid, token = created["sessionId"], created["token"] + self.assertEqual(created["manifestUrl"], f"/v1/share/{sid}/manifest") + + status, man = self._req("GET", f"/v1/share/{sid}/manifest", token=token) + self.assertEqual(status, 200) + self.assertEqual(man["totalBytes"], len(self.payload) + 7) + self.assertEqual([i["itemRef"] for i in man["items"]], ["0", "1"]) + + c = self._conn() + c.request("GET", f"/v1/share/{sid}/fetch/0", headers={"Connection": "close", "X-Share-Token": token}) + resp = c.getresponse() + self.assertEqual(resp.status, 200) + got = resp.read() + c.close() + self.assertEqual(got, self.payload) + + # A clean fetch marks the item delivered in the manifest (resume skips it). + status, man2 = self._req("GET", f"/v1/share/{sid}/manifest", token=token) + delivered = {i["itemRef"]: i["delivered"] for i in man2["items"]} + self.assertTrue(delivered["0"]) + self.assertFalse(delivered["1"]) + + status, out = self._req("POST", f"/v1/share/{sid}/complete", token=token) + self.assertEqual(status, 200) + self.assertTrue(out["ok"]) + + status, _ = self._req("GET", f"/v1/share/{sid}/manifest", token=token) + self.assertEqual(status, 404) # released + + def test_token_in_query_is_not_accepted(self): + # FIX: the token must NOT be read from the URL (it would land in access logs). A query + # token with no header is unauthenticated. + _, created = self._req("POST", "/v1/share", {"items": [{"app": "kiwix", "localRef": "a.zim"}]}) + sid, token = created["sessionId"], created["token"] + status, _ = self._req("GET", f"/v1/share/{sid}/manifest?token={token}") # no header + self.assertEqual(status, 401) + + def test_manifest_wrong_token_header_403(self): + _, created = self._req("POST", "/v1/share", {"items": [{"app": "kiwix", "localRef": "a.zim"}]}) + sid = created["sessionId"] + status, _ = self._req("GET", f"/v1/share/{sid}/manifest", token="wrong") + self.assertEqual(status, 403) + + +if __name__ == "__main__": + unittest.main()