diff --git a/CLAUDE.md b/CLAUDE.md index 51d3e7b..958d45b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -337,6 +337,12 @@ overridable via the `U64_HOST` environment variable) and use - `test_http_local.py` — HTTP GET against a local test server - `test_http_live.py` — HTTP GET against a real internet host (requires internet access from the U64E) + - `test_https_bad_finished.py` — the client must ABORT on a forged server + Finished. Uses the hand-rolled + `tools/https_e2e/evil_listener.py` rather than + stock `ssl`. `FINISHED_MODE=good` is the control + and must be run first. See "Negative-path + coverage — the server Finished" under Smoke tests. - `test_https_local.py` — HTTPS e2e scaffolding against a local TLS 1.3 listener (ECDSA-P256 cert from `tools/https_e2e/certs/`). DMAs a 6502 stub @@ -1083,9 +1089,46 @@ the TLS state machine. For a quick sanity check after a build: - `tools/test_tls_handshake.py` — full handshake state machine - `tools/test_http.py` — HTTP request/response build + parse - `tools/test_x509.py` — X.509 parser + - `tools/test_finished_verify.py` — server-Finished **rejection** path + (18 cases, 2 vector sets; see below) All 7 pass as of the ca65-conversion branch (97/97 assertions). +### Negative-path coverage — the server Finished + +`tools/test_finished_verify.py` and `tools/uci/test_https_bad_finished.py` +exist because an audit found the client's Finished-mismatch abort had **no +test at all**: inverting the mismatch branch (`sec` -> `clc` in +`tls_verify_finished`, `src/tls_keyschedule.s`) left the full hardware e2e +reaching HTTP 200 with the correct body. Every listener the suite talks to +sends a *correct* Finished, so nothing ever exercised the reject. + + - `tools/test_finished_verify.py` (VICE) drives `tls_verify_finished` + directly over DMA with a 6502 carry-latching stub — no P-register read, + and an unwritten latch is reported as inconclusive, never a pass. Two + (secret, transcript) vector sets x 9 cases each, including the two + realistic attacks: a valid HMAC under the wrong secret, and one over the + wrong transcript. + - `tools/uci/test_https_bad_finished.py` (U64E/C64U) is the end-to-end + version, against `tools/https_e2e/evil_listener.py` — a hand-rolled TLS 1.3 + server (real X25519, real key schedule, real ChaCha20-Poly1305 records, + real P-256 CertificateVerify) that flips **one bit** of the server + Finished `verify_data` before encryption. Corrupting the *ciphertext* + instead would break the Poly1305 tag and get rejected at `aead_decrypt`, + never reaching the Finished comparison — which is why stock `ssl` cannot + produce this test case and the server side is written out by hand. + `FINISHED_MODE=good` runs the identical server with a correct Finished and + is the mandatory control; `FINISHED_MODE=bad` (default) is the test. + The oracle uses `tls_last_state`, which `src/tls13.s:@error` stashes on + abort: `tls_state=$FF` + `tls_last_state=6 (FINISHED)` proves the abort + happened at Finished rather than earlier at Certificate (4) or + CertificateVerify (5). Server-side evidence (`client_accepted_finished`) + is asserted too. + +Both flip under the mutant: 18/18 -> 2/18 in VICE, PASS -> FAIL on the U64E. +Note `evil_listener.py` is a test fixture, not a TLS stack — it has no +hardening and belongs nowhere near production. + The `tools/uci/` scripts cover the UCI backend on U64E hardware (see the "UCI test scripts" subsection above). diff --git a/tools/https_e2e/evil_listener.py b/tools/https_e2e/evil_listener.py new file mode 100644 index 0000000..f2d0488 --- /dev/null +++ b/tools/https_e2e/evil_listener.py @@ -0,0 +1,634 @@ +"""evil_listener.py — a hand-rolled TLS 1.3 server that can lie. + +Why this is not `ssl.SSLContext` +-------------------------------- +Audit finding F2: nothing in the repo exercised the client's *rejection* of a +bad server Finished. To exercise it, a server has to emit a structurally valid, +correctly-encrypted handshake flight whose Finished verify_data is wrong. +Python's `ssl` module cannot be made to do that — the handshake is entirely +inside OpenSSL and there is no hook between "compute verify_data" and "put it +on the wire". Bit-flipping the ciphertext from outside does not work either: +that breaks the Poly1305 tag, so the client rejects at the AEAD layer and never +reaches the Finished comparison, which would be a false pass for F2. + +So the server side is written out by hand here. That is much less work than it +sounds, because the c64-https client is extremely constrained: + + * exactly one cipher suite, TLS_CHACHA20_POLY1305_SHA256 (0x1303) + * exactly one group, x25519 (0x001d) + * no SNI, empty legacy_session_id, no PSK, no early data, no HRR + * no client certificates + +This module implements only what that client (and, for self-validation, a +stock OpenSSL client) needs. It is a **test fixture, not a TLS stack** — it has +no security review, no state machine hardening, and no business anywhere near +production. + +Modes +----- +``mode="good"`` + A fully correct handshake, then one HTTP response. Used as the control: the + same code that produces the bad flight must also be able to produce a + working one, otherwise a client abort proves nothing about *where* the + client aborted. + +``mode="bad_finished"`` + Identical in every byte except one: a single bit is flipped in the server + Finished ``verify_data`` before it is encrypted. Everything else — the + record layer, the AEAD tag, the certificate, the CertificateVerify + signature, the transcript — is correct, so a conforming client must get all + the way to the Finished HMAC comparison and reject *there*. The server then + records what the client actually did, in ``client_accepted_finished``: a + client that goes on to send its own Finished did not check ours. + +Self-validation +--------------- +``python3 tools/https_e2e/evil_listener.py --selftest`` runs both modes against +Python's own `ssl` client: ``good`` must complete the handshake and return the +body, ``bad_finished`` must raise an SSL error mentioning a bad MAC / decrypt +error. If that self-test does not pass, no conclusion drawn from a C64 run +against this server is worth anything. +""" + +from __future__ import annotations + +import hashlib +import hmac +import os +import socket +import struct +import sys +import threading +import time + +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec, x25519 +from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305 +from cryptography.x509 import load_pem_x509_certificate + +# Record / handshake constants +CT_CHANGE_CIPHER_SPEC = 20 +CT_ALERT = 21 +CT_HANDSHAKE = 22 +CT_APPLICATION_DATA = 23 + +HS_CLIENT_HELLO = 1 +HS_SERVER_HELLO = 2 +HS_ENCRYPTED_EXTENSIONS = 8 +HS_CERTIFICATE = 11 +HS_CERTIFICATE_VERIFY = 15 +HS_FINISHED = 20 + +TLS_CHACHA20_POLY1305_SHA256 = 0x1303 +GROUP_X25519 = 0x001D +SIG_ECDSA_SECP256R1_SHA256 = 0x0403 + +EXT_SUPPORTED_GROUPS = 0x000A +EXT_SUPPORTED_VERSIONS = 0x002B +EXT_KEY_SHARE = 0x0033 + +HASH_LEN = 32 + +DEFAULT_BODY = "HELLO FROM TLS SERVER" + + +class TlsFixtureError(Exception): + """Something about the peer's flight was not what this fixture supports.""" + + +# --------------------------------------------------------------------------- +# Key schedule (RFC 8446 Section 7.1) +# --------------------------------------------------------------------------- + +def _hkdf_extract(salt: bytes, ikm: bytes) -> bytes: + if not salt: + salt = b"\x00" * HASH_LEN + return hmac.new(salt, ikm, hashlib.sha256).digest() + + +def _hkdf_expand(prk: bytes, info: bytes, length: int) -> bytes: + out = b"" + t = b"" + counter = 1 + while len(out) < length: + t = hmac.new(prk, t + info + bytes([counter]), hashlib.sha256).digest() + out += t + counter += 1 + return out[:length] + + +def hkdf_expand_label(secret: bytes, label: bytes, context: bytes, + length: int) -> bytes: + info = struct.pack(">H", length) + info += bytes([6 + len(label)]) + b"tls13 " + label + info += bytes([len(context)]) + context + return _hkdf_expand(secret, info, length) + + +def derive_secret(secret: bytes, label: bytes, transcript_hash: bytes) -> bytes: + return hkdf_expand_label(secret, label, transcript_hash, HASH_LEN) + + +def _sha256(data: bytes) -> bytes: + return hashlib.sha256(data).digest() + + +class TrafficKeys: + """One direction's AEAD state: key, iv, and a sequence number.""" + + def __init__(self, secret: bytes): + self.secret = secret + self.key = hkdf_expand_label(secret, b"key", b"", 32) + self.iv = hkdf_expand_label(secret, b"iv", b"", 12) + self.aead = ChaCha20Poly1305(self.key) + self.seq = 0 + + def _nonce(self) -> bytes: + seq = self.seq.to_bytes(12, "big") + return bytes(a ^ b for a, b in zip(self.iv, seq)) + + def encrypt(self, inner_plaintext: bytes) -> bytes: + length = len(inner_plaintext) + 16 + aad = bytes([CT_APPLICATION_DATA, 0x03, 0x03]) + struct.pack(">H", length) + ct = self.aead.encrypt(self._nonce(), inner_plaintext, aad) + self.seq += 1 + return aad + ct + + def decrypt(self, record: bytes) -> tuple[int, bytes]: + """*record* is a complete TLSCiphertext incl. its 5-byte header.""" + aad = record[:5] + ct = record[5:] + pt = self.aead.decrypt(self._nonce(), ct, aad) + self.seq += 1 + # Strip zero padding, then the inner content type. + i = len(pt) - 1 + while i >= 0 and pt[i] == 0: + i -= 1 + if i < 0: + raise TlsFixtureError("decrypted record is all padding") + return pt[i], pt[:i] + + +# --------------------------------------------------------------------------- +# Wire helpers +# --------------------------------------------------------------------------- + +def _u24(n: int) -> bytes: + return bytes([(n >> 16) & 0xFF, (n >> 8) & 0xFF, n & 0xFF]) + + +def _handshake(msg_type: int, body: bytes) -> bytes: + return bytes([msg_type]) + _u24(len(body)) + body + + +def _plaintext_record(content_type: int, payload: bytes) -> bytes: + return bytes([content_type, 0x03, 0x03]) + struct.pack(">H", len(payload)) + payload + + +class RecordReader: + """Reassembles TLS records from a stream socket.""" + + def __init__(self, sock: socket.socket): + self.sock = sock + self.buf = bytearray() + + def read_record(self, timeout: float) -> bytes | None: + """Return one complete record (header included), or None on EOF.""" + deadline = time.monotonic() + timeout + while True: + if len(self.buf) >= 5: + length = struct.unpack(">H", self.buf[3:5])[0] + if len(self.buf) >= 5 + length: + rec = bytes(self.buf[: 5 + length]) + del self.buf[: 5 + length] + return rec + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("timed out waiting for a TLS record") + self.sock.settimeout(remaining) + chunk = self.sock.recv(4096) + if not chunk: + return None + self.buf += chunk + + +def parse_client_hello(msg: bytes) -> dict: + """Extract what the server needs from a ClientHello handshake message.""" + if not msg or msg[0] != HS_CLIENT_HELLO: + raise TlsFixtureError( + f"expected ClientHello, got handshake type {msg[0] if msg else 'EOF'}" + ) + body = msg[4:] + p = 0 + p += 2 # legacy_version + client_random = body[p:p + 32] + p += 32 + sid_len = body[p] + p += 1 + session_id = body[p:p + sid_len] + p += sid_len + cs_len = struct.unpack(">H", body[p:p + 2])[0] + p += 2 + suites = [ + struct.unpack(">H", body[p + i:p + i + 2])[0] for i in range(0, cs_len, 2) + ] + p += cs_len + comp_len = body[p] + p += 1 + comp_len + ext_total = struct.unpack(">H", body[p:p + 2])[0] + p += 2 + end = p + ext_total + + key_share = None + while p < end: + ext_type = struct.unpack(">H", body[p:p + 2])[0] + ext_len = struct.unpack(">H", body[p + 2:p + 4])[0] + data = body[p + 4:p + 4 + ext_len] + p += 4 + ext_len + if ext_type == EXT_KEY_SHARE: + q = 2 # client_shares list length + while q < len(data): + group = struct.unpack(">H", data[q:q + 2])[0] + klen = struct.unpack(">H", data[q + 2:q + 4])[0] + if group == GROUP_X25519: + key_share = data[q + 4:q + 4 + klen] + break + q += 4 + klen + + if TLS_CHACHA20_POLY1305_SHA256 not in suites: + raise TlsFixtureError( + "client did not offer TLS_CHACHA20_POLY1305_SHA256 (0x1303); " + f"offered {[hex(s) for s in suites]}" + ) + if key_share is None: + raise TlsFixtureError("client sent no x25519 key_share") + + return { + "client_random": client_random, + "session_id": session_id, + "key_share": key_share, + } + + +# --------------------------------------------------------------------------- +# The server +# --------------------------------------------------------------------------- + +class EvilTls13Server: + """One-shot TLS 1.3 server flight, optionally with a corrupted Finished. + + *mode* is ``"good"`` or ``"bad_finished"``. The two modes run the *same* + code path from end to end; they differ only in whether one bit of the + server Finished ``verify_data`` is flipped before encryption. + + Deliberately, the server folds the Finished it actually sent into its own + transcript. A client that wrongly accepts the corrupted Finished therefore + stays in lockstep with the server and completes the handshake normally, + ending at HTTP 200 — so a broken client fails loudly and quickly rather + than hanging and being written off as a flaky timeout. + """ + + def __init__(self, cert_path: str, key_path: str, *, + mode: str = "good", + body: str = DEFAULT_BODY): + if mode not in ("good", "bad_finished"): + raise ValueError(f"unknown mode {mode!r}") + self.mode = mode + self.body = body + + with open(cert_path, "rb") as f: + pem = f.read() + self.cert_der = load_pem_x509_certificate(pem).public_bytes( + serialization.Encoding.DER + ) + with open(key_path, "rb") as f: + self.key = serialization.load_pem_private_key(f.read(), password=None) + if not isinstance(self.key, ec.EllipticCurvePrivateKey): + raise TlsFixtureError("this fixture only signs with ECDSA P-256") + + self.result: dict = { + "mode": mode, + "listening": False, + "client_hello_seen": False, + "server_flight_sent": False, + "finished_corrupted": False, + # The load-bearing one: did the client go on to send its own + # Finished after our (possibly corrupted) Finished? A client that + # checks the server Finished MUST NOT. + "client_accepted_finished": None, + "client_reaction": None, + "client_finished_valid": None, + "request": None, + "response_sent": False, + "client_alert": None, + "error": None, + } + + # -- handshake message builders ---------------------------------------- + + def _server_hello(self, ch: dict, server_pub: bytes) -> bytes: + ext = b"" + ext += struct.pack(">HH", EXT_SUPPORTED_VERSIONS, 2) + b"\x03\x04" + ks = struct.pack(">HH", GROUP_X25519, len(server_pub)) + server_pub + ext += struct.pack(">HH", EXT_KEY_SHARE, len(ks)) + ks + + body = b"\x03\x03" + body += os.urandom(32) + body += bytes([len(ch["session_id"])]) + ch["session_id"] + body += struct.pack(">H", TLS_CHACHA20_POLY1305_SHA256) + body += b"\x00" + body += struct.pack(">H", len(ext)) + ext + return _handshake(HS_SERVER_HELLO, body) + + def _certificate(self) -> bytes: + entry = _u24(len(self.cert_der)) + self.cert_der + b"\x00\x00" + body = b"\x00" + _u24(len(entry)) + entry + return _handshake(HS_CERTIFICATE, body) + + def _certificate_verify(self, transcript_hash: bytes) -> bytes: + signed = b"\x20" * 64 + signed += b"TLS 1.3, server CertificateVerify" + signed += b"\x00" + signed += transcript_hash + sig = self.key.sign(signed, ec.ECDSA(hashes.SHA256())) + body = struct.pack(">H", SIG_ECDSA_SECP256R1_SHA256) + body += struct.pack(">H", len(sig)) + sig + return _handshake(HS_CERTIFICATE_VERIFY, body) + + def _finished(self, secret: bytes, transcript_hash: bytes) -> tuple[bytes, bool]: + finished_key = hkdf_expand_label(secret, b"finished", b"", HASH_LEN) + verify_data = hmac.new(finished_key, transcript_hash, hashlib.sha256).digest() + corrupted = False + if self.mode == "bad_finished": + # One bit, in the last byte. The message stays the right length and + # the right shape; only the MAC is wrong, so the client must reach + # the HMAC comparison to notice. + verify_data = verify_data[:31] + bytes([verify_data[31] ^ 0x01]) + corrupted = True + return _handshake(HS_FINISHED, verify_data), corrupted + + # -- the flight --------------------------------------------------------- + + def serve_one(self, sock: socket.socket, timeout: float) -> dict: + reader = RecordReader(sock) + + rec = reader.read_record(timeout) + if rec is None: + raise TlsFixtureError("client closed before sending ClientHello") + if rec[0] != CT_HANDSHAKE: + raise TlsFixtureError(f"expected handshake record, got type {rec[0]}") + ch_msg = rec[5:] + ch = parse_client_hello(ch_msg) + self.result["client_hello_seen"] = True + + server_priv = x25519.X25519PrivateKey.generate() + server_pub = server_priv.public_key().public_bytes_raw() + shared = server_priv.exchange( + x25519.X25519PublicKey.from_public_bytes(ch["key_share"]) + ) + + sh_msg = self._server_hello(ch, server_pub) + sock.sendall(_plaintext_record(CT_HANDSHAKE, sh_msg)) + + transcript = ch_msg + sh_msg + + early = _hkdf_extract(b"", b"\x00" * HASH_LEN) + derived = derive_secret(early, b"derived", _sha256(b"")) + handshake_secret = _hkdf_extract(derived, shared) + c_hs = derive_secret(handshake_secret, b"c hs traffic", _sha256(transcript)) + s_hs = derive_secret(handshake_secret, b"s hs traffic", _sha256(transcript)) + s_keys = TrafficKeys(s_hs) + c_keys = TrafficKeys(c_hs) + + def send_hs(msg: bytes) -> None: + # One handshake message per record: the C64 client's + # tls_recv_encrypted dispatches on tls_rec_buf[0] and handles + # exactly one message per decrypted record. + sock.sendall(s_keys.encrypt(msg + bytes([CT_HANDSHAKE]))) + + ee = _handshake(HS_ENCRYPTED_EXTENSIONS, b"\x00\x00") + send_hs(ee) + transcript += ee + + cert = self._certificate() + send_hs(cert) + transcript += cert + + cv = self._certificate_verify(_sha256(transcript)) + send_hs(cv) + transcript += cv + + fin, corrupted = self._finished(s_hs, _sha256(transcript)) + send_hs(fin) + self.result["finished_corrupted"] = corrupted + self.result["server_flight_sent"] = True + + # Fold the Finished we actually SENT. A client that accepts the + # corrupted Finished folds the same bytes, so its transcript still + # agrees with ours and the rest of the handshake would succeed. That + # is deliberate: it means a client with a broken check does not merely + # stall, it sails through to HTTP 200 — a fast, unambiguous failure + # signal instead of a test timeout. + transcript += fin + + master = _hkdf_extract( + derive_secret(handshake_secret, b"derived", _sha256(b"")), + b"\x00" * HASH_LEN, + ) + ap_transcript_hash = _sha256(transcript) + c_ap = derive_secret(master, b"c ap traffic", ap_transcript_hash) + s_ap = derive_secret(master, b"s ap traffic", ap_transcript_hash) + + expected_cf_key = hkdf_expand_label(c_hs, b"finished", b"", HASH_LEN) + expected_cf = hmac.new( + expected_cf_key, _sha256(transcript), hashlib.sha256 + ).digest() + + # --- What does the client do with our Finished? ------------------- + # This is the whole experiment. Whatever comes back next is recorded + # as server-side evidence; the client cannot fabricate it. + while True: + try: + rec = reader.read_record(timeout) + except (TimeoutError, socket.timeout, OSError) as exc: + self.result["client_accepted_finished"] = False + self.result["client_reaction"] = f"no response ({type(exc).__name__})" + return self.result + if rec is None: + self.result["client_accepted_finished"] = False + self.result["client_reaction"] = "closed connection" + return self.result + if rec[0] == CT_CHANGE_CIPHER_SPEC: + continue + if rec[0] == CT_ALERT: + self.result["client_accepted_finished"] = False + self.result["client_reaction"] = f"plaintext alert {rec[5:].hex()}" + return self.result + try: + ctype, pt = c_keys.decrypt(rec) + except Exception as exc: # noqa: BLE001 — fixture + self.result["client_accepted_finished"] = False + self.result["client_reaction"] = ( + f"undecryptable record ({type(exc).__name__})" + ) + return self.result + if ctype == CT_ALERT: + self.result["client_accepted_finished"] = False + self.result["client_reaction"] = f"encrypted alert {pt.hex()}" + return self.result + if ctype != CT_HANDSHAKE or not pt or pt[0] != HS_FINISHED: + self.result["client_accepted_finished"] = False + self.result["client_reaction"] = ( + f"unexpected record: inner type {ctype}, " + f"first byte {pt[0] if pt else None}" + ) + return self.result + self.result["client_accepted_finished"] = True + self.result["client_reaction"] = "sent its own Finished" + self.result["client_finished_valid"] = hmac.compare_digest( + pt[4:36], expected_cf + ) + break + + c_app = TrafficKeys(c_ap) + s_app = TrafficKeys(s_ap) + + req = b"" + while b"\r\n\r\n" not in req: + try: + rec = reader.read_record(timeout) + except (TimeoutError, socket.timeout, OSError): + break + if rec is None: + break + if rec[0] == CT_CHANGE_CIPHER_SPEC: + continue + ctype, pt = c_app.decrypt(rec) + if ctype == CT_APPLICATION_DATA: + req += pt + elif ctype == CT_ALERT: + self.result["client_alert"] = pt.hex() + break + self.result["request"] = req + + payload = self.body.encode() + response = ( + b"HTTP/1.1 200 OK\r\n" + b"Content-Type: text/plain\r\n" + b"Content-Length: " + str(len(payload)).encode() + b"\r\n" + b"Connection: close\r\n" + b"\r\n" + payload + ) + sock.sendall(s_app.encrypt(response + bytes([CT_APPLICATION_DATA]))) + self.result["response_sent"] = True + time.sleep(1.0) + return self.result + +def serve_one_connection(srv: socket.socket, cert_path: str, key_path: str, *, + mode: str, body: str, timeout: float, + result: dict) -> None: + """Accept exactly one connection and run the flight. Fills *result*.""" + conn = None + try: + srv.settimeout(timeout) + srv.listen(1) + result["listening"] = True + conn, addr = srv.accept() + result["client_addr"] = addr + server = EvilTls13Server(cert_path, key_path, mode=mode, body=body) + result.update(server.result) + result["client_addr"] = addr + result["listening"] = True + try: + server.serve_one(conn, timeout) + finally: + result.update(server.result) + result["client_addr"] = addr + result["listening"] = True + except Exception as exc: # noqa: BLE001 — fixture + result["error"] = f"{type(exc).__name__}: {exc}" + finally: + for s in (conn, srv): + try: + if s is not None: + s.close() + except Exception: + pass + + +# --------------------------------------------------------------------------- +# Self-test against Python's own ssl client +# --------------------------------------------------------------------------- + +def _selftest() -> int: + import ssl + + sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + from https_listener import _ensure_certs_p256 # noqa: PLC0415 + + cert_path, key_path = _ensure_certs_p256() + failures = 0 + + for mode, expect in (("good", "handshake completes"), + ("bad_finished", "client rejects")): + srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + srv.bind(("127.0.0.1", 0)) + port = srv.getsockname()[1] + + result: dict = {} + t = threading.Thread( + target=serve_one_connection, + args=(srv, cert_path, key_path), + kwargs=dict(mode=mode, body=DEFAULT_BODY, timeout=20.0, + result=result), + daemon=True, + ) + t.start() + time.sleep(0.2) + + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.load_verify_locations(cafile=cert_path) + ctx.minimum_version = ssl.TLSVersion.TLSv1_3 + + ok = False + detail = "" + try: + raw = socket.create_connection(("127.0.0.1", port), timeout=20.0) + with ctx.wrap_socket(raw, server_hostname="www.foo.bar") as tls: + tls.sendall(b"GET / HTTP/1.1\r\nHost: www.foo.bar\r\n\r\n") + data = tls.recv(4096) + if mode == "good": + ok = b"200 OK" in data and DEFAULT_BODY.encode() in data + detail = repr(data[:80]) + else: + detail = f"handshake COMPLETED — server never rejected: {data[:60]!r}" + except ssl.SSLError as exc: + detail = f"{type(exc).__name__}: {exc}" + if mode == "bad_finished": + ok = True + except Exception as exc: # noqa: BLE001 + detail = f"{type(exc).__name__}: {exc}" + + t.join(timeout=25.0) + + verdict = "PASS" if ok else "FAIL" + print(f" {verdict}: mode={mode:<13} expect {expect}") + print(f" client saw : {detail}") + print(f" server saw : {result}") + if not ok: + failures += 1 + + print() + if failures: + print(f" [-] evil_listener self-test: {failures} FAILED") + else: + print(" [+] evil_listener self-test: ALL PASSED") + return 1 if failures else 0 + + +if __name__ == "__main__": + if "--selftest" in sys.argv: + sys.exit(_selftest()) + print(__doc__) + print("Run with --selftest to validate against Python's ssl client.") diff --git a/tools/test_finished_verify.py b/tools/test_finished_verify.py new file mode 100755 index 0000000..1969aa8 --- /dev/null +++ b/tools/test_finished_verify.py @@ -0,0 +1,374 @@ +#!/usr/bin/env python3 +"""test_finished_verify.py - negative + positive coverage for tls_verify_finished. + +Why this exists +--------------- +``tls_verify_finished`` (src/tls_keyschedule.s) is the *only* thing standing +between the client and a forged server Finished: it recomputes the expected +``verify_data`` and constant-time-compares it with the 32 bytes the server sent +at ``tls_rec_buf+4``. On mismatch it returns C=1, which ``tls13.s`` turns into +a handshake abort:: + + src/tls13.s jsr tls_verify_finished + bcs @enc_error -> sec/rts out of tls_recv_encrypted + src/tls13.s jsr tls_recv_encrypted + bcs -> @error -> handshake aborted + +Before this test, *nothing in the repo exercised the mismatch path*. A +mutation audit confirmed it: inverting the mismatch branch (``sec`` -> ``clc`` +in ``tls_verify_finished``) let the full hardware end-to-end handshake still +reach HTTP 200 with the correct body, undetected, because every listener the +suite ever talks to sends a *correct* Finished. + +This test drives the routine directly over DMA with hand-built inputs, so it +can present a Finished the client must reject. It is deliberately narrow: it +tests one branch, but it tests it for real. + +Coverage +-------- +For each of two independent (server_hs_secret, transcript) vector sets: + + positive correct verify_data -> expect C=0 + flip_first_byte correct, one bit flipped in byte 0 -> expect C=1 + flip_last_byte correct, one bit flipped in byte 31 -> expect C=1 + all_zeros 32 x 0x00 -> expect C=1 + all_ones 32 x 0xFF -> expect C=1 + truncated first 31 correct bytes then 0x00 -> expect C=1 + rotated correct bytes rotated left by one -> expect C=1 + wrong_secret valid HMAC under a *different* secret -> expect C=1 + wrong_transcript valid HMAC over a *different* transcript -> expect C=1 + +The last two are the realistic attacks: an active attacker who cannot derive +the server handshake traffic secret, and one who tries to substitute a +different transcript. ``truncated`` and ``rotated`` specifically catch a +compare loop that stops early or is off by one. + +The positive case additionally asserts that the C64's *computed* +``tls_verify_data`` equals an independent Python computation, so a routine that +learned to always return C=0 without doing the HMAC cannot pass. + +Reference implementation +------------------------ +``hkdf_expand_label`` / HMAC-SHA256 are recomputed here in plain Python. That +reference is itself pinned to RFC 8448 by ``tools/test_hkdf.py`` and +``tools/test_keyschedule_steps.py``; this file reuses RFC 8448 Section 3's +server handshake traffic secret as vector set A so the inputs are not +self-invented. + +Usage: + python3 tools/test_finished_verify.py [--verbose] + +Env: + C64_SKIP_BUILD=1 reuse the already-built PRG + +Requires: Python 3.10+, c64_test_harness, VICE x64sc +""" + +from __future__ import annotations + +import hashlib +import hmac +import os +import struct +import subprocess +import sys + +from c64_test_harness import ( + Labels, + ViceInstanceManager, + read_bytes, + write_bytes, + jsr, + wait_for_text, +) + +PROJECT_ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..") +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from _vice_helpers import default_vice_config # noqa: E402 + +PRG_PATH = os.path.join(PROJECT_ROOT, "build", "c64-https.prg") +LABELS_PATH = os.path.join(PROJECT_ROOT, "build", "labels.txt") + +VERBOSE = False + +REQUIRED_LABELS = [ + "tls_verify_finished", + "tls_verify_data", + "tls_s_hs_secret", + "tls_transcript", + "tls_rec_buf", +] + +# Cassette buffer. $033C-$03FB is free once BASIC has booted. The harness's +# own jsr() trampoline lives at $0334 (5 bytes) and run_subroutine's U64 +# trampoline at $0360 (14 bytes) with flags at $03F0/$03F1 — $0340 and $034C +# collide with none of them. +CARRY_STUB_ADDR = 0x0340 +CARRY_RESULT_ADDR = 0x034C + + +# --------------------------------------------------------------------------- +# Python reference (see module docstring for provenance) +# --------------------------------------------------------------------------- + +def hkdf_expand_label(secret: bytes, label: bytes, context: bytes, + length: int) -> bytes: + """TLS 1.3 HKDF-Expand-Label (RFC 8446 Section 7.1). L <= 32 only.""" + assert length <= 32 + info = struct.pack(">H", length) + info += bytes([6 + len(label)]) + b"tls13 " + label + info += bytes([len(context)]) + context + return hmac.new(secret, info + b"\x01", hashlib.sha256).digest()[:length] + + +def finished_verify_data(traffic_secret: bytes, transcript: bytes) -> bytes: + """RFC 8446 Section 4.4.4 verify_data.""" + finished_key = hkdf_expand_label(traffic_secret, b"finished", b"", 32) + return hmac.new(finished_key, transcript, hashlib.sha256).digest() + + +# --------------------------------------------------------------------------- +# Vectors +# --------------------------------------------------------------------------- + +# RFC 8448 Section 3 server handshake traffic secret (same value the existing +# key-schedule test pins the C64 against). +SECRET_A = bytes.fromhex( + "b67b7d690cc16c4e75e54213cb2d37b4" + "e9c912bcded9105d42befd59d391ad38" +) +# An arbitrary but fixed transcript hash. Any 32 bytes is a legal input here; +# the HMAC is defined over whatever the running hash produced. +TRANSCRIPT_A = hashlib.sha256(b"c64-https lane B transcript A").digest() + +# A second, independent vector set, so a routine that happens to be correct +# for one input pair cannot coast. +SECRET_B = hashlib.sha256(b"c64-https lane B secret B").digest() +TRANSCRIPT_B = hashlib.sha256(b"c64-https lane B transcript B").digest() + +# Used only to build "valid HMAC, wrong key/context" forgeries. +DECOY_SECRET = hashlib.sha256(b"c64-https lane B decoy secret").digest() +DECOY_TRANSCRIPT = hashlib.sha256(b"c64-https lane B decoy transcript").digest() + +VECTOR_SETS = [ + ("A (RFC 8448 s_hs_traffic)", SECRET_A, TRANSCRIPT_A), + ("B (independent)", SECRET_B, TRANSCRIPT_B), +] + + +def build_cases(secret: bytes, transcript: bytes): + """Return [(name, received_verify_data, expect_carry), ...].""" + good = finished_verify_data(secret, transcript) + + flip_first = bytes([good[0] ^ 0x01]) + good[1:] + flip_last = good[:31] + bytes([good[31] ^ 0x80]) + truncated = good[:31] + b"\x00" + rotated = good[1:] + good[:1] + wrong_secret = finished_verify_data(DECOY_SECRET, transcript) + wrong_transcript = finished_verify_data(secret, DECOY_TRANSCRIPT) + + cases = [ + ("positive", good, 0), + ("flip_first_byte", flip_first, 1), + ("flip_last_byte", flip_last, 1), + ("all_zeros", b"\x00" * 32, 1), + ("all_ones", b"\xff" * 32, 1), + ("truncated", truncated, 1), + ("rotated", rotated, 1), + ("wrong_secret", wrong_secret, 1), + ("wrong_transcript", wrong_transcript, 1), + ] + + # Sanity: every negative vector must genuinely differ from the correct one, + # otherwise the "case" is not a negative case at all. Guards against a + # degenerate vector (e.g. rotated == good for an all-same-byte digest). + for name, vd, expect in cases: + assert len(vd) == 32, f"{name}: verify_data must be 32 bytes" + if expect == 1: + assert vd != good, f"{name}: negative vector is not actually wrong" + else: + assert vd == good, f"{name}: positive vector is not the correct value" + + return cases, good + + +# --------------------------------------------------------------------------- +# C64 plumbing +# --------------------------------------------------------------------------- + +def install_carry_stub(transport, target_addr: int) -> None: + """Install a stub that calls *target_addr* and latches the carry flag. + + JSR target 20 lo hi + LDA #$00 A9 00 + ROL A 2A ; carry -> bit 0 + STA result 8D lo hi + RTS 60 + + Reading the P register back over the monitor is unreliable across + backends; latching the flag into RAM from 6502 code is not. The stub is + written once and reused for every case. + """ + lo, hi = target_addr & 0xFF, (target_addr >> 8) & 0xFF + rlo, rhi = CARRY_RESULT_ADDR & 0xFF, (CARRY_RESULT_ADDR >> 8) & 0xFF + stub = bytes([0x20, lo, hi, 0xA9, 0x00, 0x2A, 0x8D, rlo, rhi, 0x60]) + write_bytes(transport, CARRY_STUB_ADDR, stub) + readback = read_bytes(transport, CARRY_STUB_ADDR, len(stub)) + if readback != stub: + raise RuntimeError( + f"carry stub readback mismatch at ${CARRY_STUB_ADDR:04X}: " + f"wrote {stub.hex()}, read {readback.hex()}" + ) + + +def call_verify_finished(transport, labels, secret: bytes, transcript: bytes, + received: bytes) -> tuple[int, bytes]: + """Set up inputs, run tls_verify_finished, return (carry, computed_vd).""" + write_bytes(transport, labels["tls_s_hs_secret"], secret) + write_bytes(transport, labels["tls_transcript"], transcript) + write_bytes(transport, labels["tls_rec_buf"] + 4, received) + + # Poison the output buffer and the carry latch so a routine that never + # runs cannot be mistaken for one that ran and agreed with us. + write_bytes(transport, labels["tls_verify_data"], b"\xa5" * 32) + write_bytes(transport, CARRY_RESULT_ADDR, b"\xa5") + + jsr(transport, CARRY_STUB_ADDR, timeout=60.0) + + carry = read_bytes(transport, CARRY_RESULT_ADDR, 1)[0] + if carry not in (0, 1): + raise RuntimeError( + f"carry latch never written (read ${carry:02X}) — the stub did " + f"not complete; treat this run as inconclusive, not a pass" + ) + computed = read_bytes(transport, labels["tls_verify_data"], 32) + return carry, computed + + +# --------------------------------------------------------------------------- +# Test driver +# --------------------------------------------------------------------------- + +def run_tests(transport, labels) -> tuple[int, int]: + passed = failed = 0 + + install_carry_stub(transport, labels["tls_verify_finished"]) + + for set_name, secret, transcript in VECTOR_SETS: + print(f"\n--- Vector set {set_name} ---") + cases, good = build_cases(secret, transcript) + + for name, received, expect_carry in cases: + carry, computed = call_verify_finished( + transport, labels, secret, transcript, received + ) + + ok = carry == expect_carry + detail = "" + + # The positive case also proves the routine actually computed the + # HMAC rather than short-circuiting to "accept". + if expect_carry == 0 and ok: + if computed != good: + ok = False + detail = ( + f"\n computed verify_data mismatch" + f"\n expected {good.hex()}" + f"\n got {computed.hex()}" + ) + + verdict = "PASS" if ok else "FAIL" + want = "C=0 accept" if expect_carry == 0 else "C=1 reject" + got = "C=0 accept" if carry == 0 else "C=1 reject" + print(f" {verdict}: {name:<17} want {want}, got {got}{detail}") + if VERBOSE: + print(f" received {received.hex()}") + print(f" computed {computed.hex()}") + + if ok: + passed += 1 + else: + failed += 1 + + return passed, failed + + +def main() -> int: + global VERBOSE + os.chdir(PROJECT_ROOT) + + if "--verbose" in sys.argv: + VERBOSE = True + + if os.environ.get("C64_SKIP_BUILD"): + print("\n=== Building (skipped: C64_SKIP_BUILD set) ===") + else: + print("\n=== Building ===") + subprocess.run(["make", "clean"], capture_output=True) + result = subprocess.run(["make"], capture_output=True, text=True) + if result.returncode != 0: + print(f"Build failed:\n{result.stderr}") + return 1 + print(" Build OK") + + if not os.path.exists(PRG_PATH): + print(f"FATAL: {PRG_PATH} not found") + return 1 + + labels = Labels.from_file(LABELS_PATH) + missing = [n for n in REQUIRED_LABELS if labels.address(n) is None] + if missing: + # A missing label means the routine under test moved or was renamed. + # That is a failure, never a skip — see audit finding F3. + print(f"FATAL: required label(s) not found: {', '.join(missing)}") + return 1 + + print("\n=== Labels ===") + for name in REQUIRED_LABELS: + print(f" {name:<22} = ${labels[name]:04X}") + + print("\n=== Starting VICE ===") + config = default_vice_config( + prg_path=PRG_PATH, + warp=True, + ntsc=True, + sound=False, + ) + + with ViceInstanceManager(config=config) as mgr: + inst = mgr.acquire() + transport = inst.transport + print(f" VICE PID={inst.pid}, port={inst.port}") + + print(" Waiting for main menu...") + grid = wait_for_text(transport, "Q=QUIT", timeout=120.0, verbose=False) + if grid is None: + print("FATAL: Main menu did not appear") + mgr.release(inst) + return 1 + print(" Main menu ready") + + print("\n=== tls_verify_finished ===") + try: + passed, failed = run_tests(transport, labels) + finally: + mgr.release(inst) + + total = passed + failed + print("\n" + "=" * 60) + print("RESULTS") + print("=" * 60) + print(f" Passed: {passed}/{total}") + print(f" Failed: {failed}/{total}") + if failed == 0: + print(f"\n [+] Finished verify: ALL {total} TESTS PASSED") + else: + print(f"\n [-] Finished verify: {failed} TEST(S) FAILED") + print("=" * 60) + + return 0 if failed == 0 else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/uci/test_https_bad_finished.py b/tools/uci/test_https_bad_finished.py new file mode 100755 index 0000000..004296c --- /dev/null +++ b/tools/uci/test_https_bad_finished.py @@ -0,0 +1,641 @@ +#!/usr/bin/env python3 +"""test_https_bad_finished.py — the client must refuse a forged server Finished. + +Audit finding F2: the client verifies the server's Finished HMAC and aborts on +mismatch (``tls_verify_finished`` in ``src/tls_keyschedule.s``, ``bcs +@enc_error`` in ``src/tls13.s``), but nothing in the repo ever exercised the +abort. Every listener the suite talks to sends a *correct* Finished, so the +mismatch branch was dead weight as far as the tests were concerned — confirmed +by mutation: inverting it (``sec`` -> ``clc``) left the full hardware e2e +reaching HTTP 200 with the correct body. + +This test closes that hole end-to-end on real hardware. It points the C64 at +``tools/https_e2e/evil_listener.py``, a hand-rolled TLS 1.3 server that emits a +completely valid flight — real X25519 ECDHE, real key schedule, real +ChaCha20-Poly1305 records, real P-256 CertificateVerify — with exactly one bit +flipped in the server Finished ``verify_data`` before encryption. The AEAD tag +is correct, so the client cannot bail out at the record layer; it has to reach +the HMAC comparison to notice anything is wrong. + +Why not just corrupt the ciphertext: that breaks the Poly1305 tag, the client +rejects at ``aead_decrypt``, and the Finished comparison never runs. That would +pass this test while proving nothing about F2. + +Two modes, selected by ``FINISHED_MODE``: + + ``bad`` (default) the server sends the corrupted Finished. PASS requires the + client to abort *at Finished*. + ``good`` the identical server sends a correct Finished. PASS requires a + complete handshake and HTTP 200. This is the control: it proves the + hand-rolled server is a working TLS 1.3 server, so an abort in + ``bad`` mode is attributable to the one flipped bit and not to a + fixture that simply cannot talk to the client. + +Run ``good`` before trusting a ``bad`` result. + +Oracle +------ +Both directions are asserted, from both sides of the wire. + +C64 side — ``src/tls13.s:@error`` stashes the state it died in: + + bad : tls_state == $FF (ERROR) and tls_last_state == 6 (FINISHED) + and http_status != 200 + good : tls_state != $FF (ERROR) and http_status == 200 and the body + +(Not ``tls_state == CONNECTED`` for the good run: ``http_get``'s success path +calls ``tls_close``, which puts the state back to IDLE.) + +``tls_last_state`` is what makes this precise rather than merely negative: it +distinguishes "aborted at Finished" from "aborted earlier at Certificate (4) or +CertificateVerify (5)". A test that only checked "handshake failed" would pass +for a fixture that produced a broken certificate. + +Server side — evidence the client cannot fabricate, recorded in +``server_result.json``: + + bad : client_accepted_finished is False (the client never sent its own + Finished), and finished_corrupted is True + good : client_accepted_finished is True, client_finished_valid is True, + response_sent is True + +Note the server folds the Finished it actually sent into its own transcript, so +a client that wrongly *accepts* the corrupted Finished stays in lockstep and +sails on to HTTP 200. A broken client therefore fails fast and unambiguously +instead of hanging until the sentinel timeout. + +Environment +----------- + U64_HOST U64E / C64U address (default 192.168.1.81) + FINISHED_MODE bad (default) | good + TURBO_MHZ C64 CPU MHz (default 48); timeouts auto-scale + HTTPS_PORT listener port (default 4433) + SENTINEL_POLL_TIMEOUT / ACCEPT_TIMEOUT per-test overrides, seconds + C64_INIT_WAIT boot/auto-init wait before triggering (default 22 s, + scaled); comb-profile builds need 90+ + UCI_DEBUG_DIR artifact base dir (default /tmp/uci_bad_finished) + +Exit codes: 0 pass, 1 fail, 2 setup error, 3 device wedged. +""" +from __future__ import annotations + +import datetime +import json +import os +import socket +import sys +import threading +import time +from pathlib import Path + +from c64_test_harness.backends.device_lock import DeviceLock, DeviceLockTimeout +from c64_test_harness.backends.ultimate64 import Ultimate64Transport +from c64_test_harness.backends.ultimate64_client import Ultimate64Client +from c64_test_harness.backends.ultimate64_helpers import ( + set_turbo_mhz, + runner_health_check, + Ultimate64RunnerStuckError, + CAT_U64_SPECIFIC, + cpu_speed_enum, +) +from c64_test_harness.uci_network import enable_uci, disable_uci +from c64_test_harness.keyboard import send_text +from c64_test_harness.labels import Labels + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from _memory_policy import ( # noqa: E402 + build_policy_and_arbiter_with_overlay_carveout, +) + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT / "tools" / "https_e2e")) +from evil_listener import ( # noqa: E402 + DEFAULT_BODY, + serve_one_connection, +) +from https_listener import _ensure_certs_p256 # noqa: E402 + +HOST = os.environ.get("U64_HOST", "192.168.1.81") +PRG_PATH = REPO_ROOT / "build" / "c64-https.prg" +LABELS_PATH = REPO_ROOT / "build" / "labels.txt" + +MODE_ENV = os.environ.get("FINISHED_MODE", "bad").lower() +if MODE_ENV not in ("bad", "good"): + print(f"ERROR: FINISHED_MODE must be 'bad' or 'good', got {MODE_ENV!r}", + file=sys.stderr) + sys.exit(2) +SERVER_MODE = "bad_finished" if MODE_ENV == "bad" else "good" + +TURBO_MHZ = int(os.environ.get("TURBO_MHZ", "48")) +_TIMEOUT_SCALE = max(1.0, 48.0 / float(TURBO_MHZ)) +SENTINEL_POLL_TIMEOUT = float( + os.environ.get("SENTINEL_POLL_TIMEOUT", str(600.0 * _TIMEOUT_SCALE)) +) +ACCEPT_TIMEOUT = float( + os.environ.get("ACCEPT_TIMEOUT", str(600.0 * _TIMEOUT_SCALE)) +) +HTTPS_PORT = int(os.environ.get("HTTPS_PORT", "4433")) +ARTIFACT_BASE = Path(os.environ.get("UCI_DEBUG_DIR", "/tmp/uci_bad_finished")) + +SENTINEL_VALUE = 0xAA + +# TLS_STATE_* from src/constants.inc +TLS_STATE_CERTIFICATE = 4 +TLS_STATE_CERT_VERIFY = 5 +TLS_STATE_FINISHED = 6 +TLS_STATE_CONNECTED = 7 +TLS_STATE_ERROR = 0xFF + +_STATE_NAMES = { + 0: "IDLE", 1: "CLIENT_HELLO", 2: "SERVER_HELLO", 3: "ENCRYPTED_EXT", + 4: "CERTIFICATE", 5: "CERT_VERIFY", 6: "FINISHED", 7: "CONNECTED", + 0xFF: "ERROR", +} + + +def _state_name(v: int) -> str: + return f"{_STATE_NAMES.get(v, '?')} (${v:02X})" + + +# Arbiter-assigned; see the long note in test_https_local.py about why these +# must never be hardcoded. +ROUTINE_ADDR = HOST_STR_ADDR = PATH_STR_ADDR = -1 +SENTINEL_ADDR = PROGRESS_ADDR = CARRY_FLAG_ADDR = -1 + + +def _detect_local_ip(target: str) -> str: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + s.connect((target, 80)) + return s.getsockname()[0] + finally: + s.close() + + +def _try_bind(bind_ip: str, port: int) -> socket.socket | None: + srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + srv.bind((bind_ip, port)) + except OSError: + srv.close() + return None + return srv + + +def _build_http_routine(labels: dict[str, int], port: int) -> tuple[bytes, int]: + """6502 stub: set up http_get's inputs, call it, latch carry, signal done. + + Mirrors tools/uci/test_https_local.py's routine — same real code path + (``http_get`` -> ``tls_connect``), so the only thing this test changes + relative to the passing e2e is what the server puts on the wire. + """ + code = bytearray() + + def emit(*bs: int) -> None: + code.extend(bs) + + def emit_lda_imm(v: int) -> None: + emit(0xA9, v & 0xFF) + + def emit_sta_abs(a: int) -> None: + emit(0x8D, a & 0xFF, (a >> 8) & 0xFF) + + def emit_lda_abs(a: int) -> None: + emit(0xAD, a & 0xFF, (a >> 8) & 0xFF) + + def emit_jsr(a: int) -> None: + emit(0x20, a & 0xFF, (a >> 8) & 0xFF) + + def emit_progress(step: int) -> None: + emit_lda_imm(step) + emit_sta_abs(PROGRESS_ADDR) + + # Bank BASIC ROM out so $A000-$BFFF reads as RAM (crypto/TLS BSS lives + # there; without this the later DMA reads would return ROM bytes). + emit_lda_abs(0x0001) + emit(0x29, 0xFE) + emit_sta_abs(0x0001) + + emit_lda_imm(0x00) + emit_sta_abs(SENTINEL_ADDR) + emit_sta_abs(PROGRESS_ADDR) + emit_sta_abs(CARRY_FLAG_ADDR) + + emit_progress(0x01) + emit_jsr(labels["net_init"]) + + emit_lda_imm(0x00) + emit_sta_abs(labels["tcp_recv_head"]) + emit_sta_abs(labels["tcp_recv_head"] + 1) + emit_sta_abs(labels["tcp_recv_tail"]) + emit_sta_abs(labels["tcp_recv_tail"] + 1) + + emit_progress(0x02) + + emit_lda_imm(HOST_STR_ADDR & 0xFF) + emit_sta_abs(labels["http_host_ptr"]) + emit_lda_imm((HOST_STR_ADDR >> 8) & 0xFF) + emit_sta_abs(labels["http_host_ptr"] + 1) + + host_len_patch_offset = len(code) + 1 + emit_lda_imm(0x00) + emit_sta_abs(labels["http_host_len"]) + + emit_lda_imm(PATH_STR_ADDR & 0xFF) + emit_sta_abs(labels["http_path_ptr"]) + emit_lda_imm((PATH_STR_ADDR >> 8) & 0xFF) + emit_sta_abs(labels["http_path_ptr"] + 1) + emit_lda_imm(1) + emit_sta_abs(labels["http_path_len"]) + + emit_lda_imm(port & 0xFF) + emit_sta_abs(labels["http_port"]) + emit_lda_imm((port >> 8) & 0xFF) + emit_sta_abs(labels["http_port"] + 1) + + emit_progress(0x03) + emit_jsr(labels["http_get"]) + + # Latch the carry into RAM rather than reading the CPU status register + # over the wire. PHP/PLA puts the whole P register in A; bit 0 is C. + emit(0x08) # PHP + emit(0x68) # PLA + emit_sta_abs(CARRY_FLAG_ADDR) + + emit_progress(0x04) + emit_lda_imm(SENTINEL_VALUE) + emit_sta_abs(SENTINEL_ADDR) + emit_progress(0x05) + + park = ROUTINE_ADDR + len(code) + emit(0x4C, park & 0xFF, (park >> 8) & 0xFF) + return bytes(code), host_len_patch_offset + + +def _decode_screen_ram(data: bytes) -> str: + """Screen codes -> ASCII, 40 columns.""" + out = [] + for row in range(min(25, len(data) // 40)): + line = [] + for col in range(40): + c = data[row * 40 + col] + if c == 0x20 or c == 0x00: + line.append(" ") + elif 0x01 <= c <= 0x1A: + line.append(chr(ord("A") + c - 1)) + elif 0x30 <= c <= 0x39: + line.append(chr(c)) + elif c == 0x2E: + line.append(".") + elif c == 0x2D: + line.append("-") + elif c == 0x3A: + line.append(":") + elif c == 0x2F: + line.append("/") + else: + line.append(".") + out.append("".join(line).rstrip()) + return "\n".join(out) + + +def _read_c64_state(transport, labels) -> dict: + def rd(name: str, n: int = 1) -> bytes: + return bytes(transport.read_memory(labels[name], n)) + + resp_len_raw = rd("http_resp_len", 2) + resp_len = resp_len_raw[0] | (resp_len_raw[1] << 8) + status_raw = rd("http_status", 2) + read_len = min(resp_len, 200) if resp_len > 0 else 64 + return { + "tls_state": rd("tls_state")[0], + "tls_last_state": rd("tls_last_state")[0], + "http_status": status_raw[0] | (status_raw[1] << 8), + "http_resp_len": resp_len, + "http_resp_buf": bytes( + transport.read_memory(labels["http_resp_buf"], read_len) + ), + "net_last_error": ( + rd("net_last_error")[0] if "net_last_error" in labels else None + ), + } + + +def _write_artifacts(run_dir: Path, *, server_result: dict, c64: dict, + screen_text: str, mode: str, outcome: str, + reasons: list[str]) -> None: + run_dir.mkdir(parents=True, exist_ok=True) + serialisable = dict(server_result) + for k, v in list(serialisable.items()): + if isinstance(v, (bytes, bytearray)): + serialisable[k] = v.decode("latin-1") + elif isinstance(v, tuple): + serialisable[k] = list(v) + (run_dir / "server_result.json").write_text( + json.dumps(serialisable, indent=2, default=str) + ) + c64_json = dict(c64) + if isinstance(c64_json.get("http_resp_buf"), (bytes, bytearray)): + c64_json["http_resp_buf"] = c64_json["http_resp_buf"].decode( + "ascii", errors="replace" + ) + (run_dir / "c64_state.json").write_text(json.dumps(c64_json, indent=2)) + (run_dir / "screen.txt").write_text(screen_text) + (run_dir / "run_info.txt").write_text( + f"mode : {mode}\n" + f"outcome : {outcome}\n" + f"host : {HOST}\n" + f"turbo_mhz : {TURBO_MHZ}\n" + f"reasons :\n" + "".join(f" - {r}\n" for r in reasons) + ) + + +def _evaluate(mode: str, server_result: dict, c64: dict, + screen_text: str) -> tuple[bool, list[str]]: + """Return (passed, reasons). Every criterion is reported, pass or fail.""" + reasons: list[str] = [] + ok = True + + def check(cond: bool, msg: str) -> None: + nonlocal ok + reasons.append(("OK " if cond else "FAIL ") + msg) + if not cond: + ok = False + + err = server_result.get("error") + check(not err, f"server reported no error (error={err!r})") + check(bool(server_result.get("client_hello_seen")), + "server received a ClientHello") + check(bool(server_result.get("server_flight_sent")), + "server sent its full handshake flight") + + body = c64.get("http_resp_buf", b"").decode("ascii", errors="replace") + + if mode == "bad": + check(bool(server_result.get("finished_corrupted")), + "server actually corrupted the Finished verify_data") + check(server_result.get("client_accepted_finished") is False, + "client did NOT send its own Finished " + f"(reaction: {server_result.get('client_reaction')!r})") + check(c64["tls_state"] == TLS_STATE_ERROR, + f"tls_state is ERROR (got {_state_name(c64['tls_state'])})") + check(c64["tls_last_state"] == TLS_STATE_FINISHED, + "abort happened AT Finished, not earlier " + f"(tls_last_state = {_state_name(c64['tls_last_state'])})") + check(c64["http_status"] != 200, + f"no HTTP 200 was parsed (http_status={c64['http_status']})") + check(DEFAULT_BODY not in body, + "response body was not received") + check(DEFAULT_BODY.upper() not in screen_text.upper(), + "response body did not reach the screen either") + else: + check(server_result.get("client_accepted_finished") is True, + "client sent its own Finished") + check(server_result.get("client_finished_valid") is True, + "client Finished verified against the server's expectation") + check(bool(server_result.get("response_sent")), + "server sent the HTTP response") + req = server_result.get("request") or b"" + if isinstance(req, str): + req = req.encode("latin-1") + check(req.startswith(b"GET "), + f"server decrypted a GET request ({req[:40]!r})") + # NOT `== CONNECTED`: on the success path http_get calls tls_close, + # which sets tls_state back to IDLE (src/tls13.s:tls_close). CONNECTED + # is only observable mid-flight. What matters here is that the + # handshake never took the error path — measured on hardware, where + # the naive CONNECTED assertion failed a genuinely passing run. + check(c64["tls_state"] != TLS_STATE_ERROR, + f"tls_state is not ERROR (got {_state_name(c64['tls_state'])})") + check(c64["http_status"] == 200, + f"http_status is 200 (got {c64['http_status']})") + check(DEFAULT_BODY in body, + f"http_resp_buf holds the expected body ({body[:40]!r})") + + return ok, reasons + + +def main() -> int: + if not PRG_PATH.is_file() or not LABELS_PATH.is_file(): + print(f"ERROR: build artifacts missing; run `make BACKEND=uci` first", + file=sys.stderr) + return 2 + + labels = dict(Labels.from_file(LABELS_PATH)) + required = [ + "http_get", "http_host_ptr", "http_host_len", + "http_path_ptr", "http_path_len", "http_port", + "net_init", "net_initialized", + "tcp_recv_head", "tcp_recv_tail", + "http_resp_buf", "http_resp_len", "http_status", + "tls_state", "tls_last_state", + ] + missing = [n for n in required if n not in labels] + if missing: + # A missing label is a broken test, not a skippable one (finding F3). + print(f"ERROR: missing labels: {missing}", file=sys.stderr) + return 2 + + print(f"=== HTTPS bad-Finished e2e ({MODE_ENV.upper()} mode) ===") + print(f"Device : {HOST} @ {TURBO_MHZ} MHz") + print(f"Server mode : {SERVER_MODE}") + print(f"PRG : {PRG_PATH}") + + global ROUTINE_ADDR, HOST_STR_ADDR, PATH_STR_ADDR + global SENTINEL_ADDR, PROGRESS_ADDR, CARRY_FLAG_ADDR + memory_policy, arbiter = build_policy_and_arbiter_with_overlay_carveout( + LABELS_PATH, PRG_PATH, + ) + ROUTINE_ADDR = arbiter.alloc(256, name="trampoline") + HOST_STR_ADDR = arbiter.alloc(64, name="host_str") + PATH_STR_ADDR = arbiter.alloc(64, name="path_str") + SENTINEL_ADDR = arbiter.alloc(1, name="sentinel") + PROGRESS_ADDR = arbiter.alloc(1, name="progress") + CARRY_FLAG_ADDR = arbiter.alloc(1, name="carry_flag") + for base, last, note in arbiter.allocations: + print(f" ${base:04X}-${last:04X} {note}") + + cert_path, key_path = _ensure_certs_p256() + test_host_ip = _detect_local_ip(HOST) + srv = _try_bind(test_host_ip, HTTPS_PORT) + if srv is None: + print(f"ERROR: could not bind {test_host_ip}:{HTTPS_PORT}", + file=sys.stderr) + return 2 + print(f"Listener : {test_host_ip}:{HTTPS_PORT} (cert {cert_path})") + + server_result: dict = {} + server_thread = threading.Thread( + target=serve_one_connection, + args=(srv, cert_path, key_path), + kwargs=dict(mode=SERVER_MODE, body=DEFAULT_BODY, + timeout=ACCEPT_TIMEOUT, result=server_result), + daemon=True, + ) + server_thread.start() + for _ in range(100): + if server_result.get("listening"): + break + time.sleep(0.05) + else: + print("ERROR: listener failed to come up", file=sys.stderr) + return 2 + + routine_raw, host_len_patch = _build_http_routine(labels, HTTPS_PORT) + routine = bytearray(routine_raw) + host_bytes = test_host_ip.encode("ascii") + routine[host_len_patch] = len(host_bytes) + routine = bytes(routine) + + prg = PRG_PATH.read_bytes() + run_dir = ARTIFACT_BASE / datetime.datetime.now().strftime("%Y%m%d_%H%M%S") + + lock = DeviceLock(HOST) + try: + lock.acquire_or_raise(timeout=300.0) + except DeviceLockTimeout as exc: + print(f"[fatal] DeviceLock({HOST}): {exc}", file=sys.stderr) + return 2 + print(f"Acquired DeviceLock({HOST})") + + client: Ultimate64Client | None = None + uci_enabled = False + try: + client = Ultimate64Client(host=HOST, timeout=15.0) + transport = Ultimate64Transport(host=HOST, timeout=15.0, client=client) + transport.memory_policy = memory_policy + + print("Enabling UCI...") + enable_uci(client) + uci_enabled = True + + try: + runner_health_check(client) + except Ultimate64RunnerStuckError as exc: + print(f"[fatal] runner wedged at {HOST}: {exc}", file=sys.stderr) + return 3 + + # Set turbo BEFORE boot, and skip a redundant write — the config write + # itself is what glitches the UCI bridge on a C64U (see the long note + # in test_https_local.py and the c64u_starlight_device memory). + try: + cat = client.get_config_category(CAT_U64_SPECIFIC) + inner = cat.get(CAT_U64_SPECIFIC, cat) + cur_speed, cur_turbo = inner.get("CPU Speed"), inner.get("Turbo Control") + except Exception as exc: + print(f" (turbo probe failed: {exc}; writing anyway)") + cur_speed = cur_turbo = None + if str(cur_speed) == str(cpu_speed_enum(TURBO_MHZ)) and cur_turbo == "Manual": + print(f"Turbo already {TURBO_MHZ} MHz — skipping config write") + else: + print(f"Setting turbo {cur_turbo}/{cur_speed} -> {TURBO_MHZ} MHz") + set_turbo_mhz(client, TURBO_MHZ) + time.sleep(float(os.environ.get("TURBO_SETTLE", "3.0"))) + + print("Resetting machine...") + client.reset() + time.sleep(2.5) + + print("run_prg(PRG)...") + client.run_prg(prg) + time.sleep(float(os.environ.get("C64_INIT_WAIT", "22")) * _TIMEOUT_SCALE) + + init_flag = transport.read_memory(labels["net_initialized"], 1)[0] + print(f"net_initialized = ${init_flag:02X}") + if init_flag == 0: + print("WARNING: net_initialized is 0 — auto-init may have failed") + + print("Sending 'Q' to exit main_loop...") + send_text(transport, "q\r") + time.sleep(2.0 * _TIMEOUT_SCALE) + + for i in range(0, len(routine), 64): + transport.write_memory(ROUTINE_ADDR + i, routine[i:i + 64]) + transport.write_memory(HOST_STR_ADDR, (host_bytes + b"\x00").ljust(32, b"\x00")) + transport.write_memory(PATH_STR_ADDR, b"/\x00".ljust(8, b"\x00")) + transport.write_memory(SENTINEL_ADDR, bytes(16)) + + print(f"Triggering: sys{ROUTINE_ADDR}") + send_text(transport, f"sys{ROUTINE_ADDR}\r") + + deadline = time.time() + SENTINEL_POLL_TIMEOUT + start = time.time() + last_progress = -1 + completed = False + while time.time() < deadline: + time.sleep(0.5) + blob = transport.read_memory(SENTINEL_ADDR, 2) + if blob[1] != last_progress: + print(f" [{time.time() - start:6.1f}s] progress=0x{blob[1]:02X}") + last_progress = blob[1] + if blob[0] == SENTINEL_VALUE: + completed = True + print(f" sentinel set after {time.time() - start:.1f}s") + break + + server_thread.join(timeout=10.0) + + c64 = _read_c64_state(transport, labels) + screen_text = _decode_screen_ram( + bytes(transport.read_memory(0x0400, 1000)) + ) + + print("\n--- C64 state ---") + print(f" tls_state = {_state_name(c64['tls_state'])}") + print(f" tls_last_state = {_state_name(c64['tls_last_state'])}") + print(f" http_status = {c64['http_status']}") + print(f" http_resp_len = {c64['http_resp_len']}") + print(f" http_resp_buf = " + f"{c64['http_resp_buf'][:48].decode('ascii', 'replace')!r}") + print("\n--- server saw ---") + for k, v in server_result.items(): + print(f" {k:26s} = {v!r}") + print("\n--- screen ---") + print(screen_text) + + if not completed: + # The 6502 stub never signalled completion, so http_get is still + # running or wedged. We cannot say what the client decided — + # inconclusive is a failure, never a pass. + reasons = [f"FAIL routine did not complete within " + f"{SENTINEL_POLL_TIMEOUT:.0f}s " + f"(progress=0x{last_progress:02X}) — inconclusive"] + passed = False + else: + passed, reasons = _evaluate(MODE_ENV, server_result, c64, screen_text) + + outcome = "PASS" if passed else "FAIL" + print(f"\n--- criteria ({MODE_ENV} mode) ---") + for r in reasons: + print(f" {r}") + _write_artifacts(run_dir, server_result=server_result, c64=c64, + screen_text=screen_text, mode=MODE_ENV, + outcome=outcome, reasons=reasons) + print(f"\nArtifacts: {run_dir}") + print(f"\n{outcome}: " + + ("client rejected the forged server Finished" + if passed and MODE_ENV == "bad" else + "handshake completed against the control listener" + if passed else + "see failed criteria above")) + return 0 if passed else 1 + + finally: + if uci_enabled and client is not None: + try: + disable_uci(client) + except Exception as exc: + print(f"WARNING: disable_uci failed: {exc}") + try: + lock.release() + except Exception: + pass + try: + srv.close() + except Exception: + pass + + +if __name__ == "__main__": + sys.exit(main())