From fe41dd502837126a6aec4fb52459ae09cb5e9124 Mon Sep 17 00:00:00 2001 From: flujo-app <300233937+flujo-app@users.noreply.github.com> Date: Sun, 4 Oct 2026 06:46:48 -0500 Subject: [PATCH 1/6] Add held Factory receiver v2 ASGI observation seam --- docs/factory_receiver_v2_candidate.md | 55 ++++ src/drift/api/server.py | 74 +++++- src/drift/factory_receiver.py | 221 ++++++++++++++++ tests/test_factory_receiver_v2.py | 361 ++++++++++++++++++++++++++ 4 files changed, 704 insertions(+), 7 deletions(-) create mode 100644 docs/factory_receiver_v2_candidate.md create mode 100644 src/drift/factory_receiver.py create mode 100644 tests/test_factory_receiver_v2.py diff --git a/docs/factory_receiver_v2_candidate.md b/docs/factory_receiver_v2_candidate.md new file mode 100644 index 000000000..fd865cf65 --- /dev/null +++ b/docs/factory_receiver_v2_candidate.md @@ -0,0 +1,55 @@ +# Factory receiver v2 candidate (offline) + +`create_factory_receiver_app_v2(receiver_profile=...)` is an opt-in route to the +existing ordinary distributed text path. It does not change the v1 +`create_factory_app(factory_admission=...)` callback. The two adapters cannot be +configured together. A v2 adapter must return `FactoryAdmissionV2`; a bare v1 +admission is denied. + +The receiver supplies a `FactoryIngressObservationV2` to the host. It contains +the actual de-chunked bytes returned by ASGI `Request.body()` (maximum 32,768 +bytes), their SHA-256, the independently computed digest of the Pydantic +normalized body, and the receiver-observed `POST` method, exact route, and +allowlisted header values. Query strings, encoded route variants, duplicate +projected headers, malformed headers, and size overflow are denied before the +host's transport or Original-admission callback. Authorization, host, forwarded, +and cookie headers are excluded from the projection. The projected names are +the ten OpenAI SDK headers in the real Flow capture plus the four routing names +`openai-organization`, `openai-project`, `http-referer`, and `x-title` if present. + +These are **receiver-observed ASGI values**, not an authenticated copy of the +SDK-final wire. A proxy can change the path, body, headers, scheme, and host; +sorting the receiver's selected headers cannot prove the sender's ordered +ten-header projection. The host must authenticate the separate Original issuer +and sender commitment, verify the actual sender principal and credential +generation, bind the call and slot, and compare the two projections under a +declared and verified proxy transformation policy. It must also authenticate +the deployed receiver identity, build and generation. The v2 profile's opaque +in-process marker and transport verifier receive only host-populated ASGI scope +extensions; caller headers and the API bearer cannot create that marker. The +host must issue it only after checking the transport channel and generation. +The source checks marker identity and generation before calling the Original +adapter. The returned admission binds the observed projection digest and +transport fields and continues to use the existing fresh authority and +one-attempt dispatch claim. These checks cannot turn an unverified host +callback into a trust root. The admission's fresh authority callback must also +recheck Original and transport validity immediately before dispatch. + +The real Flow fixture has a 299-byte SDK body with SHA-256 +`b86e125b6a031a226acc03147b5d865402ad149a621626fcdbf7629f5541fd03`. +PR35's Python-normalized body has 307 bytes and SHA-256 +`ab288a143d34de92156f01d0d4c4f5d05bbdc51eda4a335e16c8816205cd08b7`. +The separate FACTORY pure held v3 contract at `697a27d` currently constructs a +305-byte JavaScript-normalized body (SHA-256 +`f1a06120ca623ed9dafcdd31c1ec6e54468071005306c481a6d071a388237b44`) +because it serializes `temperature` as `1`, while Python emits `1.0`. Its +`HELD_BEFORE_POST` expected observation includes authenticated sender and +credential fields and the ten SDK-final headers; it cannot be filled by this +post-ingress ASGI observation. A separate authenticated post-POST receiver +receipt and a corrected normalization commitment are required for a join. + +FastAPI parses JSON before this route's `Request.body()` bound runs, so the +deployment ingress must enforce the same raw-body limit before framework +parsing. This candidate has no Original issuer, authenticated transport, +deployed receiver receipt, or physical send. The offline test host supplies +synthetic values and holds the dispatch claim. diff --git a/src/drift/api/server.py b/src/drift/api/server.py index c14de9608..a7d2fb0fc 100644 --- a/src/drift/api/server.py +++ b/src/drift/api/server.py @@ -31,6 +31,14 @@ from transformers import StoppingCriteria, StoppingCriteriaList, TextIteratorStreamer from drift.factory_admission import FactoryAdmission, RequestAdmissionDenied, synchronous_result +from drift.factory_receiver import ( + FactoryAdmissionV2, + FactoryIngressHeadersInvalid, + FactoryIngressTooLarge, + FactoryReceiverProfileV2, + HostVerifiedTransport, + observe_factory_ingress_v2, +) from drift.node.model_manager import ( AmbiguousModelError, AutoModelUnavailableError, @@ -186,6 +194,17 @@ def create_factory_app(*, factory_admission: Callable[[dict], FactoryAdmission], return create_app(factory_admission=factory_admission, **kwargs) +def create_factory_receiver_app_v2(*, receiver_profile: FactoryReceiverProfileV2, **kwargs) -> FastAPI: + """Opt in to observed ASGI ingress plus host-verified Original admission. + + The profile does not itself verify transport, Original provenance, or the + sender's SDK-final wire. Those are obligations of the embedding host. + """ + if type(receiver_profile) is not FactoryReceiverProfileV2: + raise ValueError("Factory receiver v2 requires a host-owned profile") + return create_app(factory_receiver_v2=receiver_profile, **kwargs) + + def create_app( model=None, tokenizer=None, @@ -200,6 +219,7 @@ def create_app( route_outcome_observer: Optional[Callable[..., None]] = None, request_timeout: float = 900.0, factory_admission: Optional[Callable[[dict], FactoryAdmission]] = None, + factory_receiver_v2: Optional[FactoryReceiverProfileV2] = None, ) -> FastAPI: """Create the OpenAI-compatible application. @@ -223,14 +243,18 @@ def create_app( raise ValueError("pass either model_manager or the single-model arguments, not both") if sum((bool(api_keys), api_key_verifier is not None, api_key_identifier is not None)) > 1: raise ValueError("pass only one API key authentication method") - if factory_admission is not None and ( - not callable(factory_admission) + if factory_admission is not None and factory_receiver_v2 is not None: + raise ValueError("choose one Factory receiver profile") + if factory_receiver_v2 is not None and type(factory_receiver_v2) is not FactoryReceiverProfileV2: + raise ValueError("Factory receiver v2 requires a host-owned profile") + if (factory_admission is not None or factory_receiver_v2 is not None) and ( + (factory_admission is not None and not callable(factory_admission)) or not (api_keys or api_key_verifier is not None or api_key_identifier is not None) ): raise ValueError("Factory admission requires a trusted adapter and API authentication") app = FastAPI(title="DRIFT-LLM OpenAI-compatible API") - if factory_admission is not None: + if factory_admission is not None or factory_receiver_v2 is not None: # A slash redirect would precede original admission and allow a body resend. app.router.redirect_slashes = False semaphore = asyncio.Semaphore(max_concurrent) @@ -256,8 +280,9 @@ def check_auth(request: Request) -> str | None: raise HTTPException(status_code=401, detail="Invalid API key") return identity - def request_context(body, request): + async def request_context(body, request, route): admission = None + dispatch_guard = None original_body = body.model_dump(exclude_none=True) if factory_admission is not None: try: @@ -267,6 +292,41 @@ def request_context(body, request): if type(admission) is not FactoryAdmission: raise RequestAdmissionDenied() admission.require_current(original_body) + dispatch_guard = lambda: admission.require_current(original_body) + except Exception: + raise HTTPException(status_code=403, detail="Original request admission denied") from None + elif factory_receiver_v2 is not None: + try: + # FastAPI has parsed the request by this point. The bound below + # caps what reaches the host adapter; deployment ingress must + # separately cap request size before framework parsing. + raw_body = await request.body() + observation = observe_factory_ingress_v2(request.scope, raw_body, original_body, route) + except FactoryIngressTooLarge: + raise HTTPException(status_code=413, detail="Factory ingress body too large") from None + except FactoryIngressHeadersInvalid: + raise HTTPException(status_code=431, detail="Factory ingress headers invalid") from None + except Exception: + raise HTTPException(status_code=403, detail="Original request admission denied") from None + try: + # Scope extensions are supplied by trusted host middleware. The + # HTTP bearer and projected request headers do not mint identity. + extensions = request.scope.get("extensions", {}) + if type(extensions) is not dict: + raise RequestAdmissionDenied() + transport = synchronous_result(factory_receiver_v2.transport_verifier(extensions)) + if ( + type(transport) is not HostVerifiedTransport + or transport.host_capability is not factory_receiver_v2.host_capability + or transport.generation_sha256 != factory_receiver_v2.generation_sha256 + ): + raise RequestAdmissionDenied() + decision = synchronous_result(factory_receiver_v2.original_admission(observation, transport)) + if type(decision) is not FactoryAdmissionV2: + raise RequestAdmissionDenied() + decision.require_current(observation, original_body, transport) + admission = decision.admission + dispatch_guard = lambda: decision.require_current(observation, original_body, transport) except Exception: raise HTTPException(status_code=403, detail="Original request admission denied") from None caller_id = check_auth(request) @@ -284,7 +344,7 @@ def request_context(body, request): caller_id=caller_id, request_id=admission.request_id, single_attempt=True, - dispatch_guard=lambda: admission.require_current(original_body), + dispatch_guard=dispatch_guard, dispatch_claim=admission.claim_dispatch, ) @@ -546,7 +606,7 @@ async def chat_completions(body: ChatCompletionRequest, request: Request): raise HTTPException(status_code=400, detail="n > 1 is not supported") if body.stream_options is not None and not body.stream: raise HTTPException(status_code=400, detail="stream_options requires stream=true") - context = request_context(body, request) + context = await request_context(body, request, "/v1/chat/completions") loaded = await load_with_budget(body.model, context) if context.single_attempt and ( loaded.descriptor.manifest_digest != body.model @@ -632,7 +692,7 @@ async def completions(body: CompletionRequest, request: Request): raise HTTPException(status_code=400, detail="n > 1 is not supported") if body.stream_options is not None and not body.stream: raise HTTPException(status_code=400, detail="stream_options requires stream=true") - context = request_context(body, request) + context = await request_context(body, request, "/v1/completions") loaded = await load_with_budget(body.model, context) if context.single_attempt and ( loaded.descriptor.manifest_digest != body.model diff --git a/src/drift/factory_receiver.py b/src/drift/factory_receiver.py new file mode 100644 index 000000000..deb94cc0f --- /dev/null +++ b/src/drift/factory_receiver.py @@ -0,0 +1,221 @@ +"""Opt-in Factory receiver observation; provenance belongs to the embedding host. + +These values describe what this ASGI app received. A reverse proxy may have +changed the sender's method, path, headers, or body. Neither this observation +nor an HTTP bearer authenticates an Original request. The host must compare it +with a separately authenticated sender commitment and receiver generation. +""" + +import hashlib +import json +import re +from dataclasses import dataclass, field +from typing import Callable, Mapping + +from drift.factory_admission import FactoryAdmission, RequestAdmissionDenied, request_body_digest + +MAX_FACTORY_RAW_BODY_BYTES = 32_768 +MAX_FACTORY_PROJECTED_HEADER_BYTES = 4096 +MAX_FACTORY_TOTAL_HEADER_BYTES = 16_384 +MAX_FACTORY_HEADER_COUNT = 64 +_PROJECTED_HEADERS = frozenset( + { + "accept", + "content-type", + "http-referer", + "openai-organization", + "openai-project", + "user-agent", + "x-title", + "x-stainless-arch", + "x-stainless-lang", + "x-stainless-os", + "x-stainless-package-version", + "x-stainless-retry-count", + "x-stainless-runtime", + "x-stainless-runtime-version", + } +) + + +class FactoryIngressTooLarge(RequestAdmissionDenied): + pass + + +class FactoryIngressHeadersInvalid(RequestAdmissionDenied): + pass + + +@dataclass(frozen=True) +class HostVerifiedTransport: + """Opaque host-issued identity after verification outside HTTP request fields. + + Construction is not verification. Only trusted host middleware may issue a + value from a verified channel, and the Original adapter must check the + principal, channel binding and receiver generation against its own record. + """ + + host_capability: object = field(repr=False, compare=False) + principal_id: str + channel_binding_sha256: str = field(repr=False) + generation_sha256: str + + def __post_init__(self): + if ( + type(self.host_capability) is not object + or type(self.principal_id) is not str + or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}", self.principal_id) is None + or type(self.channel_binding_sha256) is not str + or re.fullmatch(r"[0-9a-f]{64}", self.channel_binding_sha256) is None + or type(self.generation_sha256) is not str + or re.fullmatch(r"[0-9a-f]{64}", self.generation_sha256) is None + ): + raise ValueError("host transport identity is malformed") + + +@dataclass(frozen=True) +class FactoryIngressObservationV2: + """Bounded receiver-observed ASGI projection, not a sender-wire proof.""" + + format: str + schema_version: int + method: str + route: str + headers: tuple[tuple[str, str], ...] + raw_body: bytes = field(repr=False) + raw_body_sha256: str + normalized_body_sha256: str + + def digest(self) -> str: + """Bind one receiver observation; the digest is not provenance.""" + projection = { + "format": self.format, + "schemaVersion": self.schema_version, + "method": self.method, + "route": self.route, + "headers": self.headers, + "rawBodyBytes": len(self.raw_body), + "rawBodySha256": self.raw_body_sha256, + "normalizedBodySha256": self.normalized_body_sha256, + } + wire = json.dumps(projection, ensure_ascii=False, allow_nan=False, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(wire.encode("utf-8")).hexdigest() + + +@dataclass(frozen=True) +class FactoryAdmissionV2: + """Host's explicit v2 admission, bound to observation and transport. + + The host must authenticate the separate Original issuer and sender wire; + these fields only prevent accidental reuse or downgrade of that decision. + """ + + admission: FactoryAdmission + observation_sha256: str + principal_id: str + channel_binding_sha256: str + generation_sha256: str + + def require_current(self, observation: FactoryIngressObservationV2, body: dict, transport: HostVerifiedTransport): + if ( + type(self.admission) is not FactoryAdmission + or type(self.observation_sha256) is not str + or re.fullmatch(r"[0-9a-f]{64}", self.observation_sha256) is None + or self.observation_sha256 != observation.digest() + or self.principal_id != transport.principal_id + or self.channel_binding_sha256 != transport.channel_binding_sha256 + or self.generation_sha256 != transport.generation_sha256 + ): + raise RequestAdmissionDenied() + self.admission.require_current(body) + + +@dataclass(frozen=True) +class FactoryReceiverProfileV2: + """Host-owned opt-in bridge to authenticated Original admission. + + ``transport_verifier`` receives *only* trusted ASGI scope extensions, not + request headers or bearer credentials. The host must set those extensions + after verifying its channel and deployed generation. ``original_admission`` + authenticates the separate Original record, compares receiver-observed and + sender-committed projections, and returns an explicit FactoryAdmissionV2. + """ + + generation_sha256: str + host_capability: object = field(repr=False) + transport_verifier: Callable[[Mapping[str, object]], HostVerifiedTransport] = field(repr=False) + original_admission: Callable[[FactoryIngressObservationV2, HostVerifiedTransport], FactoryAdmissionV2] = field( + repr=False + ) + schema_version: int = 2 + + def __post_init__(self): + if ( + type(self.schema_version) is not int + or self.schema_version != 2 + or type(self.host_capability) is not object + or type(self.generation_sha256) is not str + or re.fullmatch(r"[0-9a-f]{64}", self.generation_sha256) is None + or not callable(self.transport_verifier) + or not callable(self.original_admission) + ): + raise ValueError("Factory receiver v2 requires host verification and admission") + + +def observe_factory_ingress_v2(scope: dict, raw_body: bytes, normalized_body: dict, route: str): + """Reject unbounded/ambiguous values before any trusted host callback.""" + if type(raw_body) is not bytes or len(raw_body) > MAX_FACTORY_RAW_BODY_BYTES: + raise FactoryIngressTooLarge() + raw_path = scope.get("raw_path", route.encode("ascii")) + if ( + scope.get("method") != "POST" + or scope.get("path") != route + or raw_path != route.encode("ascii") + or scope.get("query_string", b"") != b"" + ): + raise RequestAdmissionDenied() + headers = scope.get("headers") + if type(headers) is not list or len(headers) > MAX_FACTORY_HEADER_COUNT: + raise FactoryIngressHeadersInvalid() + projected = {} + projected_bytes = 0 + total_header_bytes = 0 + for pair in headers: + if type(pair) not in (tuple, list) or len(pair) != 2: + raise FactoryIngressHeadersInvalid() + name, value = pair + if type(name) is not bytes or type(value) is not bytes or len(name) > 128 or len(value) > 4096: + raise FactoryIngressHeadersInvalid() + total_header_bytes += len(name) + len(value) + if total_header_bytes > MAX_FACTORY_TOTAL_HEADER_BYTES: + raise FactoryIngressHeadersInvalid() + try: + decoded_name = name.decode("ascii").lower() + except UnicodeDecodeError: + raise FactoryIngressHeadersInvalid() from None + if re.fullmatch(r"[!#$%&'*+.^_`|~0-9a-z-]+", decoded_name) is None or b"\r" in value or b"\n" in value: + raise FactoryIngressHeadersInvalid() + if decoded_name not in _PROJECTED_HEADERS: + continue + if decoded_name in projected or len(value) > 1024: + raise FactoryIngressHeadersInvalid() + try: + decoded_value = value.decode("ascii") + except UnicodeDecodeError: + raise FactoryIngressHeadersInvalid() from None + if any(ord(character) < 32 or ord(character) > 126 for character in decoded_value): + raise FactoryIngressHeadersInvalid() + projected_bytes += len(name) + len(value) + if projected_bytes > MAX_FACTORY_PROJECTED_HEADER_BYTES: + raise FactoryIngressHeadersInvalid() + projected[decoded_name] = decoded_value + return FactoryIngressObservationV2( + format="communityai-factory-asgi-ingress-observation", + schema_version=2, + method="POST", + route=route, + headers=tuple(sorted(projected.items())), + raw_body=raw_body, + raw_body_sha256=hashlib.sha256(raw_body).hexdigest(), + normalized_body_sha256=request_body_digest(normalized_body), + ) diff --git a/tests/test_factory_receiver_v2.py b/tests/test_factory_receiver_v2.py new file mode 100644 index 000000000..5ecd0a1e8 --- /dev/null +++ b/tests/test_factory_receiver_v2.py @@ -0,0 +1,361 @@ +"""Offline receiver-observed ASGI ingress; no Original issuer or physical send.""" + +import hashlib +import json +import unittest + +import httpx +from test_factory_sdk_asgi_hold import ( + FLOW_NORMALIZED_SHA256, + FLOW_SDK_BODY_SHA256, + FLOW_SDK_BODY_UTF8, + MANIFEST, + HeldPeer, +) + +from drift.api.server import create_factory_receiver_app_v2 +from drift.factory_admission import FactoryAdmission, RequestAdmissionDenied +from drift.factory_receiver import FactoryAdmissionV2, FactoryReceiverProfileV2, HostVerifiedTransport +from drift.node.model_manager import ModelDescriptor, ModelManager, ModelRuntime + +EXPECTED_HEADERS = ( + ("accept", "application/json"), + ("content-type", "application/json"), + ("user-agent", "OpenAI/JS 7.3.0"), + ("x-stainless-arch", "x64"), + ("x-stainless-lang", "js"), + ("x-stainless-os", "Windows"), + ("x-stainless-package-version", "7.3.0"), + ("x-stainless-retry-count", "0"), + ("x-stainless-runtime", "node"), + ("x-stainless-runtime-version", "v24.19.0"), +) + + +class HostExtension: + """Test-only outer ASGI host that has already verified a transport.""" + + def __init__(self, app, identity): + self.app = app + self.identity = identity + + async def __call__(self, scope, receive, send): + scope = dict(scope) + scope["extensions"] = dict(scope.get("extensions", {})) + if self.identity is not None: + scope["extensions"]["verified_factory_transport"] = self.identity + await self.app(scope, receive, send) + + +class FactoryReceiverV2Tests(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.manager = ModelManager() + self.peer = HeldPeer() + self.events = [] + self.observations = [] + self.claims = 0 + self.manager.register( + ModelDescriptor("Fixture", aliases=(MANIFEST,), manifest_digest=MANIFEST), + lambda: ModelRuntime(None, None, text_client=self.peer), + ) + self.host_capability = object() + self.transport = HostVerifiedTransport( + host_capability=self.host_capability, + principal_id="fixture-flow-owner", + channel_binding_sha256="a" * 64, + generation_sha256="b" * 64, + ) + + async def asyncTearDown(self): + self.manager.shutdown() + + def app(self, *, identity=True, verifier=None, original_admission=None): + def verify_transport(extensions): + self.events.append("transport") + return extensions.get("verified_factory_transport") + + def admit(observation, transport): + self.events.append("original") + self.observations.append(observation) + if ( + transport != self.transport + or observation.format != "communityai-factory-asgi-ingress-observation" + or observation.schema_version != 2 + or observation.method != "POST" + or observation.route != "/v1/chat/completions" + or observation.headers != EXPECTED_HEADERS + or observation.raw_body != FLOW_SDK_BODY_UTF8 + or observation.raw_body_sha256 != FLOW_SDK_BODY_SHA256 + or observation.normalized_body_sha256 != FLOW_NORMALIZED_SHA256 + ): + raise RequestAdmissionDenied() + admission = FactoryAdmission( + request_id="d" * 32, + body_digest=FLOW_NORMALIZED_SHA256, + manifest_digest=MANIFEST, + num_blocks=1, + confidentiality_class="ordinary", + class_version=1, + authority=lambda: None, + route_snapshot=lambda: { + "manifest_digest": MANIFEST, + "status": "complete", + "covered_blocks": 1, + "total_blocks": 1, + "missing_blocks": [], + "chat_ready": True, + "peer_count": 1, + "last_updated_age": 0.0, + "replica_counts": [1], + }, + claim_dispatch=self.hold_claim, + ) + return FactoryAdmissionV2( + admission=admission, + observation_sha256=observation.digest(), + principal_id=transport.principal_id, + channel_binding_sha256=transport.channel_binding_sha256, + generation_sha256=transport.generation_sha256, + ) + + def bearer(candidate): + self.events.append("bearer") + return candidate == "fixture-key" + + app = create_factory_receiver_app_v2( + model_manager=self.manager, + api_key_verifier=bearer, + receiver_profile=FactoryReceiverProfileV2( + generation_sha256="b" * 64, + host_capability=self.host_capability, + transport_verifier=verify_transport if verifier is None else verifier, + original_admission=admit if original_admission is None else original_admission, + ), + ) + return HostExtension(app, self.transport if identity else None) + + def hold_claim(self): + self.claims += 1 + raise RequestAdmissionDenied() + + async def post(self, raw=FLOW_SDK_BODY_UTF8, *, app=None, headers=None, path="/v1/chat/completions"): + if app is None: + app = self.app() + if headers is None: + headers = { + "accept": "application/json", + "content-type": "application/json", + "user-agent": "OpenAI/JS 7.3.0", + "x-stainless-arch": "x64", + "x-stainless-lang": "js", + "x-stainless-os": "Windows", + "x-stainless-package-version": "7.3.0", + "x-stainless-retry-count": "0", + "x-stainless-runtime": "node", + "x-stainless-runtime-version": "v24.19.0", + "authorization": "Bearer fixture-key", + } + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client: + return await client.post(path, content=raw, headers=headers) + + async def test_exact_real_flow_body_and_receiver_headers_reach_host_before_bearer(self): + self.assertEqual(len(FLOW_SDK_BODY_UTF8), 299) + self.assertEqual(hashlib.sha256(FLOW_SDK_BODY_UTF8).hexdigest(), FLOW_SDK_BODY_SHA256) + response = await self.post() + self.assertEqual(response.status_code, 200) + self.assertEqual(self.events, ["transport", "original", "bearer"]) + observation = self.observations[0] + self.assertEqual(observation.raw_body, FLOW_SDK_BODY_UTF8) + self.assertNotEqual(observation.raw_body_sha256, observation.normalized_body_sha256) + self.assertEqual(observation.headers, EXPECTED_HEADERS) + self.assertEqual(self.claims, 1) + self.assertEqual(self.peer.after_claim, 0) + self.assertIn('"code": "request_not_dispatched"', response.text) + + async def test_headers_and_bearer_cannot_supply_missing_host_transport(self): + response = await self.post( + app=self.app(identity=False), + headers={ + "content-type": "application/json", + "authorization": "Bearer fixture-key", + "x-factory-principal": "fixture-flow-owner", + "x-factory-generation": "b" * 64, + }, + ) + self.assertEqual(response.status_code, 403) + self.assertEqual(self.events, ["transport"]) + self.assertEqual(self.observations, []) + self.assertEqual(self.claims, 0) + self.assertEqual(self.peer.entries, 0) + + async def test_stale_host_generation_refuses_before_original_or_bearer(self): + stale = HostVerifiedTransport(self.host_capability, "fixture-flow-owner", "a" * 64, "c" * 64) + app = HostExtension(self.app(identity=False), stale) + response = await self.post(app=app) + self.assertEqual(response.status_code, 403) + self.assertEqual(self.events, ["transport"]) + self.assertEqual(self.observations, []) + + async def test_foreign_host_capability_refuses_before_original_or_bearer(self): + foreign = HostVerifiedTransport(object(), "fixture-flow-owner", "a" * 64, "b" * 64) + app = HostExtension(self.app(identity=False), foreign) + response = await self.post(app=app) + self.assertEqual(response.status_code, 403) + self.assertEqual(self.events, ["transport"]) + self.assertEqual(self.observations, []) + + async def test_bare_v1_admission_cannot_downgrade_v2_profile(self): + def bare_v1(observation, transport): + self.events.append("original") + return FactoryAdmission( + request_id="d" * 32, + body_digest=FLOW_NORMALIZED_SHA256, + manifest_digest=MANIFEST, + num_blocks=1, + confidentiality_class="ordinary", + class_version=1, + authority=lambda: None, + route_snapshot=lambda: {}, + claim_dispatch=self.hold_claim, + ) + + response = await self.post(app=self.app(original_admission=bare_v1)) + self.assertEqual(response.status_code, 403) + self.assertEqual(self.events, ["transport", "original"]) + self.assertEqual(self.claims, 0) + self.assertEqual(self.peer.entries, 0) + + async def test_v2_admission_bound_to_another_observation_is_denied(self): + def wrong_binding(observation, transport): + self.events.append("original") + return FactoryAdmissionV2( + admission=FactoryAdmission( + request_id="d" * 32, + body_digest=FLOW_NORMALIZED_SHA256, + manifest_digest=MANIFEST, + num_blocks=1, + confidentiality_class="ordinary", + class_version=1, + authority=lambda: None, + route_snapshot=lambda: {}, + claim_dispatch=self.hold_claim, + ), + observation_sha256="0" * 64, + principal_id=transport.principal_id, + channel_binding_sha256=transport.channel_binding_sha256, + generation_sha256=transport.generation_sha256, + ) + + response = await self.post(app=self.app(original_admission=wrong_binding)) + self.assertEqual(response.status_code, 403) + self.assertEqual(self.events, ["transport", "original"]) + self.assertEqual(self.claims, 0) + + async def test_same_normalized_body_with_changed_raw_bytes_refuses_before_bearer(self): + changed = FLOW_SDK_BODY_UTF8.replace(b'"temperature":1,', b'"temperature":1.0,') + self.assertNotEqual(changed, FLOW_SDK_BODY_UTF8) + response = await self.post(changed) + self.assertEqual(response.status_code, 403) + self.assertEqual(self.events, ["transport", "original"]) + self.assertEqual(self.observations[0].normalized_body_sha256, FLOW_NORMALIZED_SHA256) + self.assertNotEqual(self.observations[0].raw_body_sha256, FLOW_SDK_BODY_SHA256) + self.assertEqual(self.claims, 0) + + async def test_added_routing_header_or_query_refuses_before_bearer(self): + response = await self.post( + headers={ + "accept": "application/json", + "content-type": "application/json", + "user-agent": "OpenAI/JS 7.3.0", + "x-stainless-arch": "x64", + "x-stainless-lang": "js", + "x-stainless-os": "Windows", + "x-stainless-package-version": "7.3.0", + "x-stainless-retry-count": "0", + "x-stainless-runtime": "node", + "x-stainless-runtime-version": "v24.19.0", + "openai-organization": "uncommitted", + "authorization": "Bearer fixture-key", + } + ) + self.assertEqual(response.status_code, 403) + self.assertEqual(self.events, ["transport", "original"]) + self.assertIn(("openai-organization", "uncommitted"), self.observations[0].headers) + self.events.clear() + self.observations.clear() + response = await self.post(path="/v1/chat/completions?extra=1") + self.assertEqual(response.status_code, 403) + self.assertEqual(self.events, []) + self.assertEqual(self.observations, []) + + async def test_decoded_route_with_encoded_raw_path_refuses_before_host_callbacks(self): + class EncodedRawPath: + def __init__(self, app): + self.app = app + + async def __call__(self, scope, receive, send): + scope = dict(scope) + scope["raw_path"] = b"/v1/chat%2Fcompletions" + await self.app(scope, receive, send) + + response = await self.post(app=EncodedRawPath(self.app())) + self.assertEqual(response.status_code, 403) + self.assertEqual(self.events, []) + + async def test_body_overflow_refuses_before_host_or_bearer_callbacks(self): + expanded = json.loads(FLOW_SDK_BODY_UTF8) + expanded["messages"][0]["content"] = "x" * 32_768 + response = await self.post(json.dumps(expanded).encode("utf-8")) + self.assertEqual(response.status_code, 413) + self.assertEqual(self.events, []) + self.assertEqual(self.observations, []) + + async def test_duplicate_or_oversize_projected_header_refuses_before_callbacks(self): + for headers in ( + [("content-type", "application/json"), ("Content-Type", "application/json")], + {"content-type": "application/json", "user-agent": "x" * 1025}, + [("content-type", "application/json")] + [(f"x-unprojected-{index}", "x") for index in range(65)], + ): + with self.subTest(headers=headers): + response = await self.post(headers=headers) + self.assertEqual(response.status_code, 431) + self.assertEqual(self.events, []) + self.assertEqual(self.observations, []) + + async def test_async_host_verifier_is_rejected_without_running_it(self): + effects = [] + + async def invalid_verifier(extensions): + effects.append(True) + return self.transport + + response = await self.post(app=self.app(verifier=invalid_verifier)) + self.assertEqual(response.status_code, 403) + self.assertEqual(effects, []) + self.assertEqual(self.events, []) + + async def test_slash_route_has_no_redirect_or_original_callback(self): + response = await self.post(path="/v1/chat/completions/") + self.assertEqual(response.status_code, 404) + self.assertNotIn("location", response.headers) + self.assertEqual(self.events, []) + + def test_v2_requires_host_profile_and_api_authentication(self): + with self.assertRaises(ValueError): + create_factory_receiver_app_v2(model_manager=self.manager, receiver_profile=None, api_keys=["x"]) + with self.assertRaises(ValueError): + create_factory_receiver_app_v2( + model_manager=self.manager, + receiver_profile=FactoryReceiverProfileV2( + "b" * 64, self.host_capability, lambda e: self.transport, lambda o, t: None + ), + ) + with self.assertRaises(ValueError): + create_factory_receiver_app_v2( + model_manager=self.manager, + api_keys=["fixture-key"], + factory_admission=lambda body: None, + receiver_profile=FactoryReceiverProfileV2( + "b" * 64, self.host_capability, lambda e: self.transport, lambda o, t: None + ), + ) From 71a9f1ad66e7a52804fa9154c17eee5e3ab48ca1 Mon Sep 17 00:00:00 2001 From: flujo-app <300233937+flujo-app@users.noreply.github.com> Date: Sun, 4 Oct 2026 07:07:15 -0500 Subject: [PATCH 2/6] Bound Factory v2 body before API parsing --- docs/factory_receiver_v2_candidate.md | 18 ++-- src/drift/api/server.py | 8 +- src/drift/factory_receiver.py | 74 ++++++++++++++++ tests/test_factory_receiver_v2.py | 116 ++++++++++++++++++++++++-- 4 files changed, 201 insertions(+), 15 deletions(-) diff --git a/docs/factory_receiver_v2_candidate.md b/docs/factory_receiver_v2_candidate.md index fd865cf65..a9be3ddf4 100644 --- a/docs/factory_receiver_v2_candidate.md +++ b/docs/factory_receiver_v2_candidate.md @@ -6,10 +6,13 @@ existing ordinary distributed text path. It does not change the v1 configured together. A v2 adapter must return `FactoryAdmissionV2`; a bare v1 admission is denied. -The receiver supplies a `FactoryIngressObservationV2` to the host. It contains -the actual de-chunked bytes returned by ASGI `Request.body()` (maximum 32,768 -bytes), their SHA-256, the independently computed digest of the Pydantic -normalized body, and the receiver-observed `POST` method, exact route, and +An outer ASGI gate caps the two v2 inference routes before FastAPI parses JSON. +It retains at most 32,768 body bytes, stops at the first excess frame with 413, +rejects a declared oversized `Content-Length` before reading any frame, and +replays the accepted bytes unchanged. The receiver then supplies a +`FactoryIngressObservationV2` to the host. It contains those de-chunked bytes, +their SHA-256, the independently computed digest of the Pydantic normalized +body, and the receiver-observed `POST` method, exact route, and allowlisted header values. Query strings, encoded route variants, duplicate projected headers, malformed headers, and size overflow are denied before the host's transport or Original-admission callback. Authorization, host, forwarded, @@ -48,8 +51,9 @@ credential fields and the ten SDK-final headers; it cannot be filled by this post-ingress ASGI observation. A separate authenticated post-POST receiver receipt and a corrected normalization commitment are required for a join. -FastAPI parses JSON before this route's `Request.body()` bound runs, so the -deployment ingress must enforce the same raw-body limit before framework -parsing. This candidate has no Original issuer, authenticated transport, +ASGI delivers whole frames, so an oversized single frame can exist transiently +before the gate rejects it. The deployment ingress still needs a global edge +limit on request/frame size and duration, including paths outside these two +routes. This candidate has no Original issuer, authenticated transport, deployed receiver receipt, or physical send. The offline test host supplies synthetic values and holds the dispatch claim. diff --git a/src/drift/api/server.py b/src/drift/api/server.py index a7d2fb0fc..177dd1cd3 100644 --- a/src/drift/api/server.py +++ b/src/drift/api/server.py @@ -33,6 +33,7 @@ from drift.factory_admission import FactoryAdmission, RequestAdmissionDenied, synchronous_result from drift.factory_receiver import ( FactoryAdmissionV2, + FactoryBoundedBodyMiddleware, FactoryIngressHeadersInvalid, FactoryIngressTooLarge, FactoryReceiverProfileV2, @@ -254,6 +255,8 @@ def create_app( raise ValueError("Factory admission requires a trusted adapter and API authentication") app = FastAPI(title="DRIFT-LLM OpenAI-compatible API") + if factory_receiver_v2 is not None: + app.add_middleware(FactoryBoundedBodyMiddleware) if factory_admission is not None or factory_receiver_v2 is not None: # A slash redirect would precede original admission and allow a body resend. app.router.redirect_slashes = False @@ -297,9 +300,8 @@ async def request_context(body, request, route): raise HTTPException(status_code=403, detail="Original request admission denied") from None elif factory_receiver_v2 is not None: try: - # FastAPI has parsed the request by this point. The bound below - # caps what reaches the host adapter; deployment ingress must - # separately cap request size before framework parsing. + # The v2 outer ASGI gate has already capped the body before + # framework parsing. Retain this check at the adapter boundary. raw_body = await request.body() observation = observe_factory_ingress_v2(request.scope, raw_body, original_body, route) except FactoryIngressTooLarge: diff --git a/src/drift/factory_receiver.py b/src/drift/factory_receiver.py index deb94cc0f..2e1cc0735 100644 --- a/src/drift/factory_receiver.py +++ b/src/drift/factory_receiver.py @@ -18,6 +18,7 @@ MAX_FACTORY_PROJECTED_HEADER_BYTES = 4096 MAX_FACTORY_TOTAL_HEADER_BYTES = 16_384 MAX_FACTORY_HEADER_COUNT = 64 +_BODY_TOO_LARGE_RESPONSE = b'{"detail":"Factory ingress body too large"}' _PROJECTED_HEADERS = frozenset( { "accept", @@ -38,6 +39,20 @@ ) +async def _send_body_too_large(send): + await send( + { + "type": "http.response.start", + "status": 413, + "headers": [ + (b"content-type", b"application/json"), + (b"content-length", str(len(_BODY_TOO_LARGE_RESPONSE)).encode("ascii")), + ], + } + ) + await send({"type": "http.response.body", "body": _BODY_TOO_LARGE_RESPONSE}) + + class FactoryIngressTooLarge(RequestAdmissionDenied): pass @@ -46,6 +61,65 @@ class FactoryIngressHeadersInvalid(RequestAdmissionDenied): pass +class FactoryBoundedBodyMiddleware: + """Cap a v2 ASGI request before FastAPI/Pydantic can parse its body. + + ASGI delivers whole frames, so a single oversized frame is transiently + owned by the server. This middleware never retains it and stops requesting + frames at the first overflow. The deployment edge must bound frame size. + """ + + def __init__(self, app): + self.app = app + + async def __call__(self, scope, receive, send): + if ( + scope.get("type") != "http" + or scope.get("method") != "POST" + or scope.get("path") not in ("/v1/chat/completions", "/v1/completions") + ): + return await self.app(scope, receive, send) + + # A truthful Content-Length lets us refuse without asking ASGI for even + # one body frame. A missing or false length is still caught below. + for name, value in scope.get("headers", []): + if type(name) is bytes and name.lower() == b"content-length" and type(value) is bytes and value.isdigit(): + digits = value.lstrip(b"0") or b"0" + if len(digits) > 5 or int(digits) > MAX_FACTORY_RAW_BODY_BYTES: + await _send_body_too_large(send) + return + + chunks = [] + size = 0 + while True: + message = await receive() + if message["type"] == "http.disconnect": + return + if message["type"] != "http.request" or type(message.get("body", b"")) is not bytes: + raise RequestAdmissionDenied() + chunk = message.get("body", b"") + if len(chunk) > MAX_FACTORY_RAW_BODY_BYTES - size: + await _send_body_too_large(send) + return + if chunk: + chunks.append(chunk) + size += len(chunk) + if message.get("more_body") is not True: + break + + bounded_body = b"".join(chunks) + delivered = False + + async def replay(): + nonlocal delivered + if not delivered: + delivered = True + return {"type": "http.request", "body": bounded_body, "more_body": False} + return await receive() + + return await self.app(scope, replay, send) + + @dataclass(frozen=True) class HostVerifiedTransport: """Opaque host-issued identity after verification outside HTTP request fields. diff --git a/tests/test_factory_receiver_v2.py b/tests/test_factory_receiver_v2.py index 5ecd0a1e8..da45ec5c5 100644 --- a/tests/test_factory_receiver_v2.py +++ b/tests/test_factory_receiver_v2.py @@ -15,7 +15,13 @@ from drift.api.server import create_factory_receiver_app_v2 from drift.factory_admission import FactoryAdmission, RequestAdmissionDenied -from drift.factory_receiver import FactoryAdmissionV2, FactoryReceiverProfileV2, HostVerifiedTransport +from drift.factory_receiver import ( + MAX_FACTORY_RAW_BODY_BYTES, + FactoryAdmissionV2, + FactoryBoundedBodyMiddleware, + FactoryReceiverProfileV2, + HostVerifiedTransport, +) from drift.node.model_manager import ModelDescriptor, ModelManager, ModelRuntime EXPECTED_HEADERS = ( @@ -69,7 +75,7 @@ async def asyncSetUp(self): async def asyncTearDown(self): self.manager.shutdown() - def app(self, *, identity=True, verifier=None, original_admission=None): + def app(self, *, identity=True, verifier=None, original_admission=None, expected_raw=FLOW_SDK_BODY_UTF8): def verify_transport(extensions): self.events.append("transport") return extensions.get("verified_factory_transport") @@ -84,8 +90,8 @@ def admit(observation, transport): or observation.method != "POST" or observation.route != "/v1/chat/completions" or observation.headers != EXPECTED_HEADERS - or observation.raw_body != FLOW_SDK_BODY_UTF8 - or observation.raw_body_sha256 != FLOW_SDK_BODY_SHA256 + or observation.raw_body != expected_raw + or observation.raw_body_sha256 != hashlib.sha256(expected_raw).hexdigest() or observation.normalized_body_sha256 != FLOW_NORMALIZED_SHA256 ): raise RequestAdmissionDenied() @@ -304,12 +310,112 @@ async def __call__(self, scope, receive, send): async def test_body_overflow_refuses_before_host_or_bearer_callbacks(self): expanded = json.loads(FLOW_SDK_BODY_UTF8) - expanded["messages"][0]["content"] = "x" * 32_768 + expanded["messages"][0]["content"] = "x" * MAX_FACTORY_RAW_BODY_BYTES response = await self.post(json.dumps(expanded).encode("utf-8")) self.assertEqual(response.status_code, 413) + # Invalid JSON would normally be 422; the ASGI gate must run first. + invalid = b"{" + b"x" * MAX_FACTORY_RAW_BODY_BYTES + self.assertEqual((await self.post(invalid)).status_code, 413) self.assertEqual(self.events, []) self.assertEqual(self.observations, []) + async def test_exact_body_limit_replays_all_bytes_into_original_adapter(self): + raw = FLOW_SDK_BODY_UTF8 + b" " * (MAX_FACTORY_RAW_BODY_BYTES - len(FLOW_SDK_BODY_UTF8)) + self.assertEqual(len(raw), MAX_FACTORY_RAW_BODY_BYTES) + response = await self.post(raw, app=self.app(expected_raw=raw)) + self.assertEqual(response.status_code, 200) + self.assertEqual(self.events, ["transport", "original", "bearer"]) + self.assertEqual(self.observations[0].raw_body, raw) + self.assertEqual(self.observations[0].normalized_body_sha256, FLOW_NORMALIZED_SHA256) + self.assertEqual(self.claims, 1) + self.assertEqual(self.peer.after_claim, 0) + + async def test_chunked_overflow_stops_at_first_excess_frame_before_inner_app(self): + entered = [] + received = [] + sent = [] + frames = iter( + [ + {"type": "http.request", "body": b"x" * MAX_FACTORY_RAW_BODY_BYTES, "more_body": True}, + {"type": "http.request", "body": b"y", "more_body": True}, + {"type": "http.request", "body": b"z", "more_body": False}, + ] + ) + + async def inner(scope, receive, send): + entered.append(True) + + async def receive(): + received.append(True) + return next(frames) + + async def send(message): + sent.append(message) + + gate = FactoryBoundedBodyMiddleware(inner) + await gate({"type": "http", "method": "POST", "path": "/v1/chat/completions"}, receive, send) + self.assertEqual(entered, []) + self.assertEqual(len(received), 2) + self.assertEqual(sent[0]["status"], 413) + self.assertEqual(sent[1]["type"], "http.response.body") + + async def test_chunked_299_byte_fixture_replays_exact_body_once(self): + frames = iter( + [ + {"type": "http.request", "body": FLOW_SDK_BODY_UTF8[:100], "more_body": True}, + {"type": "http.request", "body": b"", "more_body": True}, + {"type": "http.request", "body": FLOW_SDK_BODY_UTF8[100:], "more_body": False}, + ] + ) + received = [] + sent = [] + + async def inner(scope, receive, send): + received.append(await receive()) + await send({"type": "http.response.start", "status": 204, "headers": []}) + await send({"type": "http.response.body", "body": b""}) + + async def receive(): + return next(frames) + + async def send(message): + sent.append(message) + + gate = FactoryBoundedBodyMiddleware(inner) + await gate({"type": "http", "method": "POST", "path": "/v1/chat/completions"}, receive, send) + self.assertEqual(received, [{"type": "http.request", "body": FLOW_SDK_BODY_UTF8, "more_body": False}]) + self.assertEqual(sent[0]["status"], 204) + + async def test_declared_oversize_refuses_without_reading_any_body_frame(self): + entered = [] + received = [] + sent = [] + + async def inner(scope, receive, send): + entered.append(True) + + async def receive(): + received.append(True) + raise AssertionError("body frame should not be requested") + + async def send(message): + sent.append(message) + + gate = FactoryBoundedBodyMiddleware(inner) + await gate( + { + "type": "http", + "method": "POST", + "path": "/v1/chat/completions", + "headers": [(b"content-length", str(MAX_FACTORY_RAW_BODY_BYTES + 1).encode("ascii"))], + }, + receive, + send, + ) + self.assertEqual(entered, []) + self.assertEqual(received, []) + self.assertEqual(sent[0]["status"], 413) + async def test_duplicate_or_oversize_projected_header_refuses_before_callbacks(self): for headers in ( [("content-type", "application/json"), ("Content-Type", "application/json")], From 6a55f0b64b7ccb78f7343419feddcd2add9181fe Mon Sep 17 00:00:00 2001 From: flujo-app <300233937+flujo-app@users.noreply.github.com> Date: Sun, 4 Oct 2026 07:08:33 -0500 Subject: [PATCH 3/6] Cap mounted Factory v2 POST paths before routing --- docs/factory_receiver_v2_candidate.md | 6 +++--- src/drift/factory_receiver.py | 6 +----- tests/test_factory_receiver_v2.py | 11 ++++++++++- 3 files changed, 14 insertions(+), 9 deletions(-) diff --git a/docs/factory_receiver_v2_candidate.md b/docs/factory_receiver_v2_candidate.md index a9be3ddf4..5df724807 100644 --- a/docs/factory_receiver_v2_candidate.md +++ b/docs/factory_receiver_v2_candidate.md @@ -6,7 +6,7 @@ existing ordinary distributed text path. It does not change the v1 configured together. A v2 adapter must return `FactoryAdmissionV2`; a bare v1 admission is denied. -An outer ASGI gate caps the two v2 inference routes before FastAPI parses JSON. +An outer ASGI gate caps every POST to the v2 app before FastAPI parses JSON. It retains at most 32,768 body bytes, stops at the first excess frame with 413, rejects a declared oversized `Content-Length` before reading any frame, and replays the accepted bytes unchanged. The receiver then supplies a @@ -53,7 +53,7 @@ receipt and a corrected normalization commitment are required for a join. ASGI delivers whole frames, so an oversized single frame can exist transiently before the gate rejects it. The deployment ingress still needs a global edge -limit on request/frame size and duration, including paths outside these two -routes. This candidate has no Original issuer, authenticated transport, +limit on request/frame size and duration, including non-POST paths. This +candidate has no Original issuer, authenticated transport, deployed receiver receipt, or physical send. The offline test host supplies synthetic values and holds the dispatch claim. diff --git a/src/drift/factory_receiver.py b/src/drift/factory_receiver.py index 2e1cc0735..7b9476b1a 100644 --- a/src/drift/factory_receiver.py +++ b/src/drift/factory_receiver.py @@ -73,11 +73,7 @@ def __init__(self, app): self.app = app async def __call__(self, scope, receive, send): - if ( - scope.get("type") != "http" - or scope.get("method") != "POST" - or scope.get("path") not in ("/v1/chat/completions", "/v1/completions") - ): + if scope.get("type") != "http" or scope.get("method") != "POST": return await self.app(scope, receive, send) # A truthful Content-Length lets us refuse without asking ASGI for even diff --git a/tests/test_factory_receiver_v2.py b/tests/test_factory_receiver_v2.py index da45ec5c5..8e96394c3 100644 --- a/tests/test_factory_receiver_v2.py +++ b/tests/test_factory_receiver_v2.py @@ -353,7 +353,16 @@ async def send(message): sent.append(message) gate = FactoryBoundedBodyMiddleware(inner) - await gate({"type": "http", "method": "POST", "path": "/v1/chat/completions"}, receive, send) + await gate( + { + "type": "http", + "method": "POST", + "path": "/mounted/v1/chat/completions", + "root_path": "/mounted", + }, + receive, + send, + ) self.assertEqual(entered, []) self.assertEqual(len(received), 2) self.assertEqual(sent[0]["status"], 413) From 1df490d686977f0f6fa80ccdd5d1f51f0b5846aa Mon Sep 17 00:00:00 2001 From: flujo-app <300233937+flujo-app@users.noreply.github.com> Date: Sun, 4 Oct 2026 07:12:41 -0500 Subject: [PATCH 4/6] Run Factory v2 receiver checks in hosted matrix --- .github/workflows/run-tests.yaml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.github/workflows/run-tests.yaml b/.github/workflows/run-tests.yaml index f48631911..a06ec48ff 100644 --- a/.github/workflows/run-tests.yaml +++ b/.github/workflows/run-tests.yaml @@ -52,6 +52,7 @@ jobs: source .venv/bin/activate python -m pytest -p pytest_asyncio.plugin -q --disable-warnings \ tests/test_factory_admission.py tests/test_factory_gateway.py tests/test_factory_sdk_asgi_hold.py \ + tests/test_factory_receiver_v2.py \ tests/test_text_peer_request_guards.py tests/test_text_request_deadline.py \ tests/test_openai_api.py tests/test_official_openai_client.py tests/test_text_mesh.py \ -k 'not RealTransportTests' @@ -94,6 +95,7 @@ jobs: tests/test_factory_admission.py \ tests/test_factory_gateway.py \ tests/test_factory_sdk_asgi_hold.py \ + tests/test_factory_receiver_v2.py \ tests/test_device_portability.py \ tests/test_gate16_catalog_channel.py \ tests/test_gate16_catalog_drill.py \ From 8e933f58107d599e9d54d17431140393d0cc8ed8 Mon Sep 17 00:00:00 2001 From: flujo-app <300233937+flujo-app@users.noreply.github.com> Date: Sun, 4 Oct 2026 07:14:03 -0500 Subject: [PATCH 5/6] Update Factory receiver join status --- docs/factory_receiver_v2_candidate.md | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/docs/factory_receiver_v2_candidate.md b/docs/factory_receiver_v2_candidate.md index 5df724807..965c10d9b 100644 --- a/docs/factory_receiver_v2_candidate.md +++ b/docs/factory_receiver_v2_candidate.md @@ -42,14 +42,12 @@ The real Flow fixture has a 299-byte SDK body with SHA-256 `b86e125b6a031a226acc03147b5d865402ad149a621626fcdbf7629f5541fd03`. PR35's Python-normalized body has 307 bytes and SHA-256 `ab288a143d34de92156f01d0d4c4f5d05bbdc51eda4a335e16c8816205cd08b7`. -The separate FACTORY pure held v3 contract at `697a27d` currently constructs a -305-byte JavaScript-normalized body (SHA-256 -`f1a06120ca623ed9dafcdd31c1ec6e54468071005306c481a6d071a388237b44`) -because it serializes `temperature` as `1`, while Python emits `1.0`. Its -`HELD_BEFORE_POST` expected observation includes authenticated sender and -credential fields and the ten SDK-final headers; it cannot be filled by this -post-ingress ASGI observation. A separate authenticated post-POST receiver -receipt and a corrected normalization commitment are required for a join. +The separate FACTORY held v3 candidate now commits that same 307-byte Python +normalization. Its `HELD_BEFORE_POST` expected observation includes the +authenticated sender and credential fields and the ten SDK-final headers; it +cannot be filled by this post-ingress ASGI observation. FACTORY has a separate +synthetic post-POST receipt shape, but a real authenticated receipt and host +trust root are still required for the join. ASGI delivers whole frames, so an oversized single frame can exist transiently before the gate rejects it. The deployment ingress still needs a global edge From e8f85285a974c64d5276054817dadbb0dd31c6f9 Mon Sep 17 00:00:00 2001 From: flujo-app <300233937+flujo-app@users.noreply.github.com> Date: Sun, 4 Oct 2026 07:36:29 -0500 Subject: [PATCH 6/6] Fix Factory receiver v2 raw route checks under mounts --- src/drift/factory_receiver.py | 9 ++++++--- tests/test_factory_receiver_v2.py | 28 ++++++++++++++++++++++++++++ 2 files changed, 34 insertions(+), 3 deletions(-) diff --git a/src/drift/factory_receiver.py b/src/drift/factory_receiver.py index 7b9476b1a..dcf4776be 100644 --- a/src/drift/factory_receiver.py +++ b/src/drift/factory_receiver.py @@ -236,11 +236,14 @@ def observe_factory_ingress_v2(scope: dict, raw_body: bytes, normalized_body: di """Reject unbounded/ambiguous values before any trusted host callback.""" if type(raw_body) is not bytes or len(raw_body) > MAX_FACTORY_RAW_BODY_BYTES: raise FactoryIngressTooLarge() - raw_path = scope.get("raw_path", route.encode("ascii")) + root_path = scope.get("root_path", "") + if type(root_path) is not str or re.fullmatch(r"(?:/[A-Za-z0-9._~-]+)*", root_path) is None: + raise RequestAdmissionDenied() + full_route = root_path + route if ( scope.get("method") != "POST" - or scope.get("path") != route - or raw_path != route.encode("ascii") + or scope.get("path") != full_route + or scope.get("raw_path") != full_route.encode("ascii") or scope.get("query_string", b"") != b"" ): raise RequestAdmissionDenied() diff --git a/tests/test_factory_receiver_v2.py b/tests/test_factory_receiver_v2.py index 8e96394c3..1f428bc0a 100644 --- a/tests/test_factory_receiver_v2.py +++ b/tests/test_factory_receiver_v2.py @@ -5,6 +5,7 @@ import unittest import httpx +from fastapi import FastAPI from test_factory_sdk_asgi_hold import ( FLOW_NORMALIZED_SHA256, FLOW_SDK_BODY_SHA256, @@ -178,6 +179,33 @@ async def test_exact_real_flow_body_and_receiver_headers_reach_host_before_beare self.assertEqual(self.peer.after_claim, 0) self.assertIn('"code": "request_not_dispatched"', response.text) + async def test_parent_mount_preserves_exact_route_admission(self): + parent = FastAPI() + parent.mount("/factory", self.app()) + response = await self.post(app=parent, path="/factory/v1/chat/completions") + self.assertEqual(response.status_code, 200) + self.assertEqual(self.events, ["transport", "original", "bearer"]) + self.assertEqual(self.observations[0].route, "/v1/chat/completions") + self.assertEqual(self.observations[0].raw_body, FLOW_SDK_BODY_UTF8) + self.assertEqual(self.claims, 1) + self.assertEqual(self.peer.after_claim, 0) + + async def test_missing_raw_path_refuses_before_host_callbacks(self): + class WithoutRawPath: + def __init__(self, app): + self.app = app + + async def __call__(self, scope, receive, send): + scope = dict(scope) + scope.pop("raw_path", None) + await self.app(scope, receive, send) + + response = await self.post(app=WithoutRawPath(self.app())) + self.assertEqual(response.status_code, 403) + self.assertEqual(self.events, []) + self.assertEqual(self.observations, []) + self.assertEqual(self.claims, 0) + async def test_headers_and_bearer_cannot_supply_missing_host_transport(self): response = await self.post( app=self.app(identity=False),