diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7f875dc..945ba41 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -74,6 +74,7 @@ jobs: tests.test_temporal_runtime tests.test_cancellation_workflow_policy tests.test_sdk_metrics_contract + tests.test_command_response_loss tests.test_ci_trust tests.test_pip_effective_trust tests.test_cli_effective_trust diff --git a/.github/workflows/sonar.yml b/.github/workflows/sonar.yml index d6a06f2..9423d38 100644 --- a/.github/workflows/sonar.yml +++ b/.github/workflows/sonar.yml @@ -53,7 +53,7 @@ jobs: - name: Produce source coverage run: | python -m pip install --only-binary :all: coverage==7.10.7 jsonschema==4.23.0 PyYAML==6.0.2 - coverage run --source=scripts,conformance/cancellation/temporal,.github/actions/bootstrap-patched-pip -m unittest tests.test_qualification_policy tests.test_protected_policy_context tests.test_protected_policy_check tests.test_protected_policy_inputs tests.test_protected_policy_issue tests.test_native_policy_check tests.test_native_policy_workflow tests.test_regression_corpus tests.test_regression_evidence_schema tests.test_server_codec_counterfactual_schema tests.test_visual_evidence tests.test_temporal_runtime tests.test_cancellation_workflow_policy tests.test_sdk_metrics_contract tests.test_ci_trust tests.test_pip_effective_trust tests.test_cli_effective_trust tests.test_composer_trust_contract tests.test_composer_container_trust tests.test_composer_container_effective_trust tests.test_php_linux_bootstrap tests.test_php_linux_archive tests.test_php_linux_normalize tests.test_php_linux_recipe tests.test_php_linux_tools tests.test_php_linux_alternatives tests.test_setup_php_linux tests.test_php_linux_qualification tests.test_php_linux_qualification_workflow tests.test_php_linux_configuration + coverage run --source=scripts,conformance/cancellation/temporal,conformance/command-keys,.github/actions/bootstrap-patched-pip -m unittest tests.test_qualification_policy tests.test_protected_policy_context tests.test_protected_policy_check tests.test_protected_policy_inputs tests.test_protected_policy_issue tests.test_native_policy_check tests.test_native_policy_workflow tests.test_regression_corpus tests.test_regression_evidence_schema tests.test_server_codec_counterfactual_schema tests.test_visual_evidence tests.test_temporal_runtime tests.test_cancellation_workflow_policy tests.test_sdk_metrics_contract tests.test_command_response_loss tests.test_ci_trust tests.test_pip_effective_trust tests.test_cli_effective_trust tests.test_composer_trust_contract tests.test_composer_container_trust tests.test_composer_container_effective_trust tests.test_php_linux_bootstrap tests.test_php_linux_archive tests.test_php_linux_normalize tests.test_php_linux_recipe tests.test_php_linux_tools tests.test_php_linux_alternatives tests.test_setup_php_linux tests.test_php_linux_qualification tests.test_php_linux_qualification_workflow tests.test_php_linux_configuration coverage xml -o coverage.xml - uses: SonarSource/sonarqube-scan-action@d209202bc7d53ff1cc128f7f907dac145c9d6ae9 # v8.3.0 env: diff --git a/conformance/README.md b/conformance/README.md index e975eac..cc8a278 100644 --- a/conformance/README.md +++ b/conformance/README.md @@ -141,6 +141,11 @@ actions work. Report its scope separately from the stable protocol experiments. ## Evidence scope +Command-admission retry experiments under #216 can reuse the +[committed-response-loss fixture](command-keys/README.md). Its socket control +proves response loss after upstream success; the installed SDK/CLI adapter and +producer-owned legacy fixtures supply the remaining admission/retention proof. + A pinned or successfully installed SDK is not evidence that its worker executed an experiment. For each claimed language or cross-language direction, check the scenario-level result for the actual caller/parent and worker/child runtime, diff --git a/conformance/command-keys/README.md b/conformance/command-keys/README.md new file mode 100644 index 0000000..4eafcf8 --- /dev/null +++ b/conformance/command-keys/README.md @@ -0,0 +1,53 @@ +# Committed command-response loss + +This shared transport fixture supports [#216](https://github.com/durable-workflow/.github/issues/216). +It forwards one selected request to a loopback HTTP runtime, reads the complete +successful upstream response, then closes the client connection before sending +an HTTP status line. The next request forwards normally. The fixture does not +retry a request or decide whether command admission was idempotent. + +`CommittedResponseLoss(upstream)` accepts only `http://127.0.0.1:` with a +canonical explicit port. Credentials, hostnames, paths, query and fragment +components refuse. It binds an ephemeral loopback port. Start its +`serve_forever` method in a thread and route the installed client through `url`. +Call `arm(method, path)` with the exact origin-form admission route before +the selected request. Query strings and fragments refuse before forwarding or +recording; put no credentials in route identifiers. The caller owns shutdown, the runtime, synthetic data and cleanup. + +Requests and responses each have a fixed 1 MiB wire-body cap and bounded socket +timeouts. POST requires one canonical Content-Length; request transfer encodings +refuse. GET is bodyless and may omit Content-Length. Complete bounded chunked +upstream responses are supported. Invalid, oversized or incomplete upstream +responses produce a generic 502 and no success record. A matching valid request +consumes the arm even if its upstream attempt fails; re-arming is explicit. +At most 1024 metadata records, including reserved in-flight slots, are allowed. +Capacity exhaustion returns a generic 503 before contacting upstream and does +not consume the arm. Failed upstream attempts release their reserved slot. + +`records` contains method/path, request byte count/hash, complete upstream status, +response byte count/hash and the dropped flag. It retains no headers, parsed +JSON, caller keys, results or payloads. SDK adapters compare their own request +and retained-admission identities independently. Use synthetic inputs and +retain only the metadata needed by the experiment. + +Run the socket controls without dependencies or an external runtime: + +```sh +python3 -I -S -B conformance/command-keys/test_committed_response_loss.py +``` + +The controls prove upstream success was complete before the client received +`RemoteDisconnected`, and that one incoming request caused exactly one upstream +request. It also checks one-shot path selection, ordinary forwarding, errors +and numeric loopback confinement. Negative controls cover framing, byte caps, +incomplete upstream responses and body/header evidence minimization. The earlier +CLI fixture is preserved in Git history; this correction tightens its reusable +authority, resource and evidence boundaries. + +An SDK/CLI adapter still verifies the actual published tuple, original/generated +key, durable admission count, concurrent identical retries, changed-intent +conflicts, closure/continuation, namespace/run fences and retained identities. +Record those outcomes separately from this transport control. Producer-owned +legacy-shape and physical-upgrade seeds remain separate; the proxy creates +neither. The [original published reproduction](https://github.com/durable-workflow/.github/issues/216#issuecomment-6296529480) +describes the earlier loss scenarios and is not current installed acceptance. diff --git a/conformance/command-keys/committed_response_loss.py b/conformance/command-keys/committed_response_loss.py new file mode 100644 index 0000000..88b0094 --- /dev/null +++ b/conformance/command-keys/committed_response_loss.py @@ -0,0 +1,195 @@ +"""Bounded loopback fixture: consume upstream success, then lose one response.""" +import hashlib +import http.client +import http.server +import re +import socket +import threading + + +MAX_REQUEST_BYTES = 1_048_576 +MAX_RESPONSE_BYTES = 1_048_576 +MAX_RECORDS = 1024 +SOCKET_TIMEOUT_SECONDS = 40 + + +def origin_path(path): + return (isinstance(path, str) and len(path) <= 4096 and not path.startswith("//") + and re.fullmatch(r"/[^\x00-\x20\x7f#?]*", path) is not None) + + +def content_length(headers, required): + values = headers.get_all("Content-Length", []) + if not values and not required: + return None + if len(values) != 1 or re.fullmatch(r"0|[1-9]\d{0,6}", values[0], re.ASCII) is None: + raise ValueError("A single canonical bounded Content-Length is required") + return int(values[0]) + + +class RequestRefused(ValueError): + def __init__(self, status, reason): + super().__init__(reason) + self.status = status + + +class CommittedResponseLoss(http.server.ThreadingHTTPServer): + daemon_threads = True + + def __init__(self, upstream): + match = re.fullmatch(r"http://127\.0\.0\.1:([1-9]\d{0,4})", upstream, re.ASCII) if isinstance(upstream, str) else None + if match is None or not 1 <= int(match.group(1)) <= 65535: + raise ValueError("Fixture upstream must be canonical numeric loopback HTTP with an explicit port") + self.upstream_port = int(match.group(1)) + self.records = [] + self.pending_records = 0 + self.armed = None + self.lock = threading.Lock() + super().__init__(("127.0.0.1", 0), _Handler) + + def arm(self, method, path): + if method not in {"GET", "POST"} or not origin_path(path): + raise ValueError("Arm requires an exact supported method and origin-form path") + with self.lock: + if self.armed is not None: + raise ValueError("a response loss is already armed") + self.armed = (method, path) + + def reserve_record(self, method, path): + with self.lock: + if len(self.records) + self.pending_records >= MAX_RECORDS: + raise RequestRefused(503, "Fixture evidence capacity exhausted") + self.pending_records += 1 + selected = self.armed == (method, path) + if selected: + self.armed = None + return selected + + def finish_record(self, record): + with self.lock: + self.pending_records -= 1 + if record is not None: + self.records.append(record) + + @property + def url(self): + return f"http://127.0.0.1:{self.server_port}" + + +class _Handler(http.server.BaseHTTPRequestHandler): + timeout = SOCKET_TIMEOUT_SECONDS + + def log_message(self, *_args): + # Evidence uses closed metadata records rather than HTTP request logs. + pass + + def send_error(self, code, message=None, explain=None): + # Base parser messages can include an untrusted raw request line. + reasons = {400: "Fixture request framing refused", 413: "Fixture byte cap exceeded", + 501: "Fixture method unsupported", 502: "Fixture upstream response unverified", + 503: "Fixture evidence capacity exhausted"} + super().send_error(code, reasons.get(code, "Fixture request refused"), + "The fixture refused the request without retaining its content.") + + def request_body(self): + original_path = self.requestline.split()[1] + if (self.request_version not in {"HTTP/1.0", "HTTP/1.1"} or original_path != self.path + or not origin_path(original_path) or self.headers.get_all("Transfer-Encoding", [])): + raise RequestRefused(400, "Fixture requires an origin path and no request transfer encoding") + try: + length = content_length(self.headers, self.command == "POST") + except ValueError as error: + raise RequestRefused(400, "Fixture request length is invalid") from error + if self.command == "GET" and length not in (None, 0): + raise RequestRefused(400, "Fixture GET requests must be bodyless") + length = length or 0 + if length > MAX_REQUEST_BYTES: + raise RequestRefused(413, "Fixture request exceeds its fixed byte cap") + self.connection.settimeout(SOCKET_TIMEOUT_SECONDS) + try: + body = self.rfile.read(length) + except OSError as error: + raise RequestRefused(400, "Fixture request body is incomplete") from error + if len(body) != length: + raise RequestRefused(400, "Fixture request body is incomplete") + return body + + def response_body(self, response): + encodings = response.headers.get_all("Transfer-Encoding", []) + length = content_length(response.headers, False) + if encodings and (encodings != ["chunked"] or length is not None): + raise ValueError("Fixture upstream framing is unsupported") + if length is not None and length > MAX_RESPONSE_BYTES: + raise ValueError("Fixture upstream response exceeds its fixed byte cap") + body = response.read(MAX_RESPONSE_BYTES + 1) + if len(body) > MAX_RESPONSE_BYTES or (length is not None and len(body) != length): + raise ValueError("Fixture upstream body is oversized or incomplete") + if response.read(1): + raise ValueError("Fixture upstream body exceeds its fixed byte cap") + return body + + def upstream_response(self, raw): + # The destination address is literal; only the validated port varies. + upstream = http.client.HTTPConnection("127.0.0.1", self.server.upstream_port, + timeout=SOCKET_TIMEOUT_SECONDS) + try: + upstream.connect() + if upstream.sock.getpeername()[0] != "127.0.0.1": + raise ValueError("Fixture upstream peer is not loopback") + headers = {k: v for k, v in self.headers.items() + if k.lower() not in {"host", "connection", "transfer-encoding", "proxy-authorization"}} + headers["Connection"] = "close" + upstream.request(self.command, self.path, body=raw or None, headers=headers) + response = upstream.getresponse() + body = self.response_body(response) + return response.status, response.getheaders(), body + finally: + upstream.close() + + def refuse(self, status): + self.close_connection = True + try: + self.send_error(status) + except OSError: + # A failed client may already have closed its side of the fixture. + pass + + def forward(self): + try: + raw = self.request_body() + selected = self.server.reserve_record(self.command, self.path) + except RequestRefused as error: + self.refuse(error.status) + return + try: + status, headers, body = self.upstream_response(raw) + drop = selected and 200 <= status < 300 + record = {"method": self.command, "path": self.path, + "request_body_bytes": len(raw), "request_body_sha256": hashlib.sha256(raw).hexdigest(), + "upstream_status": status, "upstream_complete": True, + "upstream_body_bytes": len(body), "upstream_body_sha256": hashlib.sha256(body).hexdigest(), + "dropped": drop} + except (http.client.HTTPException, OSError, ValueError): + self.server.finish_record(None) + self.refuse(502) + return + self.server.finish_record(record) + self.close_connection = True + try: + if drop: + self.connection.shutdown(socket.SHUT_RDWR) + self.connection.close() + return + self.send_response(status) + for key, value in headers: + if key.lower() not in {"content-length", "transfer-encoding", "connection", "server", "date"}: + self.send_header(key, value) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + except OSError: + # Unplanned client loss is never recorded as intentional injection. + pass + + do_GET = forward + do_POST = forward diff --git a/conformance/command-keys/test_committed_response_loss.py b/conformance/command-keys/test_committed_response_loss.py new file mode 100644 index 0000000..9fe0d75 --- /dev/null +++ b/conformance/command-keys/test_committed_response_loss.py @@ -0,0 +1,264 @@ +"""Socket-level controls for the shared command-admission loss fixture.""" + +import concurrent.futures +import hashlib +import http.client +import http.server +import importlib.util +import json +from pathlib import Path +import socket +import threading +import unittest +from unittest import mock + + +SOURCE = Path(__file__).with_name("committed_response_loss.py") +SPEC = importlib.util.spec_from_file_location("committed_response_loss", SOURCE) +MODULE = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(MODULE) +CommittedResponseLoss = MODULE.CommittedResponseLoss + + +class Upstream(http.server.BaseHTTPRequestHandler): + def log_message(self, *_arguments): + pass + + def do_POST(self): + request = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + self.reply(request) + + def do_GET(self): + self.reply({"request_id": "synthetic-get"}) + + def reply(self, request): + self.server.requests.append((self.path, request)) + if self.server.response_gate is not None: + self.server.response_started.set() + self.server.response_gate.wait(timeout=2) + self.close_connection = True + if self.server.fail_before_headers: + return + document = {"command_id": "synthetic-command", "request_id": request["request_id"], + "unexpected_secret": "synthetic-response-secret"} + body = self.server.body_override or json.dumps(document).encode() + self.server.response_body = body + self.send_response(self.server.response_status) + if self.server.framing in {"chunked", "incomplete-chunked"}: + self.send_header("Transfer-Encoding", "chunked") + body = f"{len(body):x}\r\n".encode() + body + b"\r\n" + if self.server.framing == "chunked": + body += b"0\r\n\r\n" + elif self.server.framing == "length": + length = self.server.length_override + self.send_header("Content-Length", str(len(body) if length is None else length)) + self.end_headers() + try: + self.wfile.write(body) + except OSError: + # A refusal may close the proxy before this synthetic writer ends. + pass + + +class CommittedResponseLossTest(unittest.TestCase): + def setUp(self): + self.upstream = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Upstream) + self.upstream.response_status = 202 + self.upstream.requests = [] + self.upstream.fail_before_headers = False + self.upstream.body_override = None + self.upstream.length_override = None + self.upstream.framing = "length" + self.upstream.response_gate = None + self.upstream.response_started = threading.Event() + self.serve(self.upstream) + self.proxy = CommittedResponseLoss(f"http://127.0.0.1:{self.upstream.server_port}") + self.serve(self.proxy) + + def serve(self, server): + thread = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01}, daemon=True) + thread.start() + + def close(): + server.shutdown() + server.server_close() + thread.join(timeout=2) + + self.addCleanup(close) + + def request(self, path="/commit"): + client = http.client.HTTPConnection("127.0.0.1", self.proxy.server_port, timeout=2) + try: + client.request("POST", path, body=json.dumps({"request_id": "synthetic-key", "token": "synthetic-request-secret"}), + headers={"Authorization": "Bearer synthetic-header-token"}) + response = client.getresponse() + return response.status, json.loads(response.read()) + finally: + client.close() + + def raw_request(self, headers, body=b"", path="/commit", method="POST"): + with socket.create_connection(("127.0.0.1", self.proxy.server_port), timeout=2) as client: + head = f"{method} {path} HTTP/1.1\r\nHost: fixture\r\n{headers}\r\n" + client.sendall(head.encode("ascii") + body) + client.shutdown(socket.SHUT_WR) + response = http.client.HTTPResponse(client) + response.begin() + return response.status, response.read(8192) + + def test_complete_upstream_success_precedes_loss_before_client_status_line(self): + self.proxy.arm("POST", "/commit") + with self.assertRaises(http.client.RemoteDisconnected): + self.request() + self.assertEqual(1, len(self.upstream.requests)) + record = self.proxy.records[0] + self.assertEqual(202, record["upstream_status"]) + self.assertTrue(record["dropped"]) + self.assertEqual(len(self.upstream.response_body), record["upstream_body_bytes"]) + self.assertEqual(hashlib.sha256(self.upstream.response_body).hexdigest(), record["upstream_body_sha256"]) + self.assertTrue(record["upstream_complete"]) + self.assertEqual({"method", "path", "request_body_bytes", "request_body_sha256", "upstream_status", + "upstream_complete", "upstream_body_bytes", "upstream_body_sha256", "dropped"}, set(record)) + self.assertNotIn("synthetic-header-token", json.dumps(self.proxy.records)) + self.assertNotIn("synthetic-request-secret", json.dumps(self.proxy.records)) + self.assertNotIn("synthetic-response-secret", json.dumps(self.proxy.records)) + self.assertNotIn("synthetic-key", json.dumps(self.proxy.records)) + status, body = self.request() + self.assertEqual(202, status) + self.assertEqual("synthetic-command", body["command_id"]) + self.assertEqual(2, len(self.upstream.requests)) + self.assertFalse(self.proxy.records[1]["dropped"]) + + def test_loss_is_one_shot_for_only_the_exact_method_and_path(self): + self.proxy.arm("POST", "/commit") + with self.assertRaises(ValueError): + self.proxy.arm("POST", "/another") + self.assertEqual(202, self.request("/different")[0]) + self.assertFalse(self.proxy.records[0]["dropped"]) + with self.assertRaises(http.client.RemoteDisconnected): + self.request() + self.assertEqual(2, len(self.upstream.requests)) + self.assertTrue(self.proxy.records[1]["dropped"]) + + def test_selected_non_success_is_forwarded_and_not_retried(self): + self.upstream.response_status = 409 + self.proxy.arm("POST", "/commit") + self.assertEqual(409, self.request()[0]) + self.assertEqual(1, len(self.upstream.requests)) + self.assertFalse(self.proxy.records[0]["dropped"]) + self.assertIsNone(self.proxy.armed) + + def test_upstream_must_be_loopback_http(self): + for address in ("https://127.0.0.1:1234", "http://example.invalid:1234", "file:///tmp/fixture", + "http://localhost:1234", "http://127.0.0.2:1234", "http://127.0.0.1", + "http://user:password@127.0.0.1:1234", "http://127.0.0.1:1234/", + "http://127.0.0.1:1234?q=x", "http://127.0.0.1:1234#fragment", + "http://127.0.0.1:0", "http://127.0.0.1:65536", "http://127.0.0.1:0123", "http://127.0.0.1:1٢3", "http://127.0.0.1:1２3", None): + with self.subTest(address=address), self.assertRaises(ValueError): + CommittedResponseLoss(address) + + def test_invalid_request_framing_is_refused_before_forwarding(self): + cases = [("", b"", 400), ("Content-Length: -1\r\n", b"", 400), + ("Content-Length: 01\r\n", b"x", 400), ("Content-Length: 1 \r\n", b"x", 400), + ("Content-Length: 0\r\nContent-Length: 0\r\n", b"", 400), + (f"Content-Length: {MODULE.MAX_REQUEST_BYTES + 1}\r\n", b"", 413), + ("Transfer-Encoding: chunked\r\nContent-Length: 0\r\n", b"", 400), + ("Content-Length: 2\r\n", b"x", 400)] + for headers, body, status in cases: + with self.subTest(headers=headers): + self.assertEqual(status, self.raw_request(headers, body)[0]) + self.assertEqual([], self.upstream.requests) + self.assertEqual([], self.proxy.records) + + def test_only_origin_form_paths_can_be_forwarded_or_armed(self): + for path in ("http://example.invalid/", "//example.invalid/", "/path#fragment", "/commit?token=synthetic-private-query"): + with self.subTest(path=path): + status, body = self.raw_request("Content-Length: 0\r\n", path=path) + self.assertEqual(400, status) + self.assertNotIn(b"synthetic-private-query", body) + self.assertEqual([], self.proxy.records) + self.assertEqual([], self.upstream.requests) + with self.assertRaises(ValueError): + self.proxy.arm("POST", path) + + def test_content_length_accepts_only_ascii_digits(self): + for value in ("1٢3", "1２3", "١23", "+1", "1.0"): + headers = http.client.HTTPMessage() + headers.add_header("Content-Length", value) + with self.subTest(value=value), self.assertRaises(ValueError): + MODULE.content_length(headers, True) + + def test_record_capacity_reserves_inflight_before_any_upstream_contact(self): + with mock.patch.object(MODULE, "MAX_RECORDS", 1): + self.upstream.response_gate = threading.Event() + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: + first = executor.submit(self.request) + try: + self.assertTrue(self.upstream.response_started.wait(timeout=2)) + self.assertEqual(1, self.proxy.pending_records) + self.proxy.arm("POST", "/blocked") + body = b'{"request_id":"synthetic-key"}' + status, _ = self.raw_request(f"Content-Length: {len(body)}\r\n", body, path="/blocked") + self.assertEqual(503, status) + self.assertEqual(1, len(self.upstream.requests)) + self.assertEqual([], self.proxy.records) + self.assertEqual(("POST", "/blocked"), self.proxy.armed) + finally: + self.upstream.response_gate.set() + self.assertEqual(202, first.result(timeout=2)[0]) + self.assertEqual(1, len(self.proxy.records)) + self.assertEqual(0, self.proxy.pending_records) + status, _ = self.raw_request(f"Content-Length: {len(body)}\r\n", body, path="/blocked") + self.assertEqual(503, status) + self.assertEqual(1, len(self.upstream.requests)) + self.assertEqual(1, len(self.proxy.records)) + + def test_bodyless_get_needs_no_content_length(self): + status, body = self.raw_request("", method="GET") + self.assertEqual(202, status) + self.assertEqual("synthetic-get", json.loads(body)["request_id"]) + self.assertEqual(0, self.proxy.records[0]["request_body_bytes"]) + + def test_parser_errors_do_not_echo_request_content(self): + marker = "synthetic-private-request-line" + for method, path in (("POST " + marker, "/commit"), ("POST", "/" + marker + "#fragment")): + with self.subTest(method=method, path=path): + status, body = self.raw_request("Content-Length: 0\r\n", method=method, path=path) + self.assertEqual(400, status) + self.assertNotIn(marker.encode(), body) + self.assertEqual([], self.proxy.records) + self.assertEqual([], self.upstream.requests) + + def test_oversized_or_incomplete_upstream_is_not_an_injected_loss(self): + body = b'{"request_id":"synthetic-key"}' + marker = b"private-upstream-body" + oversized = marker + b"x" * (MODULE.MAX_RESPONSE_BYTES + 1 - len(marker)) + cases = [("length", MODULE.MAX_RESPONSE_BYTES + 1, marker, False), + ("close", None, oversized, False), + ("length", 40, marker, False), + ("incomplete-chunked", None, marker, False), + ("length", None, marker, True)] + for framing, length, response, fail in cases: + with self.subTest(framing=framing, length=length, fail=fail): + self.upstream.framing = framing + self.upstream.length_override = length + self.upstream.body_override = response + self.upstream.fail_before_headers = fail + self.proxy.arm("POST", "/commit") + status, reply = self.raw_request(f"Content-Length: {len(body)}\r\n", body) + self.assertEqual(502, status) + self.assertNotIn(marker, reply) + self.assertEqual([], self.proxy.records) + self.assertIsNone(self.proxy.armed) + self.assertEqual(0, self.proxy.pending_records) + + def test_complete_chunked_upstream_can_supply_a_bounded_loss(self): + self.upstream.framing = "chunked" + self.proxy.arm("POST", "/commit") + with self.assertRaises(http.client.RemoteDisconnected): + self.request() + self.assertTrue(self.proxy.records[0]["dropped"]) + self.assertTrue(self.proxy.records[0]["upstream_complete"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_command_response_loss.py b/tests/test_command_response_loss.py new file mode 100644 index 0000000..4fcf8f7 --- /dev/null +++ b/tests/test_command_response_loss.py @@ -0,0 +1,11 @@ +"""Include the shared socket controls in repository checks and real coverage.""" +import importlib.util +from pathlib import Path + + +def load_tests(loader, _tests, _pattern): + source = Path(__file__).resolve().parents[1] / 'conformance/command-keys/test_committed_response_loss.py' + spec = importlib.util.spec_from_file_location('command_response_loss_controls', source) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return loader.loadTestsFromTestCase(module.CommittedResponseLossTest) diff --git a/tests/test_native_policy_workflow.py b/tests/test_native_policy_workflow.py index 93c5d48..597bd68 100644 --- a/tests/test_native_policy_workflow.py +++ b/tests/test_native_policy_workflow.py @@ -28,6 +28,13 @@ def test_only_central_pinned_code_is_checked_out_without_persisted_credentials(s self.assertIs(False, settings["persist-credentials"]) self.assertEqual(".native-policy-control", settings["path"]) + def test_response_loss_adoption_binds_accepted_runtime_and_exact_workflow_fixtures(self): + checkout = self.workflow["jobs"]["policy"]["steps"][0] + self.assertEqual("d4770e6555fdb59b044f2e5919e95a24c0a25073", checkout["with"]["ref"]) + for name in ("ci", "sonar"): + self.assertEqual((ROOT / f"tests/fixtures/{name}-response-loss-source.yml").read_bytes(), + (ROOT / f".github/workflows/{name}.yml").read_bytes()) + def test_no_issuer_key_or_write_permission_is_introduced(self): self.assertEqual({"contents": "read", "pull-requests": "read"}, self.workflow["permissions"]) raw = (ROOT / ".github/workflows/native-policy.yml").read_text()