test_local_social_fixture.py (50502B)
1 import base64 2 import copy 3 import hashlib 4 import http.client 5 import importlib.util 6 import json 7 import re 8 import secrets 9 import tempfile 10 import time 11 import unittest 12 from pathlib import Path 13 from unittest import mock 14 15 from scripts.local_social_test_support import blossom_server 16 17 18 SCRIPT = Path(__file__).with_name("local-social-fixture.py") 19 SPEC = importlib.util.spec_from_file_location("local_social_fixture", SCRIPT) 20 assert SPEC is not None and SPEC.loader is not None 21 fixture = importlib.util.module_from_spec(SPEC) 22 SPEC.loader.exec_module(fixture) 23 24 25 def sign_bud11_event(event: dict) -> dict: 26 secret = secrets.randbelow(fixture.SECP256K1_ORDER - 1) + 1 27 public_point = fixture.point_multiply(secret, fixture.SECP256K1_GENERATOR) 28 assert public_point is not None 29 if public_point[1] & 1: 30 secret = fixture.SECP256K1_ORDER - secret 31 public_key = public_point[0].to_bytes(32, "big") 32 signed = copy.deepcopy(event) 33 signed["pubkey"] = public_key.hex() 34 preimage = json.dumps( 35 [ 36 0, 37 signed["pubkey"], 38 signed["created_at"], 39 signed["kind"], 40 signed["tags"], 41 signed["content"], 42 ], 43 ensure_ascii=False, 44 separators=(",", ":"), 45 ).encode("utf-8") 46 message = hashlib.sha256(preimage).digest() 47 signed["id"] = message.hex() 48 auxiliary = secrets.token_bytes(32) 49 masked_secret = bytes( 50 left ^ right 51 for left, right in zip( 52 secret.to_bytes(32, "big"), 53 fixture.tagged_hash("BIP0340/aux", auxiliary), 54 strict=True, 55 ) 56 ) 57 nonce = int.from_bytes( 58 fixture.tagged_hash("BIP0340/nonce", masked_secret + public_key + message), 59 "big", 60 ) % fixture.SECP256K1_ORDER 61 assert nonce != 0 62 nonce_point = fixture.point_multiply(nonce, fixture.SECP256K1_GENERATOR) 63 assert nonce_point is not None 64 if nonce_point[1] & 1: 65 nonce = fixture.SECP256K1_ORDER - nonce 66 challenge = int.from_bytes( 67 fixture.tagged_hash( 68 "BIP0340/challenge", 69 nonce_point[0].to_bytes(32, "big") + public_key + message, 70 ), 71 "big", 72 ) % fixture.SECP256K1_ORDER 73 signature = nonce_point[0].to_bytes(32, "big") + ( 74 (nonce + challenge * secret) % fixture.SECP256K1_ORDER 75 ).to_bytes(32, "big") 76 signed["sig"] = signature.hex() 77 assert fixture.valid_nostr_event(signed) 78 assert fixture.verify_nostr_signature(signed) 79 return signed 80 81 82 def bud11_event(now_unix_s: int, digest: str) -> dict: 83 return { 84 "created_at": now_unix_s - 5, 85 "kind": fixture.BUD11_EVENT_KIND, 86 "tags": [ 87 ["t", "upload"], 88 ["expiration", str(now_unix_s + 295)], 89 ["x", digest], 90 ["server", fixture.BUD11_SERVER_DOMAIN], 91 ], 92 "content": "Upload exact Tera image", 93 } 94 95 96 def mutate_bud11_event(event: dict, mutation: str, now_unix_s: int) -> tuple[dict, str]: 97 changed = copy.deepcopy(event) 98 if mutation == "none": 99 pass 100 elif mutation == "wrong_kind": 101 changed["kind"] = 1 102 elif mutation == "empty_content": 103 changed["content"] = "" 104 elif mutation == "oversized_content": 105 changed["content"] = "x" * (fixture.BUD11_CONTENT_MAX_BYTES + 1) 106 elif mutation == "leading_content_space": 107 changed["content"] = " Upload exact Tera image" 108 elif mutation == "control_content": 109 changed["content"] = "Upload\0image" 110 elif mutation == "missing_action": 111 changed["tags"].pop(0) 112 elif mutation == "wrong_action": 113 changed["tags"][0][1] = "delete" 114 elif mutation == "duplicate_action": 115 changed["tags"].insert(1, ["t", "upload"]) 116 elif mutation == "wrong_hash": 117 changed["tags"][2][1] = "f" * 64 118 elif mutation == "duplicate_hash": 119 changed["tags"].insert(3, copy.deepcopy(changed["tags"][2])) 120 elif mutation == "missing_server": 121 changed["tags"].pop(3) 122 elif mutation == "wrong_server": 123 changed["tags"][3][1] = "media.example" 124 elif mutation == "uppercase_server": 125 changed["tags"][3][1] = "Media.Example" 126 elif mutation == "duplicate_server": 127 changed["tags"].append(copy.deepcopy(changed["tags"][3])) 128 elif mutation == "missing_expiration": 129 changed["tags"].pop(1) 130 elif mutation == "noncanonical_expiration": 131 changed["tags"][1][1] = "0" + changed["tags"][1][1] 132 elif mutation == "expired": 133 changed["tags"][1][1] = str(now_unix_s) 134 elif mutation == "created_at_not_past": 135 changed["created_at"] = now_unix_s + 60 136 changed["tags"][1][1] = str(now_unix_s + 360) 137 elif mutation == "lifetime_too_long": 138 changed["tags"][1][1] = str(changed["created_at"] + 301) 139 elif mutation == "unknown_tag": 140 changed["tags"].append(["client", "radroots"]) 141 elif mutation in {"event_id", "signature", "authorization_scheme", "padded_base64"}: 142 pass 143 else: 144 raise AssertionError(f"unsupported mutation: {mutation}") 145 signed = sign_bud11_event(changed) 146 if mutation == "event_id": 147 signed["id"] = ("0" if signed["id"][0] != "0" else "1") + signed["id"][1:] 148 elif mutation == "signature": 149 signed["sig"] = ("0" if signed["sig"][0] != "0" else "1") + signed["sig"][1:] 150 raw = json.dumps(signed, ensure_ascii=False, separators=(",", ":")).encode("utf-8") 151 header = "Nostr " + base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") 152 if mutation == "authorization_scheme": 153 header = "Bearer " + header.removeprefix("Nostr ") 154 elif mutation == "padded_base64": 155 header += "=" 156 return signed, header 157 158 159 class LocalSocialFixtureTests(unittest.TestCase): 160 def persona_result(self) -> tuple[dict, dict]: 161 _, suite = fixture.load_persona_suite( 162 Path("test-fixtures/local-social-personas.v1.json") 163 ) 164 personas = [] 165 for persona in suite["personas"]: 166 personas.append( 167 { 168 "alias": persona["alias"], 169 "identity_sha256": hashlib.sha256( 170 persona["alias"].encode("ascii") 171 ).hexdigest(), 172 "subscriptions": 1, 173 "attempts": [ 174 { 175 "id": attempt["id"], 176 "flow": attempt["flow"], 177 "event_kind": fixture.FLOW_KINDS[attempt["flow"]], 178 "accepted": True, 179 "expected_failure_rejections": int( 180 attempt["expected_failure"] 181 == "transport_retry_relaunch" 182 ), 183 } 184 for attempt in persona["attempts"] 185 ], 186 } 187 ) 188 result = { 189 "schema": "tera.ios.local-social.persona-results.v1", 190 "schema_version": 1, 191 "run_id": "persona-result-test-001", 192 "source_commit": "1" * 40, 193 "source_tree": "2" * 40, 194 "fixture_sha256": "3" * 64, 195 "fixture_schema_sha256": "4" * 64, 196 "result_schema_sha256": "5" * 64, 197 "simulator": { 198 "udid": "11111111-2222-3333-4444-555555555555", 199 "os": "iOS 26.5", 200 "architecture": "arm64", 201 }, 202 "result_bundle_sha256": "6" * 64, 203 "evidence_sha256": "7" * 64, 204 "personas": personas, 205 "flow_counts": {flow: 3 for flow in fixture.FLOW_KINDS}, 206 "accepted_events": 15, 207 "event_kind_counts": {"1": 9, "31923": 3, "30402": 3}, 208 "accepted_uploads": 3, 209 "retrievals": 3, 210 "distinct_identities": 5, 211 "unknown_attempts": 0, 212 "duplicate_attempts": 0, 213 "expected_failure_rejections": 1, 214 "events_accepted_during_expected_failures": 0, 215 "production_network_contacts": 0, 216 "unintended_publications": 0, 217 "final_candidate_data_loss": 0, 218 "accessibility": { 219 "locale": "en_US", 220 "content_size": "accessibility-extra-extra-extra-large", 221 "reduce_motion": True, 222 "semantic_audit": "passed", 223 "voiceover_user_observed": False, 224 }, 225 "forward_repairs": [], 226 "complete_matrix_rerun": True, 227 } 228 return suite, result 229 230 def persona_attempt_attachments( 231 self, 232 *, 233 measured: bool = True, 234 ) -> tuple[dict, list[tuple[bytes, dict]]]: 235 _, suite = fixture.load_persona_suite( 236 Path("test-fixtures/local-social-personas.v1.json") 237 ) 238 attachments = [] 239 for persona in suite["personas"]: 240 identity = hashlib.sha256(persona["alias"].encode("ascii")).hexdigest() 241 for index, attempt in enumerate(persona["attempts"]): 242 validation = attempt["expected_failure"] == "validation_recovery" 243 retry = attempt["expected_failure"] == "transport_retry_relaunch" 244 network = {"state": "pending_step_258"} 245 if measured: 246 network = { 247 "state": "measured", 248 "accepted_connections": 1, 249 "rejected_connections": int(retry), 250 "non_loopback_attempts": 0, 251 "subscriptions": int(index == 0), 252 "accepted_events": 1, 253 "accepted_uploads": int(attempt["flow"] == "PhotoUpdate"), 254 "retrievals": int(attempt["flow"] == "PhotoUpdate"), 255 "unintended_publications": 0, 256 "events_accepted_during_expected_failure": 0, 257 "final_candidate_data_loss": 0, 258 } 259 value = { 260 "schema": fixture.PERSONA_ATTEMPT_SCHEMA, 261 "schema_version": 1, 262 "test_invocation": fixture.persona_invocation(), 263 "source": {"commit": "1" * 40, "tree": "2" * 40}, 264 "app_build_sha256": "3" * 64, 265 "simulator": { 266 "udid": "11111111-2222-3333-4444-555555555555", 267 "os": "iOS 26.5", 268 "architecture": "arm64", 269 }, 270 "run_id": "persona-result-test-001", 271 "persona_run_id": ( 272 f"persona-{persona['alias'].lower()}-" + "a" * 24 273 ), 274 "persona_alias": persona["alias"], 275 "attempt_id": attempt["id"], 276 "attempt_order": attempt["order"], 277 "flow": attempt["flow"], 278 "expected_failure": attempt["expected_failure"], 279 "public_identity_sha256": identity, 280 "endpoint_policy_sha256": "4" * 64, 281 "ui_observation": { 282 "validation_attempted": validation, 283 "validation_rejected": validation, 284 "retry_attempts": int(retry), 285 "relaunches": int(retry), 286 "retention_verified": True, 287 "today_projection_verified": True, 288 }, 289 "network_observation": network, 290 "accessibility": { 291 "locale": "en_US", 292 "content_size": "accessibility-extra-extra-extra-large", 293 "reduce_motion": True, 294 "progressive_disclosure": persona["interaction_profile"] 295 == "novice_progressive_disclosure", 296 "labels_values_traits": persona["interaction_profile"] 297 == "novice_accessibility_keyboard", 298 "keyboard_focus": persona["interaction_profile"] 299 == "novice_accessibility_keyboard", 300 "visible_actions": True, 301 "voiceover_user_observed": False, 302 }, 303 "artifact_digests": [], 304 } 305 raw = fixture.canonical_evidence_bytes(value) 306 attachments.append((raw, value)) 307 return suite, attachments 308 309 def test_bip340_reference_signature_and_mutation(self) -> None: 310 public_key = bytes.fromhex( 311 "f9308a019258c31049344f85f89d5229b531c845836f99b08601f113bce036f9" 312 ) 313 message = bytes(32) 314 signature = bytes.fromhex( 315 "e907831f80848d1069a5371b402410364bdf1c5f8307b0084c55f1ce2dca8215" 316 "25f66a4a85ea8b71e482a74f382d2ce5ebeee8fdb2172f477df4900d310536c0" 317 ) 318 self.assertTrue(fixture.verify_bip340(public_key, message, signature)) 319 self.assertFalse( 320 fixture.verify_bip340(public_key, message, signature[:-1] + b"\x00") 321 ) 322 323 def test_bud11_corpus_is_exact_bounded_and_contains_no_sensitive_evidence(self) -> None: 324 path = Path("test-fixtures/bud11-upload-authorization-mutations.v1.json") 325 raw, corpus = fixture.load_bud11_mutation_corpus(path) 326 self.assertLessEqual(len(raw), fixture.MAX_JSON_BYTES) 327 self.assertEqual(len(corpus["vectors"]), len(fixture.BUD11_MUTATIONS)) 328 self.assertNotRegex( 329 raw.decode("utf-8"), 330 r'"(?:private_key|secret|seed|authorization|event|sig|pubkey)"\s*:', 331 ) 332 333 def test_bud11_corpus_rejects_unknown_duplicate_and_missing_vectors(self) -> None: 334 path = Path("test-fixtures/bud11-upload-authorization-mutations.v1.json") 335 value = json.loads(path.read_text(encoding="utf-8")) 336 value["unknown"] = True 337 with self.assertRaisesRegex(ValueError, "field inventory"): 338 fixture.validate_bud11_mutation_corpus(value) 339 value = json.loads(path.read_text(encoding="utf-8")) 340 value["vectors"].pop() 341 with self.assertRaisesRegex(ValueError, "inventory"): 342 fixture.validate_bud11_mutation_corpus(value) 343 with tempfile.TemporaryDirectory() as directory: 344 duplicate = Path(directory, "duplicate.json") 345 duplicate.write_text( 346 '{"schema":"x","schema":"x","schema_version":1,"vectors":[]}\n', 347 encoding="utf-8", 348 ) 349 with self.assertRaisesRegex(ValueError, "duplicate JSON member"): 350 fixture.load_bud11_mutation_corpus(duplicate) 351 352 def test_bud11_scalar_bounds_and_server_domain_grammar_are_fail_closed(self) -> None: 353 self.assertEqual(fixture.canonical_unsigned_decimal("0"), 0) 354 self.assertEqual( 355 fixture.canonical_unsigned_decimal(str(0xFFFF_FFFF_FFFF_FFFF)), 356 0xFFFF_FFFF_FFFF_FFFF, 357 ) 358 for value in ["", "00", "+1", "١", "18446744073709551616", "1" * 10_000]: 359 self.assertIsNone(fixture.canonical_unsigned_decimal(value)) 360 for value in ["127.0.0.1", "media.example", "a-b.example"]: 361 self.assertTrue(fixture.valid_bud11_server_domain(value), value) 362 for value in [ 363 "127.000.0.1", 364 "123", 365 "Media.Example", 366 "https://media.example", 367 "media.example:443", 368 "-media.example", 369 "media-.example", 370 ]: 371 self.assertFalse(fixture.valid_bud11_server_domain(value), value) 372 oversized = "Nostr " + "a" * ( 373 fixture.BUD11_AUTHORIZATION_ENCODED_MAX_BYTES + 1 374 ) 375 self.assertFalse( 376 fixture.valid_blossom_authorization( 377 oversized, 378 "a" * 64, 379 fixture.BUD11_SERVER_DOMAIN, 380 1, 381 ) 382 ) 383 384 def test_bud11_mutation_corpus_matches_strict_admission_and_relay_denial(self) -> None: 385 _, corpus = fixture.load_bud11_mutation_corpus( 386 Path("test-fixtures/bud11-upload-authorization-mutations.v1.json") 387 ) 388 now = int(time.time()) 389 digest = hashlib.sha256(b"canonical-bud11-photo").hexdigest() 390 for vector in corpus["vectors"]: 391 with self.subTest(vector=vector["id"]): 392 event, header = mutate_bud11_event( 393 bud11_event(now, digest), vector["mutation"], now 394 ) 395 if vector["surface"] == "http": 396 accepted = fixture.valid_blossom_authorization( 397 header, 398 digest, 399 fixture.BUD11_SERVER_DOMAIN, 400 now, 401 ) 402 else: 403 with tempfile.TemporaryDirectory() as directory: 404 root = Path(directory) 405 state = fixture.FixtureState( 406 root / "evidence.json", root / "control", 21100 407 ) 408 accepted = bool(state.publish(event)) 409 self.assertEqual(accepted, vector["expected_accepted"]) 410 411 def test_local_blossom_http_runtime_enforces_the_shared_bud11_corpus(self) -> None: 412 _, corpus = fixture.load_bud11_mutation_corpus( 413 Path("test-fixtures/bud11-upload-authorization-mutations.v1.json") 414 ) 415 body = b"canonical-bud11-photo" 416 digest = hashlib.sha256(body).hexdigest() 417 now = int(time.time()) 418 with tempfile.TemporaryDirectory() as directory: 419 root = Path(directory) 420 control = root / "control" 421 control.touch() 422 state = fixture.FixtureState(root / "evidence.json", control, 0) 423 with blossom_server(fixture.BlossomHandler, state): 424 cases = [(v, "/upload", digest) for v in corpus["vectors"] if v["surface"] == "http"] 425 valid = next(v for v, _, _ in cases if v["expected_accepted"]) 426 cases += [(valid, f"/{digest}.png", digest), (valid, "/upload", ""), (valid, "/upload", digest.upper())] 427 for vector, path, supplied_hash in cases: 428 with self.subTest(vector=vector["id"], path=path, supplied_hash=supplied_hash): 429 _, header = mutate_bud11_event( 430 bud11_event(now, digest), vector["mutation"], now 431 ) 432 connection = http.client.HTTPConnection( 433 "127.0.0.1", state.blossom_port, timeout=5 434 ) 435 connection.request( 436 "PUT", 437 path, 438 body=body, 439 headers={ 440 "Authorization": header, 441 "Content-Type": "image/png", 442 "Content-Length": str(len(body)), 443 "X-SHA-256": supplied_hash, 444 }, 445 ) 446 response = connection.getresponse() 447 response.read() 448 connection.close() 449 self.assertEqual( 450 response.status == 200, 451 vector["expected_accepted"] and path == "/upload" and supplied_hash == digest, 452 ) 453 evidence = json.loads((root / "evidence.json").read_text(encoding="utf-8")) 454 self.assertEqual(evidence["accepted_uploads"], 1) 455 456 def test_fixture_rejects_duplicate_and_unknown_fields(self) -> None: 457 source = Path("test-fixtures/local-social-personas.v1.json") 458 payload = json.loads(source.read_text(encoding="utf-8")) 459 payload["unexpected"] = True 460 with self.assertRaisesRegex(ValueError, "field inventory"): 461 fixture.validate_persona_suite(payload) 462 463 with tempfile.TemporaryDirectory() as directory: 464 duplicate = Path(directory, "duplicate.json") 465 duplicate.write_text('{"schema":1,"schema":1}\n', encoding="utf-8") 466 with self.assertRaisesRegex(ValueError, "duplicate JSON member"): 467 fixture.read_json(duplicate) 468 469 def test_json_reads_enforce_maximum_plus_one_before_decoding(self) -> None: 470 with tempfile.TemporaryDirectory() as directory: 471 path = Path(directory, "bounded.json") 472 path.write_bytes(b'{"x":1}') 473 self.assertEqual(fixture.read_json(path, 7)[1], {"x": 1}) 474 path.write_bytes(b'{"x":1}\n') 475 with self.assertRaisesRegex(ValueError, "byte bound"): 476 fixture.read_json(path, 7) 477 478 def test_schemas_pass_meta_validation_and_match_semantic_corpora(self) -> None: 479 fixture_raw, suite = fixture.load_persona_suite( 480 Path("test-fixtures/local-social-personas.v1.json") 481 ) 482 self.assertTrue(fixture_raw) 483 _, persona_schema = fixture.load_schema_file( 484 Path("test-fixtures/local-social-personas.v1.schema.json"), 485 "https://radroots.org/schemas/ios/local-social-personas.v1.schema.json", 486 ) 487 fixture.validate_schema_instance(persona_schema, suite, "persona fixture") 488 _, corpus = fixture.load_bud11_mutation_corpus( 489 Path("test-fixtures/bud11-upload-authorization-mutations.v1.json") 490 ) 491 _, corpus_schema = fixture.load_schema_file( 492 Path( 493 "test-fixtures/bud11-upload-authorization-mutations.v1.schema.json" 494 ), 495 "https://radroots.org/schemas/ios/bud11-upload-authorization-mutations.v1.schema.json", 496 ) 497 fixture.validate_schema_instance(corpus_schema, corpus, "BUD-11 corpus") 498 499 changed = copy.deepcopy(suite) 500 changed["unexpected"] = True 501 with self.assertRaisesRegex(ValueError, "disagrees"): 502 fixture.validate_schema_instance(persona_schema, changed, "persona fixture") 503 504 with tempfile.TemporaryDirectory() as directory: 505 invalid = Path(directory, "invalid.schema.json") 506 invalid.write_text( 507 json.dumps( 508 { 509 "$schema": "https://json-schema.org/draft/2020-12/schema", 510 "$id": "https://radroots.org/schemas/ios/invalid.json", 511 "type": "object", 512 "additionalProperties": False, 513 "properties": {"x": {"type": "not-a-json-schema-type"}}, 514 } 515 ), 516 encoding="utf-8", 517 ) 518 with self.assertRaisesRegex(ValueError, "meta-validation"): 519 fixture.load_schema_file( 520 invalid, "https://radroots.org/schemas/ios/invalid.json" 521 ) 522 invalid.write_text( 523 json.dumps( 524 { 525 "$schema": "https://json-schema.org/draft/2020-12/schema", 526 "$id": "https://radroots.org/schemas/ios/invalid.json", 527 "type": "object", 528 "additionalProperties": False, 529 "properties": { 530 "x": {"$ref": "https://example.com/external.json"} 531 }, 532 } 533 ), 534 encoding="utf-8", 535 ) 536 with self.assertRaisesRegex(ValueError, "external reference"): 537 fixture.load_schema_file( 538 invalid, "https://radroots.org/schemas/ios/invalid.json" 539 ) 540 541 def test_result_bundle_digest_is_framed_bounded_and_pinned(self) -> None: 542 with tempfile.TemporaryDirectory() as directory: 543 root = Path(directory) 544 (root / "a.txt").write_bytes(b"A") 545 (root / "nested").mkdir() 546 (root / "nested" / "b.bin").write_bytes(b"\x00B") 547 self.assertEqual( 548 fixture.directory_digest(root, 3, 2, 3), 549 "25fb5f36c3b044de2716ddfbbd95c2e5eb39cf38a5bcf2feb793cd57ce27147a", 550 ) 551 with self.assertRaisesRegex(ValueError, "entry bound"): 552 fixture.directory_digest(root, 2, 2, 3) 553 with self.assertRaisesRegex(ValueError, "byte bound"): 554 fixture.directory_digest(root, 3, 1, 3) 555 with self.assertRaisesRegex(ValueError, "byte bound"): 556 fixture.directory_digest(root, 3, 2, 2) 557 558 alternate = root / "alternate" 559 alternate.mkdir() 560 (alternate / "a").write_bytes(b".txtA") 561 self.assertNotEqual( 562 fixture.directory_digest(root, 5, 8, 16), 563 fixture.directory_digest(alternate, 1, 8, 8), 564 ) 565 566 def test_fixture_freezes_exact_persona_and_flow_matrix(self) -> None: 567 _, suite = fixture.load_persona_suite( 568 Path("test-fixtures/local-social-personas.v1.json") 569 ) 570 attempts = fixture.persona_attempts(suite) 571 self.assertEqual( 572 tuple(persona["alias"] for persona in suite["personas"]), 573 fixture.PERSONA_ALIASES, 574 ) 575 self.assertEqual(len(attempts), 15) 576 self.assertEqual( 577 { 578 flow: sum(item["flow"] == flow for item in attempts.values()) 579 for flow in fixture.FLOW_KINDS 580 }, 581 {flow: 3 for flow in fixture.FLOW_KINDS}, 582 ) 583 584 def test_legacy_fixture_control_remains_file_presence_based(self) -> None: 585 body = b"legacy-photo" 586 digest = fixture.hashlib.sha256(body).hexdigest() 587 with tempfile.TemporaryDirectory() as directory: 588 root = Path(directory) 589 control = root / "control" 590 control.touch() 591 state = fixture.FixtureState(root / "evidence.json", control, 21100) 592 with mock.patch.object( 593 fixture, "valid_blossom_authorization", return_value=True 594 ): 595 accepted, _ = state.upload(body, "image/png", digest, "Nostr valid") 596 self.assertTrue(accepted) 597 self.assertIsNotNone(state.retrieve(digest)) 598 evidence = json.loads((root / "evidence.json").read_text(encoding="utf-8")) 599 self.assertEqual(evidence["accepted_uploads"], 1) 600 self.assertEqual(evidence["retrievals"], 1) 601 602 def test_persona_retrieval_counts_only_its_own_uploaded_digest(self) -> None: 603 body = b"persona-photo" 604 digest = fixture.hashlib.sha256(body).hexdigest() 605 _, suite = fixture.load_persona_suite( 606 Path("test-fixtures/local-social-personas.v1.json") 607 ) 608 with tempfile.TemporaryDirectory() as directory: 609 root = Path(directory) 610 control = root / "control.json" 611 state = fixture.FixtureState( 612 root / "evidence.json", control, 21100, suite 613 ) 614 self.write_persona_control(control, "P01") 615 with mock.patch.object( 616 fixture, "valid_blossom_authorization", return_value=True 617 ): 618 accepted, _ = state.upload(body, "image/png", digest, "Nostr valid") 619 self.assertTrue(accepted) 620 self.assertIsNotNone(state.retrieve(digest)) 621 self.write_persona_control(control, "P02") 622 self.assertIsNotNone(state.retrieve(digest)) 623 624 evidence = json.loads( 625 (root / "evidence.json").read_text(encoding="utf-8") 626 ) 627 self.assertEqual(evidence["retrievals"], 1) 628 self.assertEqual(evidence["personas"][0]["retrievals"], 1) 629 self.assertEqual(evidence["personas"][1]["retrievals"], 0) 630 631 def test_transport_retry_drops_one_response_then_accepts(self) -> None: 632 _, suite = fixture.load_persona_suite( 633 Path("test-fixtures/local-social-personas.v1.json") 634 ) 635 attempt = fixture.persona_attempts(suite)["P05-A01"] 636 event = { 637 "id": "1" * 64, 638 "kind": fixture.FLOW_KINDS[attempt["flow"]], 639 "pubkey": "2" * 64, 640 } 641 with tempfile.TemporaryDirectory() as directory: 642 root = Path(directory) 643 control = root / "control.json" 644 self.write_persona_control(control, "P05") 645 state = fixture.FixtureState( 646 root / "evidence.json", control, 21100, suite 647 ) 648 with mock.patch.object( 649 fixture, "classify_attempt", return_value=attempt 650 ): 651 self.assertIsNone(state._publish_persona_event(event)) 652 self.assertTrue(state._publish_persona_event(event)) 653 654 evidence = json.loads( 655 (root / "evidence.json").read_text(encoding="utf-8") 656 ) 657 self.assertEqual(evidence["expected_failure_rejections"], 1) 658 self.assertEqual(evidence["accepted_events"], 1) 659 660 @staticmethod 661 def write_persona_control(path: Path, alias: str) -> None: 662 path.write_text( 663 json.dumps( 664 { 665 "schema": fixture.PERSONA_CONTROL_SCHEMA, 666 "active_persona": alias, 667 "blossom_enabled": True, 668 } 669 ), 670 encoding="utf-8", 671 ) 672 673 def test_result_contract_accepts_future_ios_and_rejects_drift(self) -> None: 674 suite, result = self.persona_result() 675 result_schema = json.loads( 676 Path("test-fixtures/local-social-persona-results.v1.schema.json").read_text( 677 encoding="utf-8" 678 ) 679 ) 680 os_pattern = result_schema["properties"]["simulator"]["properties"]["os"][ 681 "pattern" 682 ] 683 self.assertIsNotNone(re.fullmatch(os_pattern, "iOS 26.5")) 684 self.assertIsNone(re.fullmatch(os_pattern, "iOS 17.7")) 685 self.assertIs( 686 fixture.validate_persona_result( 687 result, suite, "3" * 64, "4" * 64, "5" * 64 688 ), 689 result, 690 ) 691 fixture.validate_schema_instance(result_schema, result, "persona v1 result") 692 for mutation in ( 693 lambda value: value.update({"unexpected": True}), 694 lambda value: value["simulator"].update({"os": "iOS 17.7"}), 695 lambda value: value["accessibility"].update( 696 {"voiceover_user_observed": True} 697 ), 698 ): 699 changed = copy.deepcopy(result) 700 mutation(changed) 701 with self.assertRaises(ValueError): 702 fixture.validate_persona_result( 703 changed, suite, "3" * 64, "4" * 64, "5" * 64 704 ) 705 706 def test_attempt_evidence_is_strict_bound_and_secret_free(self) -> None: 707 suite, attachments = self.persona_attempt_attachments(measured=False) 708 for raw, value in attachments: 709 self.assertLessEqual(len(raw), fixture.MAX_PERSONA_ATTACHMENT_BYTES) 710 self.assertIs( 711 fixture.validate_persona_attempt_evidence( 712 value, suite, require_measured_network=False 713 ), 714 value, 715 ) 716 self.assertNotRegex( 717 raw.decode("utf-8"), 718 r'"(?:private_key|secret|seed|signed_event|event_content|raw_event)"', 719 ) 720 with self.assertRaisesRegex(ValueError, "not measured"): 721 fixture.validate_persona_attempt_evidence( 722 value, suite, require_measured_network=True 723 ) 724 725 def test_attempt_evidence_rejects_one_field_identity_and_outcome_drift(self) -> None: 726 suite, attachments = self.persona_attempt_attachments() 727 _, canonical = attachments[0] 728 mutations = ( 729 lambda value: value.update({"unknown": True}), 730 lambda value: value["test_invocation"].update( 731 {"identifier": "TeraUITests/testOther"} 732 ), 733 lambda value: value["source"].update({"tree": "A" * 40}), 734 lambda value: value.update({"app_build_sha256": "0" * 63}), 735 lambda value: value.update({"run_id": "x"}), 736 lambda value: value.update({"persona_alias": "P02"}), 737 lambda value: value.update({"attempt_order": 2}), 738 lambda value: value["ui_observation"].update( 739 {"today_projection_verified": False} 740 ), 741 lambda value: value["network_observation"].update( 742 {"non_loopback_attempts": 1} 743 ), 744 lambda value: value["accessibility"].update( 745 {"voiceover_user_observed": True} 746 ), 747 lambda value: value.update({"private_key": "canary"}), 748 ) 749 for mutation in mutations: 750 changed = copy.deepcopy(canonical) 751 mutation(changed) 752 with self.assertRaises(ValueError): 753 fixture.validate_persona_attempt_evidence( 754 changed, suite, require_measured_network=True 755 ) 756 757 def test_persona_result_v2_is_reconstructed_only_from_measured_attempts(self) -> None: 758 suite, attachments = self.persona_attempt_attachments() 759 result = fixture.reconstruct_persona_result_v2( 760 suite, 761 attachments, 762 fixture_sha256="5" * 64, 763 fixture_schema_sha256="6" * 64, 764 attempt_schema_sha256="7" * 64, 765 result_schema_sha256="8" * 64, 766 result_bundle_sha256="9" * 64, 767 forward_repairs=[], 768 ) 769 self.assertEqual(result["schema"], fixture.PERSONA_RESULT_V2_SCHEMA) 770 self.assertEqual(result["accepted_events"], 15) 771 self.assertEqual(result["expected_failure_rejections"], 2) 772 self.assertEqual(result["accepted_uploads"], 3) 773 self.assertEqual(result["retrievals"], 3) 774 self.assertEqual(result["non_loopback_attempts"], 0) 775 self.assertEqual(len(result["attachments"]), 15) 776 _, attempt_schema = fixture.load_schema_file( 777 Path( 778 "test-fixtures/local-social-persona-attempt-evidence.v1.schema.json" 779 ), 780 "https://radroots.org/schemas/ios/local-social-persona-attempt-evidence.v1.schema.json", 781 ) 782 for _, attempt in attachments: 783 fixture.validate_schema_instance( 784 attempt_schema, attempt, "persona attempt evidence" 785 ) 786 _, result_schema = fixture.load_schema_file( 787 Path("test-fixtures/local-social-persona-results.v2.schema.json"), 788 "https://radroots.org/schemas/ios/local-social-persona-results.v2.schema.json", 789 ) 790 fixture.validate_schema_instance(result_schema, result, "persona v2 result") 791 792 _, pending = self.persona_attempt_attachments(measured=False) 793 with self.assertRaisesRegex(ValueError, "not measured"): 794 fixture.reconstruct_persona_result_v2( 795 suite, 796 pending, 797 fixture_sha256="5" * 64, 798 fixture_schema_sha256="6" * 64, 799 attempt_schema_sha256="7" * 64, 800 result_schema_sha256="8" * 64, 801 result_bundle_sha256="9" * 64, 802 forward_repairs=[], 803 ) 804 805 def test_xcresult_test_identity_and_attachment_inventory_are_exact(self) -> None: 806 suite, attachments = self.persona_attempt_attachments() 807 tests = { 808 "devices": [], 809 "testNodes": [ 810 { 811 "children": [ 812 { 813 "name": "testLocalSocialDeterministicPersonas()", 814 "nodeIdentifier": fixture.PERSONA_XCRESULT_NODE_IDENTIFIER, 815 "nodeIdentifierURL": fixture.PERSONA_XCRESULT_NODE_URL, 816 "nodeType": "Test Case", 817 "result": "Passed", 818 } 819 ], 820 "name": "Tera", 821 "nodeType": "Test Plan", 822 "result": "Passed", 823 } 824 ], 825 "testPlanConfigurations": [], 826 } 827 self.assertEqual( 828 fixture.exact_persona_test_node(tests)["result"], "Passed" 829 ) 830 for mutation in ( 831 lambda value: value["testNodes"][0]["children"][0].update( 832 {"result": "Failed"} 833 ), 834 lambda value: value["testNodes"][0]["children"][0].update( 835 {"nodeIdentifierURL": "test://wrong"} 836 ), 837 lambda value: value["testNodes"].append( 838 copy.deepcopy(value["testNodes"][0]) 839 ), 840 ): 841 changed = copy.deepcopy(tests) 842 mutation(changed) 843 with self.assertRaises(ValueError): 844 fixture.exact_persona_test_node(changed) 845 846 with tempfile.TemporaryDirectory() as directory: 847 root = Path(directory) 848 rows = [] 849 for index, (name, (raw, _)) in enumerate( 850 zip(fixture.PERSONA_ATTACHMENT_NAMES, attachments, strict=True) 851 ): 852 exported = "exported-" + name 853 suggested = ( 854 f"{name[:-5]}_0_00000000-0000-0000-0000-{index:012X}.json" 855 ) 856 (root / exported).write_bytes(raw) 857 rows.append( 858 { 859 "exportedFileName": exported, 860 "suggestedHumanReadableName": suggested, 861 "isAssociatedWithFailure": False, 862 "configurationName": "Test Scheme Action", 863 "deviceName": "iPhone 17 Pro", 864 "deviceId": "11111111-2222-3333-4444-555555555555", 865 } 866 ) 867 manifest = [ 868 { 869 "testIdentifier": fixture.PERSONA_XCRESULT_NODE_IDENTIFIER, 870 "testIdentifierURL": fixture.PERSONA_XCRESULT_NODE_URL, 871 "attachments": rows, 872 } 873 ] 874 (root / "manifest.json").write_text( 875 json.dumps(manifest), encoding="utf-8" 876 ) 877 loaded = fixture.load_exported_persona_attachments( 878 root, suite, require_measured_network=True 879 ) 880 self.assertEqual(len(loaded), 15) 881 self.assertEqual(loaded[0][1]["attempt_id"], "P01-A01") 882 883 unexpected = root / "unexpected.json" 884 unexpected.write_text("{}", encoding="utf-8") 885 with self.assertRaisesRegex(ValueError, "entry bound"): 886 fixture.load_exported_persona_attachments( 887 root, suite, require_measured_network=True 888 ) 889 unexpected.unlink() 890 891 rows[0]["suggestedHumanReadableName"] = rows[1][ 892 "suggestedHumanReadableName" 893 ] 894 (root / "manifest.json").write_text( 895 json.dumps(manifest), encoding="utf-8" 896 ) 897 with self.assertRaises(ValueError): 898 fixture.load_exported_persona_attachments( 899 root, suite, require_measured_network=True 900 ) 901 902 rows[0]["suggestedHumanReadableName"] = ( 903 "radroots-local-social-P01-A01.json" 904 ) 905 (root / "manifest.json").write_text( 906 json.dumps(manifest), encoding="utf-8" 907 ) 908 with self.assertRaises(ValueError): 909 fixture.load_exported_persona_attachments( 910 root, suite, require_measured_network=True 911 ) 912 913 def test_xcresult_attachment_read_rejects_maximum_plus_one(self) -> None: 914 suite, attachments = self.persona_attempt_attachments() 915 with tempfile.TemporaryDirectory() as directory: 916 root = Path(directory) 917 name = ( 918 "radroots-local-social-P01-A01_0_" 919 "00000000-0000-0000-0000-000000000000.json" 920 ) 921 exported = "oversized.json" 922 (root / exported).write_bytes( 923 b"{" + b" " * fixture.MAX_PERSONA_ATTACHMENT_BYTES + b"}" 924 ) 925 rows = [ 926 { 927 "exportedFileName": exported, 928 "suggestedHumanReadableName": name, 929 "isAssociatedWithFailure": False, 930 "configurationName": "Test Scheme Action", 931 "deviceName": "iPhone 17 Pro", 932 "deviceId": "11111111-2222-3333-4444-555555555555", 933 } 934 ] 935 for index, canonical_name in enumerate( 936 fixture.PERSONA_ATTACHMENT_NAMES[1:], 1 937 ): 938 other_name = ( 939 f"{canonical_name[:-5]}_0_" 940 f"00000000-0000-0000-0000-{index:012X}.json" 941 ) 942 other_exported = f"exported-{index}.json" 943 (root / other_exported).write_bytes(attachments[index][0]) 944 rows.append( 945 { 946 "exportedFileName": other_exported, 947 "suggestedHumanReadableName": other_name, 948 "isAssociatedWithFailure": False, 949 "configurationName": "Test Scheme Action", 950 "deviceName": "iPhone 17 Pro", 951 "deviceId": "11111111-2222-3333-4444-555555555555", 952 } 953 ) 954 (root / "manifest.json").write_text( 955 json.dumps( 956 [ 957 { 958 "testIdentifier": fixture.PERSONA_XCRESULT_NODE_IDENTIFIER, 959 "testIdentifierURL": fixture.PERSONA_XCRESULT_NODE_URL, 960 "attachments": rows, 961 } 962 ] 963 ), 964 encoding="utf-8", 965 ) 966 with self.assertRaisesRegex(ValueError, "byte bound"): 967 fixture.load_exported_persona_attachments( 968 root, suite, require_measured_network=True 969 ) 970 971 def test_simulator_metadata_uses_exact_result_bundle_device(self) -> None: 972 udid = "11111111-2222-3333-4444-555555555555" 973 result_bundle = Path("result.xcresult") 974 simctl_devices = { 975 "devices": { 976 "com.apple.CoreSimulator.SimRuntime.iOS-26-5": [ 977 { 978 "udid": udid, 979 "isAvailable": True, 980 "name": "iPhone 17 Pro", 981 } 982 ] 983 } 984 } 985 result_summary = { 986 "failedTests": 0, 987 "passedTests": 1, 988 "result": "Passed", 989 "skippedTests": 0, 990 "devicesAndConfigurations": [ 991 { 992 "device": { 993 "architecture": "arm64", 994 "deviceId": udid, 995 "osVersion": "26.5", 996 "platform": "iOS Simulator", 997 } 998 } 999 ], 1000 } 1001 1002 with mock.patch.object( 1003 fixture, 1004 "run_json_command_bounded", 1005 side_effect=[simctl_devices, result_summary], 1006 ) as run_json: 1007 self.assertEqual( 1008 fixture.simulator_metadata(udid.lower(), result_bundle), 1009 {"udid": udid, "os": "iOS 26.5.0", "architecture": "arm64"}, 1010 ) 1011 commands = [call.args[0] for call in run_json.call_args_list] 1012 self.assertEqual(commands[0][:3], ["xcrun", "simctl", "list"]) 1013 self.assertEqual(commands[1][:3], ["xcrun", "xcresulttool", "get"]) 1014 self.assertNotIn("spawn", commands[0] + commands[1]) 1015 1016 patch_summary = copy.deepcopy(result_summary) 1017 patch_summary["devicesAndConfigurations"][0]["device"]["osVersion"] = ( 1018 "26.5.1" 1019 ) 1020 patch_devices = { 1021 "devices": { 1022 "com.apple.CoreSimulator.SimRuntime.iOS-26-5-1": [ 1023 simctl_devices["devices"][ 1024 "com.apple.CoreSimulator.SimRuntime.iOS-26-5" 1025 ][0] 1026 ] 1027 } 1028 } 1029 with mock.patch.object( 1030 fixture, 1031 "run_json_command_bounded", 1032 side_effect=[patch_devices, patch_summary], 1033 ): 1034 self.assertEqual( 1035 fixture.simulator_metadata(udid, result_bundle)["os"], "iOS 26.5.1" 1036 ) 1037 1038 for mutation in ( 1039 lambda value: value["devicesAndConfigurations"][0]["device"].update( 1040 {"architecture": "x86_64"} 1041 ), 1042 lambda value: value["devicesAndConfigurations"][0]["device"].update( 1043 {"deviceId": "AAAAAAAA-BBBB-CCCC-DDDD-EEEEEEEEEEEE"} 1044 ), 1045 lambda value: value["devicesAndConfigurations"][0]["device"].update( 1046 {"osVersion": "17.7"} 1047 ), 1048 lambda value: value["devicesAndConfigurations"][0]["device"].update( 1049 {"platform": "macOS"} 1050 ), 1051 lambda value: value.update({"result": "Failed", "failedTests": 1}), 1052 ): 1053 changed = copy.deepcopy(result_summary) 1054 mutation(changed) 1055 with ( 1056 mock.patch.object( 1057 fixture, 1058 "run_json_command_bounded", 1059 side_effect=[simctl_devices, changed], 1060 ), 1061 self.assertRaises(ValueError), 1062 ): 1063 fixture.simulator_metadata(udid, result_bundle) 1064 1065 def test_subprocess_json_is_bounded_before_decoding(self) -> None: 1066 command = [ 1067 fixture.sys.executable, 1068 "-c", 1069 "import sys; sys.stdout.write('{\\\"x\\\":1}')", 1070 ] 1071 self.assertEqual(fixture.run_json_command_bounded(command, 7), {"x": 1}) 1072 with self.assertRaisesRegex(ValueError, "byte bound"): 1073 fixture.run_json_command_bounded(command, 6) 1074 1075 def test_verifier_toolchain_identity_is_exact(self) -> None: 1076 fixture.verify_toolchain_identity() 1077 with ( 1078 mock.patch.object(fixture.sys, "version_info", (3, 14, 6)), 1079 self.assertRaisesRegex(RuntimeError, "Python identity"), 1080 ): 1081 fixture.verify_toolchain_identity() 1082 with ( 1083 mock.patch.object( 1084 fixture.importlib.metadata, "version", return_value="4.25.1" 1085 ), 1086 self.assertRaisesRegex(RuntimeError, "schema dependency"), 1087 ): 1088 fixture.verify_toolchain_identity() 1089 1090 def test_control_is_bounded_and_deny_unknown(self) -> None: 1091 with tempfile.TemporaryDirectory() as directory: 1092 path = Path(directory, "control.json") 1093 path.write_text( 1094 json.dumps( 1095 { 1096 "schema": fixture.PERSONA_CONTROL_SCHEMA, 1097 "active_persona": "P03", 1098 "blossom_enabled": True, 1099 } 1100 ), 1101 encoding="utf-8", 1102 ) 1103 self.assertEqual(fixture.read_control(path)["active_persona"], "P03") 1104 path.write_text( 1105 json.dumps( 1106 { 1107 "schema": fixture.PERSONA_CONTROL_SCHEMA, 1108 "active_persona": "P03", 1109 "blossom_enabled": True, 1110 "secret": "forbidden", 1111 } 1112 ), 1113 encoding="utf-8", 1114 ) 1115 self.assertIsNone(fixture.read_control(path)) 1116 1117 def test_observable_loopback_factory_counts_and_rejects_connections(self) -> None: 1118 with tempfile.TemporaryDirectory() as directory: 1119 root = Path(directory) 1120 state = fixture.FixtureState(root / "evidence.json", root / "control", 0) 1121 server = fixture.LoopbackConnectionFactory.relay(0, state) 1122 try: 1123 self.assertTrue( 1124 server.verify_request(mock.Mock(), ("127.0.0.1", 20_000)) 1125 ) 1126 self.assertFalse( 1127 server.verify_request(mock.Mock(), ("192.0.2.1", 20_001)) 1128 ) 1129 finally: 1130 server.server_close() 1131 _, evidence = fixture.read_json(root / "evidence.json") 1132 self.assertEqual(evidence["accepted_connections"], 1) 1133 self.assertEqual(evidence["rejected_connections"], 1) 1134 self.assertEqual(evidence["non_loopback_attempts"], 1) 1135 self.assertEqual(evidence["production_network_contacts"], 1) 1136 1137 def test_attempt_classification_accepts_only_exact_photo_wire_shape(self) -> None: 1138 marker = "rr-p01-a02-photo" 1139 digest = "a" * 64 1140 url = f"http://127.0.0.1:21101/{digest}.png" 1141 attempt = {"flow": "PhotoUpdate", "marker": marker} 1142 event = { 1143 "content": f"{marker}\n{url}", 1144 "tags": [ 1145 [ 1146 "imeta", 1147 f"url {url}", 1148 f"x {digest}", 1149 "m image/png", 1150 "dim 1x1", 1151 "size 1", 1152 "alt Local qualification image", 1153 ] 1154 ], 1155 } 1156 1157 self.assertIs( 1158 fixture.classify_attempt(event, {"P01-A02": attempt}), attempt 1159 ) 1160 for mutation in ( 1161 lambda value: value.update({"content": marker}), 1162 lambda value: value.update({"content": f"prefix {marker}\n{url}"}), 1163 lambda value: value.update( 1164 {"content": f"{marker}\nhttps://example.com/{digest}.png"} 1165 ), 1166 lambda value: value["tags"][0].remove(f"x {digest}"), 1167 ): 1168 changed = copy.deepcopy(event) 1169 mutation(changed) 1170 self.assertIsNone( 1171 fixture.classify_attempt(changed, {"P01-A02": attempt}) 1172 ) 1173 1174 text_attempt = {"flow": "Update", "marker": "rr-p01-a01-update"} 1175 embedded = {"content": "prefix rr-p01-a01-update", "tags": []} 1176 self.assertIsNone( 1177 fixture.classify_attempt(embedded, {"P01-A01": text_attempt}) 1178 ) 1179 1180 1181 if __name__ == "__main__": 1182 unittest.main()