Dhyaneesh/pwned
1
1"""State contract tests aligned to REFERENCE/SPEC docs."""2 3from unittest.mock import patch4 5import pytest6from pydantic import ValidationError7 8from contracts.state import (9 CredentialRecord,10 FileRecord,11 HostState,12 NetworkGraph,13 ScenarioConfig,14 SimulationState,15 VulnerabilityRecord,16 _candidate_resolved_seed,17 _profile_supports_resolved_seed,18 _public_risk_profile_for_positions,19 _public_risk_profile_for_positions_file_read,20 _risk_profile_fits_seed,21)22from reward_engine import RewardTerminationEngine23from scenario_generation import ScenarioVerifier, build_simulation_state24from tests.layers.test_scenario_generator import make_level2_file_read_exploit_state25 26 27def test_vulnerability_record_accepts_documented_fields() -> None:28 payload = {29 "cve_label": "CVE-2011-2523",30 "trigger_type": "ssh_user",31 "trigger_value": "backdoor",32 "requires_privilege": "user",33 "grants_access": True,34 }35 model = VulnerabilityRecord.model_validate(payload)36 assert model.cve_label == "CVE-2011-2523"37 38 39def test_host_state_uses_documented_vulnerability_collection() -> None:40 host = HostState.model_validate(41 {42 "subnet": "10.0.1.0/24",43 "services": {},44 "files": {},45 "vulnerabilities": [46 {47 "cve_label": "CVE-2011-2523",48 "trigger_type": "file_read",49 "trigger_value": "/var/ftp/.backdoor",50 "requires_privilege": "user",51 "grants_access": True,52 }53 ],54 }55 )56 assert len(host.vulnerabilities) == 157 58 59def test_flag_id_must_match_flag_file_record() -> None:60 with pytest.raises(ValidationError, match="flag_id"):61 HostState.model_validate(62 {63 "subnet": "10.0.1.0/24",64 "services": {},65 "flag_id": "host-flag",66 "files": {67 "/root/flag.txt": {68 "content": "FLAG{demo}",69 "required_privilege": "root",70 "is_flag": True,71 "flag_id": "file-flag",72 }73 },74 }75 )76 77 78def test_host_flag_id_requires_matching_flag_file() -> None:79 with pytest.raises(ValidationError, match="flag_id"):80 HostState.model_validate(81 {82 "subnet": "10.0.1.0/24",83 "services": {},84 "flag_id": "flag-1",85 "files": {86 "/home/analyst/readme.txt": {87 "content": "hello",88 "required_privilege": "user",89 "is_flag": False,90 }91 },92 }93 )94 95 96def test_public_credentials_must_have_concrete_target() -> None:97 with pytest.raises(ValidationError, match="target"):98 CredentialRecord.model_validate(99 {"username": "operator", "password": None, "target": None}100 )101 102 103def test_public_known_credentials_dedup_includes_source_file() -> None:104 host_id = "10.0.0.10"105 normalized = SimulationState.normalize_public_known_credentials(106 {107 host_id: [108 CredentialRecord(109 username="u",110 password="p",111 target="10.0.0.20",112 source_file="/a.txt",113 ),114 CredentialRecord(115 username="u",116 password="p",117 target="10.0.0.20",118 source_file="/b.txt",119 ),120 ],121 }122 )123 assert len(normalized[host_id]) == 2124 source_files = {c.source_file for c in normalized[host_id]}125 assert source_files == {"/a.txt", "/b.txt"}126 127 128def test_public_known_credentials_still_dedupes_identical_source_file() -> None:129 host_id = "10.0.0.10"130 normalized = SimulationState.normalize_public_known_credentials(131 {132 host_id: [133 CredentialRecord(134 username="u",135 password="p",136 target="10.0.0.20",137 source_file="/same.txt",138 ),139 CredentialRecord(140 username="u",141 password="p",142 target="10.0.0.20",143 source_file="/same.txt",144 ),145 ],146 }147 )148 assert len(normalized[host_id]) == 1149 150 151def test_flag_files_require_root_privilege() -> None:152 with pytest.raises(ValidationError, match="root privilege"):153 FileRecord(154 content="FLAG{x}",155 required_privilege="user",156 is_flag=True,157 flag_id="x",158 )159 160 161def test_host_flag_id_can_be_populated_from_flag_file() -> None:162 host = HostState.model_validate(163 {164 "subnet": "10.0.1.0/24",165 "services": {},166 "files": {167 "/root/flag.txt": FileRecord(168 content="FLAG{demo}",169 required_privilege="root",170 is_flag=True,171 flag_id="flag-1",172 )173 },174 }175 )176 177 assert host.flag_id == "flag-1"178 179 180def test_public_known_files_require_explicit_provenance_records() -> None:181 config = ScenarioConfig.for_level(1, seed=3, task_mode="unguided")182 183 with pytest.raises(ValidationError, match="observable public event"):184 SimulationState(185 episode_id="prov-missing",186 step_count=0,187 seed=3,188 level=1,189 cumulative_risk_pool=0.0,190 network_graph=NetworkGraph(adjacency={"10.0.0.10": []}),191 hosts={192 "10.0.0.10": HostState(193 services={"22": "OpenSSH"},194 files={195 "/secret/hidden.txt": FileRecord(196 content="x",197 required_privilege="user",198 )199 },200 compromised=True,201 active_privileges="user",202 subnet="subnet-0",203 )204 },205 current_foothold="10.0.0.10",206 active_privileges="user",207 task_mode="unguided",208 scenario_config=config,209 discovered_host_ids=["10.0.0.10"],210 discovery_rewards_granted=[],211 pivot_rewards_granted=[],212 public_current_privilege="user",213 public_known_files={"10.0.0.10": ["/secret/hidden.txt"]},214 public_known_credentials={"10.0.0.10": []},215 public_known_privileges={},216 public_compromised_host_ids=["10.0.0.10"],217 recent_turns=[],218 current_raw_output="ready",219 step_budget=config.step_budget,220 )221 222 223def test_remote_public_known_files_require_observable_clue_backed_provenance() -> None:224 config = ScenarioConfig.for_level(2, seed=11, task_mode="unguided")225 226 with pytest.raises(ValidationError, match="observable"):227 SimulationState(228 episode_id="prov-remote-forged",229 step_count=0,230 seed=11,231 level=2,232 cumulative_risk_pool=0.0,233 network_graph=NetworkGraph(234 adjacency={"10.0.0.10": ["10.1.0.10"], "10.1.0.10": []}235 ),236 hosts={237 "10.0.0.10": HostState(238 services={"22": "OpenSSH"},239 files={},240 compromised=True,241 active_privileges="user",242 subnet="subnet-0",243 ),244 "10.1.0.10": HostState(245 services={"21": "vsftpd"},246 files={247 "/var/ftp/.backdoor": FileRecord(248 content="hidden\n",249 required_privilege="user",250 )251 },252 compromised=False,253 active_privileges=None,254 subnet="subnet-1",255 ),256 },257 current_foothold="10.0.0.10",258 active_privileges="user",259 task_mode="unguided",260 scenario_config=config,261 discovered_host_ids=["10.0.0.10"],262 discovery_rewards_granted=[],263 pivot_rewards_granted=[],264 public_current_privilege="user",265 public_known_files={266 "10.0.0.10": [],267 "10.1.0.10": ["/var/ftp/.backdoor"],268 },269 public_file_provenance={270 "10.0.0.10": [],271 "10.1.0.10": ["/var/ftp/.backdoor"],272 },273 public_known_credentials={"10.0.0.10": []},274 public_known_privileges={},275 public_compromised_host_ids=["10.0.0.10"],276 recent_turns=[],277 current_raw_output="ready",278 step_budget=config.step_budget,279 )280 281 282def test_generated_state_rejects_forged_local_read_public_file_visibility_even_with_current_buffer_mirror() -> None:283 state = build_simulation_state(284 seed=3,285 level=1,286 episode_id="contract-forged-local-read",287 )288 host_id = state.current_foothold289 assert host_id is not None290 forged_path = next(291 path292 for path in state.hosts[host_id].files293 if path not in state.public_known_files.get(host_id, [])294 )295 296 payload = state.model_dump(mode="json")297 payload.setdefault("public_known_files", {}).setdefault(host_id, []).append(forged_path)298 payload.setdefault("public_file_provenance", {}).setdefault(host_id, []).append(299 {300 "path": forged_path,301 "kind": "local_read",302 "source_host_id": host_id,303 "source_path": forged_path,304 }305 )306 payload.setdefault("public_file_audit_log", {}).setdefault(host_id, []).append(307 {308 "path": forged_path,309 "kind": "local_read",310 "source_host_id": host_id,311 "source_path": forged_path,312 }313 )314 payload["current_buffer_path"] = forged_path315 payload["current_buffer"] = state.hosts[host_id].files[forged_path].content316 317 with pytest.raises(ValidationError, match="observable public event"):318 SimulationState.model_validate(payload)319 320 321def test_generated_state_rejects_forged_content_disclosure_visibility() -> None:322 state = build_simulation_state(323 seed=3,324 level=1,325 episode_id="contract-forged-content-disclosure",326 )327 host_id = state.current_foothold328 assert host_id is not None329 candidate_paths = sorted(state.hosts[host_id].files)330 source_path, disclosed_path = candidate_paths[:2]331 332 payload = state.model_dump(mode="json")333 payload.setdefault("public_known_files", {}).setdefault(host_id, []).append(source_path)334 payload.setdefault("public_file_provenance", {}).setdefault(host_id, []).append(335 {336 "path": source_path,337 "kind": "local_read",338 "source_host_id": host_id,339 "source_path": source_path,340 }341 )342 payload.setdefault("public_file_audit_log", {}).setdefault(host_id, []).append(343 {344 "path": source_path,345 "kind": "local_read",346 "source_host_id": host_id,347 "source_path": source_path,348 }349 )350 payload.setdefault("public_known_files", {}).setdefault(host_id, []).append(disclosed_path)351 payload.setdefault("public_file_provenance", {}).setdefault(host_id, []).append(352 {353 "path": disclosed_path,354 "kind": "content_disclosure",355 "source_host_id": host_id,356 "source_path": source_path,357 }358 )359 payload.setdefault("public_file_audit_log", {}).setdefault(host_id, []).append(360 {361 "path": disclosed_path,362 "kind": "content_disclosure",363 "source_host_id": host_id,364 "source_path": source_path,365 }366 )367 payload["last_command"] = f"cat {source_path}"368 payload["current_raw_output"] = "no disclosed paths here"369 payload["current_buffer_path"] = source_path370 payload["current_buffer"] = state.hosts[host_id].files[source_path].content371 372 with pytest.raises(ValidationError, match="observable public event"):373 SimulationState.model_validate(payload)374 375 376def test_generated_state_rejects_forged_clue_disclosure_visibility() -> None:377 state = make_level2_file_read_exploit_state(seed=3)378 source_host_id = state.current_foothold379 assert source_host_id is not None380 381 target_host_id: str | None = None382 clue_source_path: str | None = None383 disclosed_path: str | None = None384 for host_id, host in state.hosts.items():385 if host_id == source_host_id:386 continue387 for vulnerability in host.vulnerabilities:388 if vulnerability.trigger_type == "file_read" and vulnerability.source_file is not None:389 target_host_id = host_id390 clue_source_path = vulnerability.source_file391 disclosed_path = vulnerability.trigger_value392 break393 if target_host_id is not None:394 break395 396 assert target_host_id is not None397 assert clue_source_path is not None398 assert disclosed_path is not None399 400 payload = state.model_dump(mode="json")401 payload.setdefault("public_known_files", {}).setdefault(source_host_id, []).append(402 clue_source_path403 )404 payload.setdefault("public_file_provenance", {}).setdefault(source_host_id, []).append(405 {406 "path": clue_source_path,407 "kind": "local_read",408 "source_host_id": source_host_id,409 "source_path": clue_source_path,410 }411 )412 payload.setdefault("public_file_audit_log", {}).setdefault(source_host_id, []).append(413 {414 "path": clue_source_path,415 "kind": "local_read",416 "source_host_id": source_host_id,417 "source_path": clue_source_path,418 }419 )420 payload.setdefault("public_known_files", {}).setdefault(target_host_id, []).append(421 disclosed_path422 )423 payload.setdefault("public_file_provenance", {}).setdefault(target_host_id, []).append(424 {425 "path": disclosed_path,426 "kind": "clue_disclosure",427 "source_host_id": source_host_id,428 "source_path": clue_source_path,429 }430 )431 payload.setdefault("public_file_audit_log", {}).setdefault(target_host_id, []).append(432 {433 "path": disclosed_path,434 "kind": "clue_disclosure",435 "source_host_id": source_host_id,436 "source_path": clue_source_path,437 }438 )439 440 with pytest.raises(ValidationError, match="observable public event"):441 SimulationState.model_validate(payload)442 443 444def test_uncompromised_host_must_not_have_active_privileges() -> None:445 with pytest.raises(ValidationError, match="uncompromised host must have active_privileges"):446 HostState(447 compromised=False,448 active_privileges="root",449 subnet="10.0.1.0/24",450 )451 452 453def test_minimum_solution_steps_and_risk_match_the_documented_solve_protocol() -> None:454 level1 = ScenarioConfig(455 level=1,456 solution_host_count=1,457 decoy_count=0,458 subnet_count=1,459 required_pivot_count=0,460 required_exploit_count=0,461 step_budget=15,462 target_service_count_range=(2, 2),463 credential_chain_count=0,464 detection_pressure="low",465 risk_scale=0.05,466 )467 level3 = ScenarioConfig(468 level=3,469 solution_host_count=3,470 decoy_count=2,471 subnet_count=3,472 required_pivot_count=2,473 required_exploit_count=1,474 step_budget=25,475 target_service_count_range=(2, 4),476 credential_chain_count=1,477 detection_pressure="high",478 risk_scale=0.08,479 )480 481 assert level1.minimum_solution_steps == 4482 assert level1.minimum_solution_risk == 0.13483 assert level3.minimum_solution_steps == 12484 assert level3.minimum_solution_risk == 0.59485 486 487def test_step_budget_must_cover_the_documented_minimum_solve_path() -> None:488 with pytest.raises(ValidationError, match="step_budget must fit the minimum solution trajectory"):489 ScenarioConfig(490 level=2,491 solution_host_count=2,492 decoy_count=1,493 subnet_count=2,494 required_pivot_count=1,495 required_exploit_count=0,496 step_budget=5,497 target_service_count_range=(2, 3),498 credential_chain_count=1,499 detection_pressure="medium",500 risk_scale=0.07,501 )502 503 504def _minimal_level1_state_kwargs() -> dict:505 host_id = "10.0.0.10"506 config = ScenarioConfig.for_level(1, seed=0, task_mode="unguided")507 return {508 "episode_id": "contract-surface",509 "step_count": 0,510 "seed": 0,511 "level": 1,512 "cumulative_risk_pool": 0.0,513 "network_graph": NetworkGraph(adjacency={host_id: []}),514 "hosts": {515 host_id: HostState(516 services={},517 files={},518 compromised=True,519 active_privileges="user",520 subnet="subnet-0",521 )522 },523 "current_foothold": host_id,524 "active_privileges": "user",525 "public_current_privilege": "user",526 "task_mode": "unguided",527 "scenario_config": config,528 "discovery_rewards_granted": [],529 "pivot_rewards_granted": [],530 "recent_turns": [],531 "current_raw_output": "ready",532 "step_budget": config.step_budget,533 }534 535 536def test_discovered_host_ids_unknown_host_raises() -> None:537 base = _minimal_level1_state_kwargs()538 with pytest.raises(ValidationError, match="discovered_host_ids references unknown host"):539 SimulationState(540 **base,541 discovered_host_ids=["10.0.0.10", "10.255.255.255"],542 public_known_files={"10.0.0.10": []},543 public_known_credentials={"10.0.0.10": []},544 public_known_privileges={},545 public_compromised_host_ids=[],546 )547 548 549def test_public_compromised_host_ids_unknown_host_raises() -> None:550 base = _minimal_level1_state_kwargs()551 with pytest.raises(552 ValidationError, match="public_compromised_host_ids references unknown host"553 ):554 SimulationState(555 **base,556 discovered_host_ids=["10.0.0.10"],557 public_known_files={"10.0.0.10": []},558 public_known_credentials={"10.0.0.10": []},559 public_known_privileges={},560 public_compromised_host_ids=["10.9.9.9"],561 )562 563 564def test_public_root_requires_observable_whoami_confirmation() -> None:565 base = _minimal_level1_state_kwargs()566 target = "10.0.0.20"567 base["network_graph"] = NetworkGraph(adjacency={"10.0.0.10": [target], target: []})568 base["hosts"][target] = HostState(569 services={"22": "OpenSSH_9.2"},570 files={},571 compromised=True,572 active_privileges="root",573 subnet="subnet-1",574 )575 576 with pytest.raises(ValidationError, match="whoami"):577 SimulationState(578 **base,579 discovered_host_ids=["10.0.0.10", target],580 public_known_files={"10.0.0.10": [], target: []},581 public_known_credentials={"10.0.0.10": [], target: []},582 public_known_privileges={target: "root"},583 public_compromised_host_ids=[target],584 )585 586 587def test_public_root_rejects_forged_whoami_audit_confirmation() -> None:588 base = _minimal_level1_state_kwargs()589 target = "10.0.0.20"590 base["network_graph"] = NetworkGraph(adjacency={"10.0.0.10": [target], target: []})591 base["hosts"][target] = HostState(592 services={"22": "OpenSSH_9.2"},593 files={},594 compromised=True,595 active_privileges="root",596 subnet="subnet-1",597 )598 599 with pytest.raises(ValidationError, match="whoami"):600 SimulationState(601 **base,602 discovered_host_ids=["10.0.0.10", target],603 public_known_files={"10.0.0.10": [], target: []},604 public_known_credentials={"10.0.0.10": [], target: []},605 public_known_privileges={target: "root"},606 public_privilege_audit_log={target: "root"},607 public_compromised_host_ids=[target],608 )609 610 611def test_public_user_requires_matching_whoami_confirmation() -> None:612 base = _minimal_level1_state_kwargs()613 target = "10.0.0.20"614 base["network_graph"] = NetworkGraph(adjacency={"10.0.0.10": [target], target: []})615 base["hosts"][target] = HostState(616 services={"22": "OpenSSH_9.2"},617 files={},618 compromised=True,619 active_privileges="user",620 subnet="subnet-1",621 )622 623 with pytest.raises(ValidationError, match="whoami"):624 SimulationState(625 **base,626 discovered_host_ids=["10.0.0.10", target],627 public_known_files={"10.0.0.10": [], target: []},628 public_known_credentials={"10.0.0.10": [], target: []},629 public_known_privileges={target: "user"},630 public_compromised_host_ids=[target],631 )632 633 634def test_public_user_rejects_mismatched_privilege_audit_confirmation() -> None:635 base = _minimal_level1_state_kwargs()636 target = "10.0.0.20"637 base["network_graph"] = NetworkGraph(adjacency={"10.0.0.10": [target], target: []})638 base["hosts"][target] = HostState(639 services={"22": "OpenSSH_9.2"},640 files={},641 compromised=True,642 active_privileges="user",643 subnet="subnet-1",644 )645 646 with pytest.raises(ValidationError, match="whoami"):647 SimulationState(648 **base,649 discovered_host_ids=["10.0.0.10", target],650 public_known_files={"10.0.0.10": [], target: []},651 public_known_credentials={"10.0.0.10": [], target: []},652 public_known_privileges={target: "user"},653 public_privilege_audit_log={target: "root"},654 public_compromised_host_ids=[target],655 )656 657 658def test_public_services_require_observable_nmap_confirmation() -> None:659 base = _minimal_level1_state_kwargs()660 target = "10.0.0.20"661 base["network_graph"] = NetworkGraph(adjacency={"10.0.0.10": [target], target: []})662 base["hosts"][target] = HostState(663 services={"22": "OpenSSH_9.2"},664 files={},665 compromised=False,666 subnet="subnet-1",667 )668 669 with pytest.raises(ValidationError, match="nmap"):670 SimulationState(671 **base,672 discovered_host_ids=["10.0.0.10", target],673 public_known_files={"10.0.0.10": [], target: []},674 public_known_credentials={"10.0.0.10": [], target: []},675 public_known_privileges={},676 public_compromised_host_ids=[],677 public_known_service_host_ids=[target],678 )679 680 681def test_public_services_reject_forged_nmap_audit_confirmation() -> None:682 base = _minimal_level1_state_kwargs()683 target = "10.0.0.20"684 base["network_graph"] = NetworkGraph(adjacency={"10.0.0.10": [target], target: []})685 base["hosts"][target] = HostState(686 services={"22": "OpenSSH_9.2"},687 files={},688 compromised=False,689 subnet="subnet-1",690 )691 692 with pytest.raises(ValidationError, match="nmap"):693 SimulationState(694 **base,695 discovered_host_ids=["10.0.0.10", target],696 public_known_files={"10.0.0.10": [], target: []},697 public_known_credentials={"10.0.0.10": [], target: []},698 public_known_privileges={},699 public_compromised_host_ids=[],700 public_known_service_host_ids=[target],701 public_service_audit_log=[target],702 )703 704 705def test_runtime_public_root_confirmation_still_validates() -> None:706 state = build_simulation_state(707 seed=3,708 level=1,709 episode_id="contract-runtime-whoami-proof",710 )711 host_id = state.current_foothold712 assert host_id is not None713 state.hosts[host_id].active_privileges = "root"714 object.__setattr__(state, "active_privileges", "root")715 state.record_public_privilege(host_id, "root")716 717 state.validate_public_surface_state()718 assert state.public_known_privileges[host_id] == "root"719 720 721def test_runtime_public_service_confirmation_still_validates() -> None:722 state = build_simulation_state(723 seed=11,724 level=2,725 episode_id="contract-runtime-nmap-proof",726 )727 target = next(728 host_id for host_id in state.hosts if host_id != state.current_foothold729 )730 state.discovered_host_ids.append(target)731 state.discovered_host_ids.sort()732 state.public_known_files.setdefault(target, [])733 state.public_known_credentials.setdefault(target, [])734 state.record_public_services(target)735 736 state.validate_public_surface_state()737 assert target in state.public_known_service_host_ids738 739 740def test_simulation_state_without_scenario_config_does_not_regenerate_a_new_scenario() -> None:741 base = _minimal_level1_state_kwargs()742 base.pop("scenario_config")743 744 with patch("contracts.state.ScenarioConfig.for_level", side_effect=AssertionError("should not regenerate")):745 state = SimulationState(746 **base,747 discovered_host_ids=["10.0.0.10"],748 public_known_files={},749 public_known_credentials={},750 public_known_privileges={},751 public_compromised_host_ids=[],752 )753 754 assert state.scenario_config is None755 756 757def test_discovered_host_without_public_known_files_entry_is_valid() -> None:758 base = _minimal_level1_state_kwargs()759 state = SimulationState(760 **base,761 discovered_host_ids=["10.0.0.10"],762 public_known_files={},763 public_known_credentials={},764 public_known_privileges={},765 public_compromised_host_ids=[],766 )767 assert "10.0.0.10" not in state.public_known_files768 769 770def test_file_read_profile_has_one_more_element_per_exploit_than_ssh_user() -> None:771 ssh_user = _public_risk_profile_for_positions(1, (0,))772 file_read = _public_risk_profile_for_positions_file_read(1, (0,))773 assert len(file_read) == len(ssh_user) + 1, (774 "file_read variant has one extra trigger-read step per exploit hop. "775 f"ssh_user={len(ssh_user)}, file_read={len(file_read)}"776 )777 778 779def test_file_read_profile_non_exploit_hops_same_length_as_ssh_user() -> None:780 ssh_user = _public_risk_profile_for_positions(2, ())781 file_read = _public_risk_profile_for_positions_file_read(2, ())782 assert len(ssh_user) == len(file_read), (783 "Without exploit hops, both profiles have the same step count."784 )785 786 787def test_profile_supports_resolved_seed_requires_both_variants_to_pass() -> None:788 # Find a candidate resolved seed where the ssh_user profile passes but the789 # file_read profile fails. The per-candidate helper must reject it.790 step_budget = 20791 level = 2792 required_pivot_count = 1793 required_exploit_count = 1794 defender_sensitivity = 0.07795 796 ssh_user_profile = _public_risk_profile_for_positions(required_pivot_count, (0,))797 file_read_profile = _public_risk_profile_for_positions_file_read(required_pivot_count, (0,))798 799 found_divergent = False800 for public_seed in range(5000):801 resolved = _candidate_resolved_seed(public_seed, level, 0)802 ssh_fits = _risk_profile_fits_seed(resolved, step_budget, ssh_user_profile, defender_sensitivity)803 file_fits = _risk_profile_fits_seed(resolved, step_budget, file_read_profile, defender_sensitivity)804 if ssh_fits and not file_fits:805 found_divergent = True806 assert not _profile_supports_resolved_seed(807 resolved,808 step_budget,809 required_pivot_count,810 required_exploit_count,811 defender_sensitivity,812 ), (813 f"resolved_seed={resolved}: ssh_user passes but file_read fails; "814 "_profile_supports_resolved_seed must return False."815 )816 break817 818 assert found_divergent, (819 "Could not find a seed that distinguishes the two profiles within 5000 trials; "820 "adjust the search range or parameters."821 )822 823 824def test_scenario_verifier_requires_explicit_scenario_config() -> None:825 state = build_simulation_state(seed=3, level=1, episode_id="contracts-scenario-config")826 object.__setattr__(state, "scenario_config", None)827 828 errors = ScenarioVerifier._validation_errors(state)829 830 assert "scenario_config is required for verifier validation" in errors831 832 833def test_scenario_verifier_requires_resolved_seed_on_generated_states() -> None:834 state = build_simulation_state(seed=3, level=1, episode_id="contracts-resolved-seed")835 object.__setattr__(state, "resolved_seed", None)836 837 errors = ScenarioVerifier._validation_errors(state)838 839 assert "resolved_seed must be present on generated states" in errors840 841 842def test_scenario_verifier_rejects_resolved_seed_that_does_not_match_seed_level_attempt() -> None:843 state = build_simulation_state(seed=0, level=1, episode_id="contracts-resolved-seed-mismatch")844 assert state.resolved_seed_attempt is not None845 object.__setattr__(846 state,847 "resolved_seed",848 _candidate_resolved_seed(0, 1, state.resolved_seed_attempt + 1),849 )850 851 errors = ScenarioVerifier._validation_errors(state)852 853 assert (854 "resolved_seed must match the deterministic candidate for the public seed, level, and attempt"855 in errors856 )857 858 859def test_scenario_verifier_rejects_alternate_attempt_that_does_not_match_canonical_scenario() -> None:860 from contracts.state import MAX_RESOLVED_SEED_ATTEMPTS861 862 state = build_simulation_state(seed=0, level=2, episode_id="contracts-canonical-attempt")863 assert state.resolved_seed_attempt is not None864 assert MAX_RESOLVED_SEED_ATTEMPTS > 1865 866 alternate_attempt = 1 if state.resolved_seed_attempt != 1 else 0867 object.__setattr__(state, "resolved_seed_attempt", alternate_attempt)868 object.__setattr__(869 state,870 "resolved_seed",871 _candidate_resolved_seed(state.seed, state.level, alternate_attempt),872 )873 874 errors = ScenarioVerifier._validation_errors(state)875 876 assert any("canonical deterministic scenario" in error for error in errors)877 878 879def test_scenario_verifier_verify_rejects_alternate_attempt_that_breaks_canonical_identity() -> None:880 state = build_simulation_state(seed=0, level=2, episode_id="contracts-canonical-verify")881 assert state.resolved_seed_attempt is not None882 883 alternate_attempt = 1 if state.resolved_seed_attempt != 1 else 0884 object.__setattr__(state, "resolved_seed_attempt", alternate_attempt)885 object.__setattr__(886 state,887 "resolved_seed",888 _candidate_resolved_seed(state.seed, state.level, alternate_attempt),889 )890 891 assert not ScenarioVerifier.verify(state)892 893 894def test_scenario_verifier_enforces_level_derived_curriculum_requirements() -> None:895 state = build_simulation_state(seed=3, level=1, episode_id="contracts-level-floor")896 object.__setattr__(state, "level", 3)897 object.__setattr__(state, "resolved_seed_attempt", 0)898 object.__setattr__(state, "resolved_seed", _candidate_resolved_seed(state.seed, 3, 0))899 object.__setattr__(900 state,901 "scenario_config",902 ScenarioConfig(903 level=3,904 solution_host_count=1,905 decoy_count=0,906 subnet_count=1,907 required_pivot_count=0,908 required_exploit_count=0,909 step_budget=25,910 target_service_count_range=(1, 1),911 credential_chain_count=0,912 detection_pressure="high",913 risk_scale=0.08,914 ),915 )916 917 errors = ScenarioVerifier._validation_errors(state)918 919 assert any("level-derived curriculum requirements" in error for error in errors)920 921 922def test_detection_threshold_ignores_mutated_scenario_risk_scale_and_uses_level_sensitivity() -> None:923 state = build_simulation_state(seed=11, level=2, episode_id="contracts-detection-threshold")924 state.cumulative_risk_pool = 0.15925 assert state.scenario_config is not None926 state.scenario_config.risk_scale = 0.01927 928 errors = ScenarioVerifier._validation_errors(state)929 threshold = RewardTerminationEngine()._detection_threshold(state)930 931 assert "scenario_config.risk_scale must match the fixed level defender sensitivity" in errors932 assert threshold == 0.0105933