#!/usr/bin/env python3 """ Live End-to-End Test Suite for Portal TV (10.0.0.10:5654) and portalkit-cli. Test Scenarios: a) Status Verification: - Runs portalkit-cli status 10.0.0.10:5654. - Asserts service is online, leaf cert SHA-256 is present and matches genuine cert. b) Direct Channel Binding Defense Verification: - Runs portalkit-cli test-mitm 10.0.0.10:5654. - Asserts server rejection ("wrong PIN or MITM detected") and verdict PASSED. c) Live Network MITM Proxy Interception Test: - Generates a rogue self-signed TLS certificate. - Spawns an in-process multithreaded TLS reverse proxy listening on 127.0.0.1:8888 terminating client TLS with the rogue certificate and forwarding upstream to 10.0.0.10:5654. - Client connects to the proxy at 127.0.0.1:8888. - Extracts the rogue cert, initiates SRP handshake, computes M1 incorporating the rogue cert hash, and submits it through the proxy. - Asserts Portal TV rejects handshake with HTTP 401 ("Authentication failed (wrong PIN or MITM detected)"). d) Pinned Certificate TLS Challenge Rejection (all MITM-protected endpoints): - Client session pinned to Portal TV's genuine certificate. - For every post-pairing pinned path (/control/*, /video.h264, /audio.aac, /control/events): attempt through the rogue proxy must hard-fail at TLS (no HTTP bytes sent). - portalkit-cli control against the rogue proxy must also abort before sending requests. e) Live Rate-Limiting Enforcement: - Initiates a pairing session on Portal TV. - Sends 3 consecutive failed verification attempts. - Asserts attempt counter decrements: 2 left -> 1 left -> 0 left. - Asserts subsequent attempt is rejected due to session cancellation / wipeout. f) Legitimate End-to-End Pairing & Camera Control: - Initiates a pairing session on Portal TV. - Retrieves active PIN using adb shell logcat. - Executes portalkit-cli pair 10.0.0.10:5654 --pin . - Asserts pairing success and certificate pinned. - Executes portalkit-cli control 10.0.0.10:5654 mode Desk (expects {"ok":true}). - Reads state via control … state; asserts mode Desk. - Restores mode to DefaultAuto. """ import hashlib import json import os import re import secrets import socket import ssl import subprocess import tempfile import threading import time import unittest PORTAL_HOST = "10.0.0.10:5654" PORTAL_IP = "10.0.0.10" PORTAL_PORT = 5654 PORTALKIT_CLI = "/Users/zim/Projects/portaltv/mac2/PortalKit/.build/release/portalkit-cli" GENUINE_CERT_SHA256 = "314da0083dea8e80aa5f4194e51fb40ec27f7f40d39ea3ef8782f58ca79b3826" PROXY_HOST = "127.0.0.1" PROXY_PORT = 8888 # Post-pairing endpoints that MUST hard-fail under a rogue TLS cert (pin enforced). # Pairing (/auth/srp/*) intentionally accepts any leaf and relies on channel binding instead. MITM_PROTECTED_PATHS = [ "/control/state", "/control/mode?mode=Desk", "/control/fixed?x=0.5&y=0.5&scale=1.0", "/control/desk?tightness=0.5", "/control/events", "/video.h264", "/audio.aac", ] # CLI control verbs that map onto the same pinned HTTPS surface. MITM_PROTECTED_CLI_COMMANDS = [ ["state"], ["mode", "Desk"], ["fixed?x=0.5&y=0.5&scale=1.0"], ["desk?tightness=0.5"], ] # RFC 5054 2048-bit prime group N_HEX = ( "FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC74" "020BBEA63B139B22514A08798E3404DDEF9519B3CD3A431B302B0A6DF25F1437" "4FE1356D6D51C245E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7ED" "EE386BFB5A899FA5AE9F24117C4B1FE649286651ECE45B3DC2007CB8A163BF05" "98DA48361C55D39A69163FA8FD24CF5F83655D23DCA3AD961C62F356208552BB" "9ED529077096966D670C354E4ABC9804F1746C08CA18217C32905E462E36CE3B" "E39E772C180E86039B2783A2EC07A28FB5C55DF06F4C52C9DE2BCBF695581718" "3995497CEA956AE515D2261898FA051015728E5A8AACAA68FFFFFFFFFFFFFFFF" ) N = int(N_HEX, 16) g = 2 def pad256(val: int) -> bytes: return val.to_bytes(256, byteorder="big") SRP_K = int.from_bytes(hashlib.sha256(pad256(N) + pad256(g)).digest(), byteorder="big") def srp_compute_client_m1( salt_bytes: bytes, pub_b_int: int, pin_str: str, tls_cert_sha256_bytes: bytes, ): """ Computes SRP-6a client ephemeral key A and evidence M1 incorporating tls_cert_sha256. Returns (pad256(A).hex(), M1.hex()). """ a = secrets.randbelow(N - 2) + 1 A = pow(g, a, N) u = int.from_bytes(hashlib.sha256(pad256(A) + pad256(pub_b_int)).digest(), "big") x = int.from_bytes(hashlib.sha256(salt_bytes + pin_str.encode("utf-8")).digest(), "big") gx = pow(g, x, N) base = (pub_b_int - (SRP_K * gx) % N) % N exp = a + u * x S = pow(base, exp, N) K = hashlib.sha256(pad256(S)).digest() M1 = hashlib.sha256(pad256(A) + pad256(pub_b_int) + K + salt_bytes + tls_cert_sha256_bytes).digest() return pad256(A).hex(), M1.hex() def post_http_over_tls(host: str, port: int, path: str, payload_obj=None, timeout: float = 8.0): """Direct HTTPS POST request helper using raw TLS socket.""" ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE raw_s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) raw_s.settimeout(timeout) raw_s.connect((host, port)) ssl_s = ctx.wrap_socket(raw_s) body_bytes = json.dumps(payload_obj).encode("utf-8") if payload_obj is not None else b"" req_lines = [ f"POST {path} HTTP/1.1", f"Host: {host}:{port}", "Content-Type: application/json" if payload_obj is not None else "Accept: application/json", f"Content-Length: {len(body_bytes)}", "Connection: close", "", "", ] req_header = "\r\n".join(req_lines).encode("utf-8") ssl_s.sendall(req_header + body_bytes) resp_data = b"" while True: try: chunk = ssl_s.recv(4096) if not chunk: break resp_data += chunk except socket.timeout: break ssl_s.close() header_part, body_part = resp_data.split(b"\r\n\r\n", 1) status_line = header_part.split(b"\r\n")[0].decode("utf-8") status_code = int(status_line.split()[1]) try: body_json = json.loads(body_part.decode("utf-8")) except Exception: body_json = {"raw": body_part.decode("utf-8", errors="replace")} return status_code, body_json def retrieve_active_pin() -> str: """Retrieves active SRP PIN from Portal TV logcat.""" cmd = 'adb shell "logcat -d -s PortalService | grep \'SRP pairing started with PIN:\' | tail -1"' out = subprocess.check_output(cmd, shell=True, text=True).strip() match = re.search(r"SRP pairing started with PIN:\s*(\d{6})", out) if not match: raise ValueError(f"Could not parse PIN from logcat output: '{out}'") return match.group(1) class CertificatePinningMismatchError(Exception): """Raised when client TLS delegate rejects a server certificate mismatch.""" pass def pinned_https_get(host: str, port: int, path: str, pinned_sha256_hex: str, timeout: float = 5.0): """ Mimic PortalPinnedSessionDelegate: complete TLS, evaluate leaf pin, and only then send HTTP. On mismatch, abort with no request bytes written. """ ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE raw_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) raw_sock.settimeout(timeout) raw_sock.connect((host, port)) ssl_sock = ctx.wrap_socket(raw_sock) peer_der = ssl_sock.getpeercert(binary_form=True) peer_sha256 = hashlib.sha256(peer_der).hexdigest().lower() if peer_sha256 != pinned_sha256_hex.lower(): ssl_sock.close() raise CertificatePinningMismatchError( f"PortalKit TLS Pinning Mismatch! Expected: {pinned_sha256_hex}, Got: {peer_sha256}" ) req = ( f"GET {path} HTTP/1.1\r\n" f"Host: {host}:{port}\r\n" f"Connection: close\r\n\r\n" ).encode("utf-8") ssl_sock.sendall(req) data = ssl_sock.recv(4096) ssl_sock.close() return data class LiveMitmProxy: """ In-process multithreaded TLS reverse proxy listening on 127.0.0.1:8888. Terminates client TLS with a rogue self-signed certificate and forwards traffic upstream to 10.0.0.10:5654 over genuine TLS. """ def __init__( self, listen_host: str = PROXY_HOST, listen_port: int = PROXY_PORT, upstream_host: str = PORTAL_IP, upstream_port: int = PORTAL_PORT, ): self.listen_host = listen_host self.listen_port = listen_port self.upstream_host = upstream_host self.upstream_port = upstream_port self.temp_dir = tempfile.TemporaryDirectory() self.key_path = os.path.join(self.temp_dir.name, "rogue_mitm.key") self.cert_path = os.path.join(self.temp_dir.name, "rogue_mitm.crt") self._generate_rogue_cert() self.server_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) self.server_ctx.load_cert_chain(certfile=self.cert_path, keyfile=self.key_path) self.upstream_ctx = ssl.create_default_context() self.upstream_ctx.check_hostname = False self.upstream_ctx.verify_mode = ssl.CERT_NONE self.server_sock = None self.thread = None self.running = False self.client_requests_received = [] self.lock = threading.Lock() def _generate_rogue_cert(self): """Generates a rogue self-signed RSA-2048 certificate.""" cmd = [ "openssl", "req", "-x509", "-newkey", "rsa:2048", "-keyout", self.key_path, "-out", self.cert_path, "-days", "1", "-nodes", "-subj", "/CN=RogueMitmInterceptor/O=Attacker", ] subprocess.run(cmd, check=True, capture_output=True) @property def request_count(self) -> int: with self.lock: return len(self.client_requests_received) def clear_requests(self): with self.lock: self.client_requests_received.clear() def start(self): self.server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.server_sock.bind((self.listen_host, self.listen_port)) self.server_sock.listen(25) self.running = True self.thread = threading.Thread(target=self._accept_loop, daemon=True) self.thread.start() time.sleep(0.1) def stop(self): self.running = False if self.server_sock: try: self.server_sock.close() except Exception: pass self.temp_dir.cleanup() def _accept_loop(self): while self.running: try: client_raw, _ = self.server_sock.accept() except Exception: break threading.Thread(target=self._handle_client, args=(client_raw,), daemon=True).start() def _handle_client(self, client_raw: socket.socket): client_ssl = None upstream_ssl = None try: client_ssl = self.server_ctx.wrap_socket(client_raw, server_side=True) except Exception: try: client_raw.close() except Exception: pass return try: client_ssl.settimeout(6.0) initial_data = client_ssl.recv(4096) if not initial_data: return with self.lock: self.client_requests_received.append(initial_data) # Connect upstream to genuine Portal TV upstream_raw = socket.socket(socket.AF_INET, socket.SOCK_STREAM) upstream_raw.settimeout(6.0) upstream_ssl = self.upstream_ctx.wrap_socket(upstream_raw) upstream_ssl.connect((self.upstream_host, self.upstream_port)) upstream_ssl.sendall(initial_data) # Bidirectional forwarding def pipe(src, dst): try: while True: buf = src.recv(4096) if not buf: break dst.sendall(buf) except Exception: pass finally: try: dst.shutdown(socket.SHUT_WR) except Exception: pass t1 = threading.Thread(target=pipe, args=(client_ssl, upstream_ssl), daemon=True) t2 = threading.Thread(target=pipe, args=(upstream_ssl, client_ssl), daemon=True) t1.start() t2.start() t1.join() t2.join() except Exception: pass finally: if client_ssl is not None: try: client_ssl.close() except Exception: pass if upstream_ssl is not None: try: upstream_ssl.close() except Exception: pass try: client_raw.close() except Exception: pass class TestPortalTVLiveE2E(unittest.TestCase): """Automated Live End-to-End Test Suite against Portal TV.""" @classmethod def setUpClass(cls): self_check = subprocess.run([PORTALKIT_CLI, "--help"], capture_output=True, text=True) if self_check.returncode != 0: raise RuntimeError(f"portalkit-cli not executable at {PORTALKIT_CLI}") def test_01_status_verification(self): """ Scenario A: Status Verification - Runs `portalkit-cli status 10.0.0.10:5654`. - Asserts service is online, leaf cert SHA-256 is present. """ print("\n--- [Scenario A] Status Verification ---") cmd = [PORTALKIT_CLI, "status", PORTAL_HOST] proc = subprocess.run(cmd, capture_output=True, text=True) print(proc.stdout) self.assertEqual(proc.returncode, 0, f"portalkit-cli status failed: {proc.stderr}") self.assertIn("Service Online: YES", proc.stdout) self.assertIn("Leaf Cert SHA-256:", proc.stdout) # Check leaf cert matches genuine cert leaf_match = re.search(r"Leaf Cert SHA-256:\s+([a-f0-9]{64})", proc.stdout) self.assertIsNotNone(leaf_match, "Leaf Cert SHA-256 hex string not found in status output") presented_leaf = leaf_match.group(1).lower() self.assertEqual( presented_leaf, GENUINE_CERT_SHA256.lower(), f"Presented leaf cert {presented_leaf} does not match expected {GENUINE_CERT_SHA256}", ) print("✓ Service is online and genuine leaf cert SHA-256 verified.") def test_02_direct_channel_binding_defense_verification(self): """ Scenario B: Direct Channel Binding Defense Verification - Runs `portalkit-cli test-mitm 10.0.0.10:5654`. - Asserts server rejection ("wrong PIN or MITM detected") and verdict PASSED. """ print("\n--- [Scenario B] Direct Channel Binding Defense Verification ---") cmd = [PORTALKIT_CLI, "test-mitm", PORTAL_HOST] proc = subprocess.run(cmd, capture_output=True, text=True) print(proc.stdout) self.assertEqual(proc.returncode, 0, f"portalkit-cli test-mitm failed: {proc.stderr}") self.assertIn("wrong PIN or MITM detected", proc.stdout) self.assertIn("Verdict: PASSED", proc.stdout) print("✓ Direct channel binding defense successfully verified (verdict PASSED).") def test_03_live_network_mitm_proxy_interception(self): """ Scenario C: Live Network MITM Proxy Interception Test - Generates a rogue self-signed TLS certificate. - Spawns in-process multithreaded TLS reverse proxy listening on 127.0.0.1:8888 terminating client TLS with rogue cert and forwarding upstream to 10.0.0.10:5654. - Client connects to proxy at 127.0.0.1:8888. - Extracts rogue cert, initiates SRP handshake, computes M1 incorporating rogue cert hash, and submits through proxy. - Asserts Portal TV rejects handshake with HTTP 401 ("Authentication failed (wrong PIN or MITM detected)"). """ print("\n--- [Scenario C] Live Network MITM Proxy Interception Test ---") proxy = LiveMitmProxy(listen_host=PROXY_HOST, listen_port=PROXY_PORT) proxy.start() print(f"Rogue MITM proxy listening on {PROXY_HOST}:{PROXY_PORT}") try: # 1. Connect client to proxy over TLS and extract rogue cert client_ctx = ssl.create_default_context() client_ctx.check_hostname = False client_ctx.verify_mode = ssl.CERT_NONE s_init = client_ctx.wrap_socket(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) s_init.connect((PROXY_HOST, PROXY_PORT)) rogue_cert_der = s_init.getpeercert(binary_form=True) self.assertIsNotNone(rogue_cert_der, "Failed to capture peer certificate from proxy") rogue_cert_sha256 = hashlib.sha256(rogue_cert_der).digest() rogue_cert_hex = rogue_cert_sha256.hex() print(f"Captured Rogue Cert SHA-256: {rogue_cert_hex}") self.assertNotEqual( rogue_cert_hex.lower(), GENUINE_CERT_SHA256.lower(), "Rogue cert must differ from genuine Portal TV certificate", ) # 2. Initiate SRP pairing through the proxy req_init = ( f"POST /auth/srp/init HTTP/1.1\r\n" f"Host: {PROXY_HOST}:{PROXY_PORT}\r\n" f"Accept: application/json\r\n" f"Connection: close\r\n\r\n" ).encode("utf-8") s_init.sendall(req_init) resp_init_data = b"" while True: chunk = s_init.recv(4096) if not chunk: break resp_init_data += chunk s_init.close() _, init_body = resp_init_data.split(b"\r\n\r\n", 1) init_json = json.loads(init_body.decode("utf-8")) pairing_id = init_json["pairingId"] salt_bytes = bytes.fromhex(init_json["salt"]) pub_b = int(init_json["B"], 16) print(f"Pairing initiated through proxy: pairingId={pairing_id}") # 3. Retrieve PIN from device logcat pin = retrieve_active_pin() print(f"Active PIN retrieved from Portal TV: {pin}") # 4. Compute M1 bound to the ROGUE certificate pub_a_hex, m1_hex = srp_compute_client_m1( salt_bytes=salt_bytes, pub_b_int=pub_b, pin_str=pin, tls_cert_sha256_bytes=rogue_cert_sha256, ) print(f"Computed M1 bound to rogue cert: {m1_hex[:16]}…") # 5. Submit M1 through the proxy to Portal TV s_verify = client_ctx.wrap_socket(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) s_verify.connect((PROXY_HOST, PROXY_PORT)) verify_payload = json.dumps({ "pairingId": pairing_id, "A": pub_a_hex, "M1": m1_hex, }).encode("utf-8") req_verify = ( f"POST /auth/srp/verify HTTP/1.1\r\n" f"Host: {PROXY_HOST}:{PROXY_PORT}\r\n" f"Content-Type: application/json\r\n" f"Content-Length: {len(verify_payload)}\r\n" f"Connection: close\r\n\r\n" ).encode("utf-8") + verify_payload s_verify.sendall(req_verify) resp_verify_data = b"" while True: chunk = s_verify.recv(4096) if not chunk: break resp_verify_data += chunk s_verify.close() verify_headers, verify_body = resp_verify_data.split(b"\r\n\r\n", 1) status_code = int(verify_headers.split(b"\r\n")[0].split()[1]) verify_json = json.loads(verify_body.decode("utf-8")) print(f"Server response through proxy: HTTP {status_code} {verify_json}") # 6. Assert rejection due to channel binding mismatch self.assertEqual(status_code, 401, f"Expected HTTP 401, got {status_code}") rejection_message = verify_json.get("message", "") self.assertIn("wrong PIN or MITM detected", rejection_message) print("✓ Live network MITM proxy successfully rejected with HTTP 401 (channel binding defense held).") finally: proxy.stop() def test_04_pinned_certificate_tls_challenge_rejection(self): """ Scenario D: every MITM-protected endpoint must hard-fail under a rogue cert. Soft failures (system warning / continue / leak HTTP) are not acceptable: the client must abort at TLS pin check with zero request bytes observed by the proxy for each protected path and for portalkit-cli control verbs. """ print("\n--- [Scenario D] Pinned Certificate TLS Challenge Rejection (all protected endpoints) ---") proxy = LiveMitmProxy(listen_host=PROXY_HOST, listen_port=PROXY_PORT) proxy.start() print(f"Rogue MITM proxy listening on {PROXY_HOST}:{PROXY_PORT}") print(f"Protected paths under test: {len(MITM_PROTECTED_PATHS)}") try: for path in MITM_PROTECTED_PATHS: with self.subTest(path=path): proxy.clear_requests() hard_fail = False try: pinned_https_get( host=PROXY_HOST, port=PROXY_PORT, path=path, pinned_sha256_hex=GENUINE_CERT_SHA256, ) except CertificatePinningMismatchError as e: hard_fail = True print(f" {path}: hard-fail pin mismatch — {e}") self.assertTrue( hard_fail, f"{path}: expected CertificatePinningMismatchError hard-fail; " f"connection must not proceed past TLS (soft warning is insufficient)", ) self.assertEqual( proxy.request_count, 0, f"{path}: HTTP bytes were sent despite pin mismatch " f"({proxy.request_count} request(s) observed by rogue proxy)", ) print("✓ All MITM-protected HTTP paths hard-failed at pin check with zero request leakage.") for cmd_parts in MITM_PROTECTED_CLI_COMMANDS: with self.subTest(cli=" ".join(cmd_parts)): proxy.clear_requests() cli_res = subprocess.run( [PORTALKIT_CLI, "control", f"{PROXY_HOST}:{PROXY_PORT}", *cmd_parts], capture_output=True, text=True, ) combined = (cli_res.stdout + "\n" + cli_res.stderr).lower() print(f" portalkit-cli control {' '.join(cmd_parts)} → rc={cli_res.returncode}") print(f" stdout: {cli_res.stdout.strip()}") self.assertNotEqual( cli_res.returncode, 0, f"portalkit-cli control {' '.join(cmd_parts)} must exit non-zero on rogue proxy", ) # Must be a hard abort (cancel / fail / pin), not a soft warning with success. self.assertTrue( any( token in combined for token in ("cancelled", "cancel", "failed", "mismatch", "pinning", "error") ), f"portalkit-cli control {' '.join(cmd_parts)} did not report a hard failure: {combined!r}", ) self.assertEqual( proxy.request_count, 0, f"portalkit-cli control {' '.join(cmd_parts)} leaked HTTP through rogue proxy " f"({proxy.request_count} request(s))", ) print("✓ portalkit-cli control verbs hard-failed against rogue proxy with zero request leakage.") finally: proxy.stop() def test_05_live_rate_limiting_enforcement(self): """ Scenario E: Live Rate-Limiting Enforcement - Initiates a pairing session on Portal TV. - Sends 3 consecutive failed verification attempts. - Asserts attempt counter decrements: 2 left -> 1 left -> 0 left. - Asserts subsequent attempt is rejected due to session cancellation / wipeout. """ print("\n--- [Scenario E] Live Rate-Limiting Enforcement ---") # Ensure a clean pairing state (exhaust any stale pairing session) st_clean, init_clean = post_http_over_tls(PORTAL_IP, PORTAL_PORT, "/auth/srp/init") if st_clean == 200 and "pairingId" in init_clean: pid = init_clean["pairingId"] for _ in range(4): st_v, _ = post_http_over_tls( PORTAL_IP, PORTAL_PORT, "/auth/srp/verify", {"pairingId": pid, "A": "01" * 256, "M1": "00" * 32}, ) if st_v == 400: break # 1. Initiate fresh pairing session on Portal TV st_init, init_data = post_http_over_tls(PORTAL_IP, PORTAL_PORT, "/auth/srp/init") self.assertEqual(st_init, 200, f"Failed to initiate pairing: {init_data}") pairing_id = init_data["pairingId"] print(f"Initiated pairing session: {pairing_id}") dummy_payload = { "pairingId": pairing_id, "A": "01" * 256, "M1": "00" * 32, } # 2. Attempt 1 (expect 2 left) st1, r1 = post_http_over_tls(PORTAL_IP, PORTAL_PORT, "/auth/srp/verify", dummy_payload) print(f"Attempt 1: HTTP {st1}, resp: {r1}") self.assertEqual(st1, 401) self.assertEqual(r1.get("attemptsLeft"), 2) # 3. Attempt 2 (expect 1 left) st2, r2 = post_http_over_tls(PORTAL_IP, PORTAL_PORT, "/auth/srp/verify", dummy_payload) print(f"Attempt 2: HTTP {st2}, resp: {r2}") self.assertEqual(st2, 401) self.assertEqual(r2.get("attemptsLeft"), 1) # 4. Attempt 3 (expect 0 left and session wiped) st3, r3 = post_http_over_tls(PORTAL_IP, PORTAL_PORT, "/auth/srp/verify", dummy_payload) print(f"Attempt 3: HTTP {st3}, resp: {r3}") self.assertEqual(st3, 401) self.assertEqual(r3.get("attemptsLeft"), 0) # 5. Subsequent Attempt 4 (expect rejected due to session cancellation / wipeout) st4, r4 = post_http_over_tls(PORTAL_IP, PORTAL_PORT, "/auth/srp/verify", dummy_payload) print(f"Attempt 4: HTTP {st4}, resp: {r4}") self.assertIn(st4, [400, 401], f"Unexpected status code for wiped session: {st4}") is_wiped = (r4.get("error") == "no_active_pairing") or ("Too many failed attempts" in r4.get("message", "")) self.assertTrue(is_wiped, f"Subsequent attempt was not rejected due to wipeout: {r4}") print("✓ Rate limiting verified: counter decremented 2 -> 1 -> 0 and session wiped out.") def test_06_legitimate_end_to_end_pairing_and_camera_control(self): """ Scenario F: Legitimate End-to-End Pairing & Camera Control Mutations return {"ok":true}; live mode is read back via /control/state. """ print("\n--- [Scenario F] Legitimate End-to-End Pairing & Camera Control ---") # 1. Initiate pairing session on Portal TV st_init, init_data = post_http_over_tls(PORTAL_IP, PORTAL_PORT, "/auth/srp/init") self.assertEqual(st_init, 200, f"Init failed: {init_data}") print(f"Pairing session initiated: pairingId={init_data.get('pairingId')}") # 2. Retrieve active PIN via adb logcat command pin = retrieve_active_pin() print(f"Active PIN retrieved from logcat: {pin}") self.assertTrue(pin.isdigit() and len(pin) == 6, f"Invalid PIN format: {pin}") # 3. Execute portalkit-cli pair pair_cmd = [PORTALKIT_CLI, "pair", PORTAL_HOST, "--pin", pin] pair_proc = subprocess.run(pair_cmd, capture_output=True, text=True) print(pair_proc.stdout) self.assertEqual(pair_proc.returncode, 0, f"Pairing failed: {pair_proc.stderr}") self.assertIn("Pairing successful!", pair_proc.stdout) self.assertIn("TLS Certificate Pinned:", pair_proc.stdout) self.assertIn(GENUINE_CERT_SHA256.lower(), pair_proc.stdout.lower()) print("✓ Pairing successful and certificate pinned.") # 4. Execute portalkit-cli control mode Desk (ack only) ctrl_cmd = [PORTALKIT_CLI, "control", PORTAL_HOST, "mode", "Desk"] ctrl_proc = subprocess.run(ctrl_cmd, capture_output=True, text=True) print(ctrl_proc.stdout) self.assertEqual(ctrl_proc.returncode, 0, f"Control mode Desk failed: {ctrl_proc.stderr}") self.assertIn('"ok":true', ctrl_proc.stdout.replace(" ", "")) print("✓ Control command 'mode Desk' succeeded (HTTP 200, ack).") # 5. Read back state state_cmd = [PORTALKIT_CLI, "control", PORTAL_HOST, "state"] state_proc = subprocess.run(state_cmd, capture_output=True, text=True) print(state_proc.stdout) self.assertEqual(state_proc.returncode, 0, f"Control state failed: {state_proc.stderr}") self.assertIn('"mode":"Desk"', state_proc.stdout) print("✓ /control/state reports mode Desk.") # 6. Restore mode to DefaultAuto restore_cmd = [PORTALKIT_CLI, "control", PORTAL_HOST, "mode", "DefaultAuto"] restore_proc = subprocess.run(restore_cmd, capture_output=True, text=True) print(restore_proc.stdout) self.assertEqual(restore_proc.returncode, 0, f"Restore mode DefaultAuto failed: {restore_proc.stderr}") self.assertIn('"ok":true', restore_proc.stdout.replace(" ", "")) restore_state = subprocess.run(state_cmd, capture_output=True, text=True) print(restore_state.stdout) self.assertEqual(restore_state.returncode, 0) self.assertIn('"mode":"DefaultAuto"', restore_state.stdout) print("✓ Mode restored to DefaultAuto (ack + state).") if __name__ == "__main__": unittest.main(verbosity=2)