diff --git a/openadapt_flow/connector/client.py b/openadapt_flow/connector/client.py index 0b2a0d4f..93e5fb1f 100644 --- a/openadapt_flow/connector/client.py +++ b/openadapt_flow/connector/client.py @@ -28,6 +28,8 @@ from openadapt_flow.hosted import HostedError +MANAGED_DELIVERY_AUTHORITY_CAPABILITY = "managed_delivery_authority_v1" + class ConnectorClientError(HostedError): """An outbound control-plane call failed.""" @@ -88,7 +90,12 @@ def poll(self, wait_s: int) -> Optional[dict[str, Any]]: """Long-poll for the next leased job. Returns the poll envelope ``{"job": {...}}`` or None on a 204 (no work in the wait window).""" resp = self._client.post( - "/api/connector/poll", json={"wait": wait_s}, headers=self._bearer() + "/api/connector/poll", + json={ + "wait": wait_s, + "capabilities": [MANAGED_DELIVERY_AUTHORITY_CAPABILITY], + }, + headers=self._bearer(), ) if resp.status_code == 204: return None diff --git a/openadapt_flow/connector/executor.py b/openadapt_flow/connector/executor.py index 37fc40b3..1791372d 100644 --- a/openadapt_flow/connector/executor.py +++ b/openadapt_flow/connector/executor.py @@ -41,13 +41,21 @@ import tempfile from dataclasses import dataclass, field from pathlib import Path -from typing import Any, Callable, Optional +from typing import Any, Callable, Mapping, Optional +from urllib.parse import urlsplit from openadapt_flow.connector.config import ConnectorSettings from openadapt_flow.connector.protocol import ByocGovernanceError, ByocJob from openadapt_flow.connector.storage import CustomerStorage, extract_bundle_archive from openadapt_flow.failure_signals import automation_failure_signal from openadapt_flow.ir import RunReport +from openadapt_flow.runner.dispatch_envelope import ( + write_managed_dispatch_envelope_value, +) +from openadapt_flow.runtime.durable.authority import ( + REMOTE_AUTHORITY_TOKEN_ENV, + REMOTE_AUTHORITY_URL_ENV, +) #: A run-gate refusal (fail-closed admission denied) exits 2 before the replay #: creates report.json. @@ -64,7 +72,9 @@ class RunOutcome: #: A runner maps argv -> RunOutcome. The default shells the governed ``run`` CLI #: in a child process; tests inject a fake to avoid launching a GUI. -Runner = Callable[[list[str], Path], RunOutcome] +Runner = Callable[[list[str], Path, Mapping[str, str]], RunOutcome] + +_MANAGED_DELIVERY_AUTHORITY_PATH = "/api/internal/managed-delivery-permit" @dataclass @@ -104,6 +114,7 @@ def build_run_argv( bundle_dir: Path, run_dir: Path, params_file: Optional[Path], + managed_dispatch_file: Optional[Path] = None, ) -> list[str]: """The exact governed CLI invocation for a verified BYOC dispatch. @@ -132,14 +143,23 @@ def build_run_argv( argv += ["--params-file", str(params_file)] if job.target_url: argv += ["--url", job.target_url] + if managed_dispatch_file is not None: + argv += ["--managed-dispatch-file", str(managed_dispatch_file)] if settings.allow_unencrypted: # Local escape hatch, mirroring the governed run CLI; default OFF. argv.append("--allow-unencrypted") return argv -def _subprocess_runner(argv: list[str], run_dir: Path) -> RunOutcome: - proc = subprocess.run(argv, capture_output=True, text=True) # nosec - fixed argv +def _subprocess_runner( + argv: list[str], run_dir: Path, child_env: Mapping[str, str] +) -> RunOutcome: + proc = subprocess.run( # nosec - fixed argv and exact process-local env + argv, + capture_output=True, + text=True, + env=dict(child_env), + ) report_path = run_dir / "report.json" report: dict[str, Any] = {} if report_path.is_file(): @@ -267,6 +287,44 @@ def _write_policy_audit(job: ByocJob, run_dir: Path) -> None: pass +def _assert_managed_authority_is_pinned( + job: ByocJob, settings: ConnectorSettings +) -> None: + """Keep the run capability on the enrolled control-plane origin.""" + + if job.managed_dispatch is None: + return + try: + configured = urlsplit(settings.control_plane_url) + delivered = urlsplit(job.managed_delivery_authority_url or "") + configured_port = configured.port or ( + 443 if configured.scheme == "https" else None + ) + delivered_port = delivered.port or ( + 443 if delivered.scheme == "https" else None + ) + except ValueError as exc: + raise ByocGovernanceError( + "byoc managed delivery authority URL is invalid" + ) from exc + if ( + configured.scheme != "https" + or not configured.hostname + or configured.username is not None + or configured.password is not None + or bool(configured.query) + or bool(configured.fragment) + or delivered.scheme != "https" + or delivered.hostname != configured.hostname + or delivered_port != configured_port + or delivered.path != _MANAGED_DELIVERY_AUTHORITY_PATH + ): + raise ByocGovernanceError( + "byoc managed delivery authority is not pinned to the enrolled " + "control plane (fail closed)" + ) + + def execute_job( job: ByocJob, settings: ConnectorSettings, @@ -284,6 +342,7 @@ def execute_job( # 1. Governance gates (fail closed BEFORE any bundle is fetched or run). try: job.ensure_governed(require_run_token=require_run_token) + _assert_managed_authority_is_pinned(job, settings) except ByocGovernanceError as exc: return ExecutionResult("failed", {}, None, job.report_ref(), str(exc)) if not _grounding_env_available(job): @@ -332,10 +391,32 @@ def execute_job( try: params_file = _write_params_file(job.params, run_dir) + managed_dispatch_file = None + child_env = os.environ.copy() + # Never let a long-running Connector inherit stale authority from + # its service environment. Add the exact run capability only for + # the child that also receives the matching dispatch envelope. + child_env.pop(REMOTE_AUTHORITY_URL_ENV, None) + child_env.pop(REMOTE_AUTHORITY_TOKEN_ENV, None) + if job.managed_dispatch is not None: + managed_dispatch_file = write_managed_dispatch_envelope_value( + run_dir / "managed-dispatch.json", job.managed_dispatch + ) + child_env[REMOTE_AUTHORITY_URL_ENV] = str( + job.managed_delivery_authority_url + ) + child_env[REMOTE_AUTHORITY_TOKEN_ENV] = str(job.run_token) # 3. The governed, fail-closed child invocation. - argv = build_run_argv(job, settings, Path(bundle_dir), run_dir, params_file) - outcome = runner(argv, run_dir) + argv = build_run_argv( + job, + settings, + Path(bundle_dir), + run_dir, + params_file, + managed_dispatch_file, + ) + outcome = runner(argv, run_dir, child_env) except Exception as exc: return ExecutionResult( "failed", diff --git a/openadapt_flow/connector/protocol.py b/openadapt_flow/connector/protocol.py index be6734fe..5126d9ec 100644 --- a/openadapt_flow/connector/protocol.py +++ b/openadapt_flow/connector/protocol.py @@ -21,10 +21,12 @@ from __future__ import annotations from typing import Any, Optional +from urllib.parse import urlsplit from pydantic import BaseModel, ConfigDict, Field, ValidationError from openadapt_flow.ir import ExecutionTargetKind +from openadapt_flow.runner.dispatch_envelope import ManagedDispatchEnvelope class ByocJobParseError(ValueError): @@ -100,9 +102,15 @@ class ByocJob(BaseModel): bundle_download_url: Optional[str] = None # --- Governed callback binding (fail-closed) ------------------------------- - #: Run-scoped HMAC capability presented as ``x-run-token`` on the PHI-free - #: callback. Not a secret value — proves this run, forbids forging another. + #: Run-scoped bearer capability presented as ``x-run-token`` on the PHI-free + #: callback. It proves this run and must remain private. run_token: Optional[str] = None + #: Exact Cloud-issued, one-run authority. It stays PHI-free and is written + #: to a private local file before the governed child starts. + managed_dispatch: Optional[ManagedDispatchEnvelope] = None + #: HTTPS endpoint that issues monotonic per-action delivery permits for the + #: exact managed dispatch. The run-scoped ``run_token`` authenticates it. + managed_delivery_authority_url: Optional[str] = None bundle_version_id: Optional[str] = None runtime_validation_id: Optional[str] = None #: SHA-256 of the exact approved sanitized derivative ZIP bytes staged at @@ -162,6 +170,46 @@ def ensure_governed(self, *, require_run_token: bool = True) -> None: "byoc dispatch is missing a run-scoped callback token; refusing " "to run a job whose outcome we could not report (fail closed)" ) + if (self.managed_dispatch is None) != ( + self.managed_delivery_authority_url is None + ): + raise ByocGovernanceError( + "byoc managed dispatch and delivery authority must be supplied " + "together (fail closed)" + ) + if self.managed_dispatch is not None: + try: + authorization = self.managed_dispatch.exact_authorization() + except ValueError as exc: + raise ByocGovernanceError( + "byoc managed dispatch has an invalid authorization binding" + ) from exc + if self.managed_dispatch.run_id != self.run_id: + raise ByocGovernanceError( + "byoc managed dispatch belongs to a different run (fail closed)" + ) + if not authorization.approval_source.startswith("hosted:"): + raise ByocGovernanceError( + "byoc managed dispatch lacks hosted authorization provenance" + ) + try: + parsed = urlsplit(self.managed_delivery_authority_url or "") + except ValueError as exc: + raise ByocGovernanceError( + "byoc managed delivery authority URL is invalid" + ) from exc + if ( + parsed.scheme != "https" + or not parsed.hostname + or parsed.username is not None + or parsed.password is not None + or bool(parsed.query) + or bool(parsed.fragment) + ): + raise ByocGovernanceError( + "byoc managed delivery authority must be a credential-free " + "HTTPS endpoint without query or fragment" + ) if not self.bundle_version_id or not self.runtime_validation_id: raise ByocGovernanceError( "byoc dispatch is missing its immutable bundle-version or " diff --git a/openadapt_flow/runner/dispatch_envelope.py b/openadapt_flow/runner/dispatch_envelope.py index 5fdff6b1..693e2652 100644 --- a/openadapt_flow/runner/dispatch_envelope.py +++ b/openadapt_flow/runner/dispatch_envelope.py @@ -93,18 +93,18 @@ def exact_authorization(self) -> GovernedRunAuthorization: return self.authorization -def write_managed_dispatch_envelope(path: Path, verified: "VerifiedDispatch") -> Path: - """Write one already verified dispatch without following a path.""" - - run_id = verified.payload.run_id - authorization = verified.payload.authorization - envelope = ManagedDispatchEnvelope( - run_id=run_id, - bundle_content_digest=authorization.bundle_content_digest, - runtime_inputs_digest=authorization.runtime_inputs_digest, - authorization=authorization, - dispatch_binding_sha256=verified.payload.dispatch_binding_sha256, - ) +def write_managed_dispatch_envelope_value( + path: Path, envelope: ManagedDispatchEnvelope +) -> Path: + """Write one strictly parsed dispatch envelope without following a path. + + This is the shared process-boundary writer for both push runners and the + customer-controlled outbound-pull Connector. The caller must already hold + the strict :class:`ManagedDispatchEnvelope` model; this function rechecks + its internal authorization binding before any bytes reach disk. + """ + + envelope.exact_authorization() path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL @@ -136,6 +136,21 @@ def write_managed_dispatch_envelope(path: Path, verified: "VerifiedDispatch") -> return path +def write_managed_dispatch_envelope(path: Path, verified: "VerifiedDispatch") -> Path: + """Write one already verified runner dispatch without following a path.""" + + run_id = verified.payload.run_id + authorization = verified.payload.authorization + envelope = ManagedDispatchEnvelope( + run_id=run_id, + bundle_content_digest=authorization.bundle_content_digest, + runtime_inputs_digest=authorization.runtime_inputs_digest, + authorization=authorization, + dispatch_binding_sha256=verified.payload.dispatch_binding_sha256, + ) + return write_managed_dispatch_envelope_value(path, envelope) + + def _read_managed_dispatch_envelope(path: Path) -> ManagedDispatchEnvelope: """Read only a regular, private file owned by this effective user.""" diff --git a/openadapt_flow/runtime/durable/authority.py b/openadapt_flow/runtime/durable/authority.py index 130f4d59..ca42a0ce 100644 --- a/openadapt_flow/runtime/durable/authority.py +++ b/openadapt_flow/runtime/durable/authority.py @@ -31,7 +31,7 @@ from typing import Any, Iterator, Literal, Optional from urllib.error import HTTPError, URLError from urllib.parse import urlparse -from urllib.request import Request, urlopen +from urllib.request import HTTPRedirectHandler, Request, build_opener from pydantic import BaseModel, ConfigDict @@ -70,6 +70,27 @@ } +class _RefuseRemoteAuthorityRedirects(HTTPRedirectHandler): + """Keep the runner credential on the configured authority origin.""" + + def redirect_request( + self, + req: Request, + fp: Any, + code: int, + msg: str, + headers: Any, + newurl: str, + ) -> None: + raise HTTPError( + req.full_url, + code, + "remote delivery authority redirects are refused", + headers, + fp, + ) + + def _is_windows() -> bool: return os.name == "nt" @@ -1645,7 +1666,8 @@ def _require_remote_delivery_permit( response_bytes = self._remote_transport(url, headers, body) else: request = Request(url, data=body, headers=headers, method="POST") - with urlopen(request, timeout=10) as response: # nosec B310 - HTTPS above + opener = build_opener(_RefuseRemoteAuthorityRedirects()) + with opener.open(request, timeout=10) as response: # nosec B310 - HTTPS above if not 200 <= response.status < 300: raise DurableAuthorityBusy("remote delivery authority refused") response_bytes = response.read( diff --git a/tests/test_connector.py b/tests/test_connector.py index 6d23edf0..79a39a3e 100644 --- a/tests/test_connector.py +++ b/tests/test_connector.py @@ -20,6 +20,8 @@ import hashlib import io import json +import os +import stat import zipfile from pathlib import Path @@ -44,6 +46,13 @@ from openadapt_flow.connector.protocol import ByocGovernanceError from openadapt_flow.connector.storage import LocalCustomerStorage from openadapt_flow.failure_signals import automation_failure_signal +from openadapt_flow.runner.dispatch_envelope import ManagedDispatchEnvelope +from openadapt_flow.runner.protocol import dispatch_binding_sha256 +from openadapt_flow.runtime.authorization import GovernedRunAuthorization +from openadapt_flow.runtime.durable.authority import ( + REMOTE_AUTHORITY_TOKEN_ENV, + REMOTE_AUTHORITY_URL_ENV, +) def _bundle_archive_bytes() -> bytes: @@ -90,8 +99,27 @@ def _payload(**overrides): return payload +def _managed_dispatch(*, run_id: str | None = None) -> dict: + exact_run_id = run_id or _payload()["run_id"] + authorization = GovernedRunAuthorization( + bundle_content_digest="b" * 64, + runtime_inputs_digest="c" * 64, + admitted_policy_name="standard", + execution_profile="standard", + approval_source="hosted:execute-api", + ) + envelope = ManagedDispatchEnvelope( + run_id=exact_run_id, + bundle_content_digest=authorization.bundle_content_digest, + runtime_inputs_digest=authorization.runtime_inputs_digest, + authorization=authorization, + dispatch_binding_sha256=dispatch_binding_sha256(exact_run_id, authorization), + ) + return envelope.model_dump(mode="json") + + def _fake_success_runner(report): - def runner(argv, run_dir: Path) -> RunOutcome: + def runner(argv, run_dir: Path, _env) -> RunOutcome: (run_dir / "report.json").write_text(json.dumps(report)) return RunOutcome(returncode=0, report=report) @@ -134,6 +162,68 @@ def test_execute_job_success_writes_report_to_customer_storage_only(): assert storage.written["org_demo/run_5/report.json"]["success"] is True +def test_execute_job_consumes_exact_managed_dispatch_through_private_child_boundary( + monkeypatch, +): + authority_url = "https://app.openadapt.ai/api/internal/managed-delivery-permit" + token = "a" * 64 + managed_dispatch = _managed_dispatch() + monkeypatch.setenv(REMOTE_AUTHORITY_URL_ENV, "https://stale.invalid/permit") + monkeypatch.setenv(REMOTE_AUTHORITY_TOKEN_ENV, "stale-parent-token") + job = parse_job( + _payload( + managed_dispatch=managed_dispatch, + managed_delivery_authority_url=authority_url, + ), + lease_job_id="bjob_1", + ) + storage = InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES) + observed: dict[str, object] = {} + + def runner(argv, run_dir: Path, child_env) -> RunOutcome: + envelope_path = Path(argv[argv.index("--managed-dispatch-file") + 1]) + observed["argv"] = argv + observed["mode"] = stat.S_IMODE(envelope_path.stat().st_mode) + observed["envelope"] = json.loads(envelope_path.read_text()) + observed["authority_url"] = child_env[REMOTE_AUTHORITY_URL_ENV] + observed["authority_token"] = child_env[REMOTE_AUTHORITY_TOKEN_ENV] + observed["parent_token"] = os.environ.get(REMOTE_AUTHORITY_TOKEN_ENV) + (run_dir / "report.json").write_text(json.dumps(SUCCESS_REPORT)) + return RunOutcome(returncode=0, report=SUCCESS_REPORT) + + result = execute_job(job, ConnectorSettings(), storage, runner=runner) + + assert result.status == "success" + assert observed["mode"] == 0o600 + assert observed["envelope"] == managed_dispatch + assert observed["authority_url"] == authority_url + assert observed["authority_token"] == token + assert observed["parent_token"] == "stale-parent-token" + assert token not in json.dumps(observed["argv"]) + assert token not in json.dumps(observed["envelope"]) + + +def test_connector_drops_inherited_authority_from_a_job_without_an_envelope( + monkeypatch, +): + monkeypatch.setenv(REMOTE_AUTHORITY_URL_ENV, "https://stale.invalid/permit") + monkeypatch.setenv(REMOTE_AUTHORITY_TOKEN_ENV, "stale-parent-token") + job = parse_job(_payload(), lease_job_id="bjob_1") + storage = InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES) + + def runner(argv, run_dir: Path, child_env) -> RunOutcome: + assert REMOTE_AUTHORITY_URL_ENV not in child_env + assert REMOTE_AUTHORITY_TOKEN_ENV not in child_env + assert "--managed-dispatch-file" not in argv + (run_dir / "report.json").write_text(json.dumps(SUCCESS_REPORT)) + return RunOutcome(returncode=0, report=SUCCESS_REPORT) + + assert ( + execute_job(job, ConnectorSettings(), storage, runner=runner).status + == "success" + ) + + def test_execute_job_refuses_child_report_for_different_substrate(): job = parse_job(_payload(), lease_job_id="bjob_1") settings = ConnectorSettings(profile=None) @@ -281,7 +371,7 @@ def test_halt_maps_to_halt_status_and_present_flag(): settings = ConnectorSettings() storage = InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES) - def runner(argv, run_dir: Path) -> RunOutcome: + def runner(argv, run_dir: Path, _env) -> RunOutcome: (run_dir / "report.json").write_text(json.dumps(halt_report)) return RunOutcome(returncode=2, report=halt_report) @@ -365,6 +455,64 @@ def test_dispatch_without_run_token_is_refused(): job.ensure_governed() +def test_managed_dispatch_for_different_run_is_refused(): + job = parse_job( + _payload( + managed_dispatch=_managed_dispatch( + run_id="00000000-0000-4000-8000-000000000006" + ), + managed_delivery_authority_url=( + "https://app.openadapt.ai/api/internal/managed-delivery-permit" + ), + ), + lease_job_id="bjob_1", + ) + with pytest.raises(ByocGovernanceError, match="different run"): + job.ensure_governed() + + +def test_execute_refuses_managed_authority_on_a_different_https_origin(): + job = parse_job( + _payload( + managed_dispatch=_managed_dispatch(), + managed_delivery_authority_url=( + "https://attacker.invalid/api/internal/managed-delivery-permit" + ), + ), + lease_job_id="bjob_1", + ) + result = execute_job( + job, + ConnectorSettings(control_plane_url="https://app.openadapt.ai"), + InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES), + runner=_fake_success_runner(SUCCESS_REPORT), + ) + assert result.status == "failed" + assert "pinned to the enrolled control plane" in (result.error or "") + + +@pytest.mark.parametrize( + "overrides, match", + [ + ( + {"managed_dispatch": _managed_dispatch()}, + "must be supplied together", + ), + ( + { + "managed_dispatch": _managed_dispatch(), + "managed_delivery_authority_url": "http://localhost/permit", + }, + "credential-free HTTPS endpoint", + ), + ], +) +def test_incomplete_or_unsafe_managed_authority_is_refused(overrides, match): + job = parse_job(_payload(**overrides), lease_job_id="bjob_1") + with pytest.raises(ByocGovernanceError, match=match): + job.ensure_governed() + + @pytest.mark.parametrize("field", ["bundle_version_id", "runtime_validation_id"]) def test_dispatch_without_immutable_validation_binding_is_refused(field): job = parse_job(_payload(**{field: None}), lease_job_id="bjob_1") @@ -398,7 +546,7 @@ def test_execute_refuses_when_required_grounding_key_is_absent(monkeypatch): # A runner that would "succeed" — the governance gate must refuse BEFORE it. called = {"ran": False} - def runner(argv, run_dir: Path) -> RunOutcome: + def runner(argv, run_dir: Path, _env) -> RunOutcome: called["ran"] = True return RunOutcome(returncode=0, report=SUCCESS_REPORT) @@ -413,7 +561,7 @@ def test_execute_refuses_archive_digest_mismatch_before_runner(): storage = InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES) called = {"ran": False} - def runner(argv, run_dir: Path) -> RunOutcome: + def runner(argv, run_dir: Path, _env) -> RunOutcome: called["ran"] = True return RunOutcome(returncode=0, report=SUCCESS_REPORT) @@ -429,7 +577,7 @@ def test_child_failure_after_archive_verification_carries_verified_digest(): job = parse_job(_payload(), lease_job_id="bjob_1") storage = InMemoryCustomerStorage(bundle_bytes=_BUNDLE_BYTES) - def runner(argv, run_dir: Path) -> RunOutcome: + def runner(argv, run_dir: Path, _env) -> RunOutcome: raise RuntimeError("child unavailable") result = execute_job(job, ConnectorSettings(), storage, runner=runner) @@ -507,6 +655,7 @@ def __init__(self): self.tokens = {} # token -> org_id self.jobs = [] # queued jobs (each {id, org_id, payload, status, leased_by}) self.callbacks = [] + self.poll_bodies = [] self.callback_attempts = 0 self.callback_status = 200 self.acks = [] @@ -542,6 +691,7 @@ def handler(self, request: httpx.Request) -> httpx.Response: if path == "/api/connector/poll": if org is None: return httpx.Response(401, json={"error": "invalid connector token"}) + self.poll_bodies.append(body) for j in self.jobs: # ISOLATION: a connector only ever leases ITS OWN org jobs. if j["status"] == "queued" and j["org_id"] == org: @@ -611,6 +761,9 @@ def test_full_loop_dispatch_execute_callback_ack(): storage_factory=lambda job: storage, ) assert result["status"] == "success" + assert cp.poll_bodies == [ + {"wait": 0, "capabilities": ["managed_delivery_authority_v1"]} + ] # A PHI-free callback carrying the run-scoped token was posted. assert len(cp.callbacks) == 1 diff --git a/tests/test_durable_authority_v13.py b/tests/test_durable_authority_v13.py index f9f26953..bbc65f6e 100644 --- a/tests/test_durable_authority_v13.py +++ b/tests/test_durable_authority_v13.py @@ -17,6 +17,7 @@ import pytest +import openadapt_flow.runtime.durable.authority as durable_authority_module from openadapt_flow.ir import ActionKind, RunReport, Step, StepResult, Workflow from openadapt_flow.runtime.authorization import GovernedRunAuthorization from openadapt_flow.runtime.durable.approval import ( @@ -296,6 +297,60 @@ def transport(_url: str, _headers: dict[str, str], _body: bytes) -> bytes: assert authority.validate(manifest).delivery_sequence == 0 +def test_remote_authority_refuses_redirect_before_forwarding_bearer_token( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + manifest, authority, owner = _remote_ready_authority( + tmp_path, monkeypatch, lambda *_args: b"{}" + ) + authority._remote_transport = None + monkeypatch.setenv( + REMOTE_AUTHORITY_URL_ENV, + "https://control.example/api/internal/managed-delivery-permit", + ) + attempted_urls: list[str] = [] + + class RedirectingOpener: + def __init__(self, handler: object) -> None: + self.handler = handler + + def open(self, request: object, *, timeout: int) -> object: + assert timeout == 10 + attempted_urls.append(request.full_url) # type: ignore[attr-defined] + assert request.get_header("Authorization") == "Bearer secret-token" # type: ignore[attr-defined] + self.handler.redirect_request( # type: ignore[attr-defined] + request, + None, + 307, + "Temporary Redirect", + {}, + "https://attacker.invalid/permit", + ) + raise AssertionError("redirect refusal must stop the request") + + def build_redirecting_opener(handler: object) -> RedirectingOpener: + assert isinstance( + handler, durable_authority_module._RefuseRemoteAuthorityRedirects + ) + return RedirectingOpener(handler) + + monkeypatch.setattr( + durable_authority_module, + "build_opener", + build_redirecting_opener, + ) + with pytest.raises(DurableAuthorityBusy, match="unavailable or refused"): + authority.before_delivery( + manifest, + attempt_id="aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + owner_nonce_sha256=owner, + ) + assert attempted_urls == [ + "https://control.example/api/internal/managed-delivery-permit" + ] + assert authority.validate(manifest).delivery_sequence == 0 + + def test_initial_customer_local_delivery_needs_no_cloud_credentials( tmp_path: Path, ) -> None: