diff --git a/tools/uci/test_https_local.py b/tools/uci/test_https_local.py index 6884128..d1817cf 100644 --- a/tools/uci/test_https_local.py +++ b/tools/uci/test_https_local.py @@ -47,7 +47,17 @@ - Writes a sentinel on completion 5. Trigger with SYS 16896 via keyboard buffer. 6. Poll sentinel for up to 120 s (handshake is ~13-15 s at 48 MHz). - 7. Assert response body contains "HELLO FROM TLS SERVER". + 7. Assert BOTH sides: + - C64 side: http_resp_buf holds the complete "HELLO FROM TLS SERVER" + body (screen RAM is diagnostic only — see _check_c64_result). + - Server side: the listener completed the handshake, recorded no TLS + error, and decrypted the GET the C64 was told to send (see + _check_server_result). Skipped under EXTERNAL_LISTENER=1, which by + contract has no inline listener. + +Offline re-check of an archived run (no hardware): + ./test_https_local.py --check-artifact [--host IP] [--path P] +re-runs the server-side criteria against that run's server_result.json. """ from __future__ import annotations @@ -288,6 +298,156 @@ def _run_https_server(srv: socket.socket, ctx: ssl.SSLContext, pass +def _check_c64_result(body_ascii: str, screen_text: str) -> list[str]: + """C64-side pass criteria. Returns a list of problems; empty ⇒ pass. + + The only accepted criterion is that ``http_resp_buf`` holds the complete + expected body. Screen RAM is diagnostic, never evidence. + + History (audit finding F5): this used to fall back to passing the run + whenever the 5 characters ``HELLO`` appeared anywhere in the 1000 bytes + of screen RAM. That branch was reachable *only* after the body assertion + had already failed, so it substituted a weaker criterion at exactly the + moment the strong one did not hold — a truncated body, a mis-decrypted + body that happened to keep its first word, or a stale ``HELLO`` left on + screen by an earlier run all passed. The fallback is gone; a screen-RAM + hit without the body is now reported as a *reason the run failed*. + """ + problems: list[str] = [] + if EXPECTED_BODY not in body_ascii: + problems.append( + f"http_resp_buf does not contain the expected body " + f"{EXPECTED_BODY!r} (got {body_ascii[:120]!r})" + ) + if "HELLO" in screen_text.upper(): + problems.append( + "screen RAM contains 'HELLO' but that is not a pass — only " + "the complete body in http_resp_buf counts (audit F5)" + ) + return problems + + +def _server_request_bytes(server_result: dict) -> bytes | None: + """Normalize the listener's ``request`` field to bytes, or None if absent. + + Accepts both shapes: the in-process dict (raw ``bytes``) and the + ``server_result.json`` on-disk form written by + ``_serialize_server_result`` (``{"__type__": "bytes-b64", ...}``), so the + same checker runs live and against an archived run directory. + """ + req = server_result.get("request") + if req is None: + return None + if isinstance(req, (bytes, bytearray)): + return bytes(req) + if isinstance(req, dict) and req.get("__type__") == "bytes-b64": + try: + return base64.b64decode(req.get("b64", "")) + except Exception: + return None + if isinstance(req, str): + return req.encode("utf-8", errors="replace") + return None + + +def _check_server_result(server_result: dict, *, + expect_host: str | None, + expect_path: str = "/") -> list[str]: + """Server-side pass criteria. Returns a list of problems; empty ⇒ pass. + + Audit finding F6: every run already recorded what the listener observed + into ``server_result.json`` — handshake completion, the decrypted + request, any TLS error — and the pass criteria never read a byte of it. + Every assertion was made against C64-side memory and screen RAM, i.e. + against state the client itself produces. The listener's record is the + one piece of evidence the client cannot fabricate, so it is now part of + the verdict. + + Which fields are load-bearing, and why these and not more: + + ``error`` — decisive. Set when the TLS handshake or the socket + failed on the server side. A run that reached + ``SSLEOFError: UNEXPECTED_EOF_WHILE_READING`` did not + complete a TLS session with this listener, whatever + the C64's RAM says afterwards. + ``listening`` — the listener reached ``accept()``. Absent ⇒ there was + no server for the C64 to have talked to. + ``client_addr`` — the listener accepted a connection. Absent ⇒ the C64 + never reached this listener; anything in http_resp_buf + is then stale or fabricated, not this run's evidence. + ``request`` — the decrypted request must be the one the C64 was + configured to send. This is the strongest link + between the two sides: the server can only produce + these plaintext bytes by having completed the + handshake and derived the same application keys. + + All three of ``listening`` / ``client_addr`` / ``request`` are required, + so the function fails closed: an empty record — listener thread never + started, crashed before recording anything, artifact absent — is an + inconclusive check and therefore a failure, never an implicit "no error + recorded". + + Deliberately NOT asserted, to avoid failing healthy runs: + + - Byte-exact request equality. Only the request line and the ``Host:`` + line are checked — both are values *this script* programmed into the + C64, so they cannot drift accidentally. Trailing headers + (``Connection: close``) are src/http.s's business; a future header + change should not fail the e2e oracle. + - ``request`` in EXTERNAL_LISTENER mode. There is no inline listener + then, ``server_result`` is empty by construction, and the documented + contract is that pass criteria come from C64 state only. The caller + skips this whole function in that mode and says so out loud. + - The listener sets ``request`` to ``b""`` when the + post-handshake ``recv`` times out; that is treated as a failure, not + as an absent field, because it means the session never carried the + GET. + """ + problems: list[str] = [] + + err = server_result.get("error") + if err: + problems.append(f"listener recorded an error: {err}") + + if not server_result.get("listening"): + problems.append( + "listener never reported `listening` — no server side to this run" + ) + + if server_result.get("client_addr") is None: + problems.append( + "listener never accepted a connection (`client_addr` absent) — " + "the C64 did not reach this listener" + ) + + req = _server_request_bytes(server_result) + if req is None: + problems.append( + "listener recorded no decrypted request — the TLS session never " + "carried the GET" + ) + elif req == b"": + problems.append( + "listener timed out waiting for the request after the handshake" + ) + else: + want_line = b"GET " + expect_path.encode("ascii") + b" HTTP/1." + if not req.startswith(want_line): + problems.append( + f"decrypted request does not start with {want_line!r} " + f"(got {req[:40]!r})" + ) + if expect_host is not None: + want_host = b"Host: " + expect_host.encode("ascii") + if want_host not in req: + problems.append( + f"decrypted request lacks {want_host!r} " + f"(got {req[:80]!r})" + ) + + return problems + + def _load_labels() -> dict[str, int]: # c64-test-harness Labels is a Mapping since 0.12.4 (JC-000/c64-test-harness#64) # and parses both C: and non-C (REU/bank) label lines since #62. @@ -1388,23 +1548,38 @@ def main() -> int: except Exception: pass - if EXPECTED_BODY in body_ascii: - print(f"\nPASS: http_resp_buf contains '{EXPECTED_BODY}'") - outcome = "PASS" - exit_code = 0 - return exit_code + problems = _check_c64_result(body_ascii, screen_text) - if "HELLO" in screen_text.upper(): - print(f"\nPASS: screen RAM contains HELLO " - f"(body in resp_buf may differ in encoding)") - outcome = "PASS" - exit_code = 0 + # Server-side corroboration (audit F6). The listener's record is the + # only evidence in this test the client cannot produce on its own, so + # a run passes only when both sides agree. Skipped — loudly — under + # EXTERNAL_LISTENER=1, where by contract there is no inline listener + # and no server_result to read. + if EXTERNAL_LISTENER: + print("\nNOTE: EXTERNAL_LISTENER=1 — server-side criteria " + "skipped (no inline listener); C64-side state only") + else: + server_problems = _check_server_result( + server_result, + expect_host=test_host_ip, + expect_path=path_str.rstrip(b"\x00").decode("ascii"), + ) + problems.extend(server_problems) + if not server_problems: + print("\nServer-side check: listener completed the handshake " + "and decrypted the expected GET") + + if problems: + print("\nFAIL: pass criteria not met:", file=sys.stderr) + for p in problems: + print(f" - {p}", file=sys.stderr) + outcome = "FAIL" + exit_code = 1 return exit_code - print(f"\nFAIL: expected '{EXPECTED_BODY}' not found in response" - f" or screen", file=sys.stderr) - outcome = "FAIL" - exit_code = 1 + print(f"\nPASS: http_resp_buf contains '{EXPECTED_BODY}'") + outcome = "PASS" + exit_code = 0 return exit_code finally: @@ -1499,5 +1674,71 @@ def main() -> int: print(f"Released DeviceLock({HOST})") +def _check_artifact_main(argv: list[str]) -> int: + """`--check-artifact [--host IP] [--path P]` + + Re-run the server-side pass criteria (audit F6) against an archived run + without touching hardware. Exists so the criteria are testable — and so a + stored run can be re-adjudicated after the criteria change. + """ + if not argv: + print("usage: test_https_local.py --check-artifact " + " [--host IP] [--path P]", + file=sys.stderr) + return 2 + target = Path(argv[0]) + host: str | None = None + path = "/" + rest = argv[1:] + while rest: + flag = rest.pop(0) + if flag == "--host" and rest: + host = rest.pop(0) + elif flag == "--path" and rest: + path = rest.pop(0) + else: + print(f"unknown argument: {flag}", file=sys.stderr) + return 2 + + if target.is_dir(): + target = target / "server_result.json" + + # Fail closed: an absent or unreadable record is an inconclusive check, + # never an implicit "no error recorded". Same defect shape as the F5/F6 + # fallbacks themselves, one layer up. + if not target.is_file(): + print(f"FAIL: no server_result.json at {target} — the server-side " + f"record is missing, which is not a pass", file=sys.stderr) + return 1 + try: + server_result = json.loads(target.read_text()) + except Exception as exc: + print(f"FAIL: could not read {target}: {exc}", file=sys.stderr) + return 1 + if not isinstance(server_result, dict): + print(f"FAIL: {target} is not a JSON object " + f"(got {type(server_result).__name__})", file=sys.stderr) + return 1 + print(f"server_result : {target}") + print(f" listening : {server_result.get('listening', False)}") + print(f" client_addr : {server_result.get('client_addr')}") + print(f" request : {_server_request_bytes(server_result)!r}") + print(f" error : {server_result.get('error', '')}") + if host is None: + print(" (no --host given; the Host-header check is skipped)") + + problems = _check_server_result( + server_result, expect_host=host, expect_path=path) + if problems: + print("\nFAIL: server-side criteria not met:", file=sys.stderr) + for p in problems: + print(f" - {p}", file=sys.stderr) + return 1 + print("\nPASS: server-side criteria met") + return 0 + + if __name__ == "__main__": + if len(sys.argv) > 1 and sys.argv[1] == "--check-artifact": + raise SystemExit(_check_artifact_main(sys.argv[2:])) raise SystemExit(main())