Dhyaneesh/pwned
1
1"""Deterministic phase 4 execution engine for the Pwned environment."""2 3from __future__ import annotations4 5from dataclasses import dataclass, field6import ipaddress7import posixpath8from typing import Literal9 10try:11 from .action_interpreter import InternalAction12 from .contracts import InteractionTurn, SimulationState13 from .public_signal_parsing import extract_absolute_posix_paths14except ImportError:15 from action_interpreter import InternalAction16 from contracts import InteractionTurn, SimulationState17 from public_signal_parsing import extract_absolute_posix_paths18 19 20PRIVILEGE_ORDER = {21 "none": 0,22 "user": 1,23 "root": 2,24}25DEFAULT_WORKING_DIRECTORY = "/home/analyst"26 27 28@dataclass29class ExecutionResult:30 """Internal execution result returned by the simulation engine."""31 32 state: SimulationState33 raw_output: str34 success: bool35 done: bool = False36 terminal_reason: str | None = None37 internal_exploit: bool = False38 discovered_host_ids_added: list[str] = field(default_factory=list)39 discovered_decoy_ids_added: list[str] = field(default_factory=list)40 pivot_host_id: str | None = None41 pivot_source_subnet: str | None = None42 exploit_outcome: Literal["none", "successful", "failed"] = "none"43 action_risk_kind: Literal["scan", "enumerate", "read", "pivot"] | None = None44 risk_events: list[45 Literal[46 "scan",47 "enumerate",48 "read",49 "pivot",50 "failed_exploit",51 "successful_exploit",52 "privesc_attempt_failed",53 "privesc_success",54 "decoy_interaction",55 ]56 ] = field(default_factory=list)57 metadata: dict[str, object] = field(default_factory=dict)58 59 60class SimulationEngine:61 """Execute parsed internal actions against the hidden simulation state."""62 63 def execute(64 self,65 state: SimulationState,66 action: InternalAction,67 ) -> ExecutionResult:68 next_state = state.model_copy(deep=True)69 source_host_id = next_state.current_foothold70 71 if source_host_id is None or source_host_id not in next_state.hosts:72 return ExecutionResult(73 state=next_state,74 raw_output="Execution failed: no active foothold is available.",75 success=False,76 action_risk_kind="enumerate",77 metadata={78 "execution_status": "failure",79 "internal_action": action.kind,80 },81 )82 83 source_host = next_state.hosts[source_host_id]84 if not source_host.compromised:85 return ExecutionResult(86 state=next_state,87 raw_output="Permission denied: no active compromised foothold is available.",88 success=False,89 metadata={90 "execution_status": "failure",91 },92 )93 94 source_host.footprint += 195 if action.verb != "exfiltrate":96 object.__setattr__(next_state, "current_buffer", None)97 object.__setattr__(next_state, "current_buffer_path", None)98 99 if action.verb == "nmap":100 result = self._execute_scan(next_state, action, source_host_id)101 self._record_public_turn(result.state, action.raw_command, result.raw_output)102 return result103 if action.verb == "ls":104 result = self._execute_list(next_state, action, source_host_id)105 self._record_public_turn(result.state, action.raw_command, result.raw_output)106 return result107 if action.verb == "whoami":108 result = self._execute_whoami(next_state, action)109 self._record_public_turn(result.state, action.raw_command, result.raw_output)110 return result111 if action.verb == "cat":112 result = self._execute_read(next_state, action, source_host_id)113 self._record_public_turn(result.state, action.raw_command, result.raw_output)114 return result115 if action.verb == "ssh":116 result = self._execute_pivot(next_state, action, source_host_id)117 self._record_public_turn(result.state, action.raw_command, result.raw_output)118 return result119 if action.verb == "exfiltrate":120 result = self._execute_exfiltrate(next_state, action, source_host_id)121 self._record_public_turn(result.state, action.raw_command, result.raw_output)122 return result123 124 return ExecutionResult(125 state=next_state,126 raw_output=f"Execution failed: unsupported internal action `{action.verb}`.",127 success=False,128 metadata={129 "execution_status": "failure",130 "internal_action": action.kind,131 },132 )133 134 def _execute_scan(135 self,136 state: SimulationState,137 action: InternalAction,138 source_host_id: str,139 ) -> ExecutionResult:140 target_spec = action.target_host141 assert target_spec is not None142 143 target_host_ids = self._resolve_scan_targets(state, source_host_id, target_spec)144 if not target_host_ids:145 return ExecutionResult(146 state=state,147 raw_output=(148 f"Scan failed: target {target_spec} is unreachable from "149 f"{source_host_id}."150 ),151 success=False,152 action_risk_kind="scan",153 metadata={154 "execution_status": "failure",155 "internal_action": action.kind,156 },157 )158 159 discovered_host_ids_added: list[str] = []160 discovered_decoy_ids_added: list[str] = []161 rendered_targets: list[str] = []162 163 for target_host_id in target_host_ids:164 first_discovery = self._mark_discovered(state, target_host_id)165 self._remember_public_services(state, target_host_id)166 if first_discovery:167 target_host = state.hosts[target_host_id]168 if target_host.decoy:169 discovered_decoy_ids_added.append(target_host_id)170 else:171 discovered_host_ids_added.append(target_host_id)172 173 target_host = state.hosts[target_host_id]174 services = "\n".join(175 f"{port}/tcp open {version}"176 for port, version in sorted(177 target_host.services.items(),178 key=lambda item: int(item[0]),179 )180 )181 rendered_targets.append(182 "\n".join(183 [184 (185 f"Scan discovered host {target_host_id}"186 if first_discovery187 else f"Scan cached for host {target_host_id}"188 ),189 services,190 ]191 )192 )193 194 return ExecutionResult(195 state=state,196 raw_output="\n".join(rendered_targets),197 success=True,198 discovered_host_ids_added=discovered_host_ids_added,199 discovered_decoy_ids_added=discovered_decoy_ids_added,200 action_risk_kind="scan",201 risk_events=["scan"],202 metadata={203 "execution_status": "success",204 "internal_action": action.kind,205 "target_host": target_spec,206 "new_discovery": bool(discovered_host_ids_added),207 },208 )209 210 def _execute_list(211 self,212 state: SimulationState,213 action: InternalAction,214 source_host_id: str,215 ) -> ExecutionResult:216 host = state.hosts[source_host_id]217 target_path = action.target_path or "."218 normalized_dir = self._resolve_directory_path(target_path)219 entries = self._list_directory_entries(host.files, normalized_dir)220 if not entries:221 return ExecutionResult(222 state=state,223 raw_output=f"ls: cannot access '{target_path}': No such file or directory",224 success=False,225 action_risk_kind="enumerate",226 risk_events=["enumerate"],227 metadata={228 "execution_status": "failure",229 "internal_action": action.kind,230 },231 )232 233 for visible_path in self._listed_file_paths(host.files, normalized_dir):234 self._remember_public_file(state, source_host_id, visible_path)235 236 return ExecutionResult(237 state=state,238 raw_output="\n".join(entries),239 success=True,240 action_risk_kind="enumerate",241 risk_events=["enumerate"],242 metadata={243 "execution_status": "success",244 "internal_action": action.kind,245 },246 )247 248 def _execute_whoami(249 self,250 state: SimulationState,251 action: InternalAction,252 ) -> ExecutionResult:253 if state.current_foothold is not None and state.active_privileges != "none":254 state.record_public_privilege(state.current_foothold, state.active_privileges)255 state.public_current_privilege = state.active_privileges256 return ExecutionResult(257 state=state,258 raw_output=state.active_privileges,259 success=True,260 action_risk_kind="enumerate",261 risk_events=["enumerate"],262 metadata={263 "execution_status": "success",264 "internal_action": action.kind,265 },266 )267 268 def _execute_read(269 self,270 state: SimulationState,271 action: InternalAction,272 source_host_id: str,273 ) -> ExecutionResult:274 target_path = action.target_path275 assert target_path is not None276 277 host = state.hosts[source_host_id]278 resolved_path = self._resolve_file_path(host.files, target_path)279 file_record = None if resolved_path is None else host.files.get(resolved_path)280 281 if file_record is not None:282 if resolved_path not in state.public_known_files.get(source_host_id, []):283 return ExecutionResult(284 state=state,285 raw_output=f"cat: {target_path}: No such file or directory",286 success=False,287 action_risk_kind="read",288 metadata={289 "execution_status": "failure",290 "internal_action": action.kind,291 },292 )293 return self._execute_local_read(294 state=state,295 action=action,296 source_host_id=source_host_id,297 resolved_path=resolved_path,298 file_record=file_record,299 )300 301 remote_target_id = self._resolve_remote_read_target(302 state,303 source_host_id,304 target_path,305 )306 attempted_remote_target_id = remote_target_id307 if attempted_remote_target_id is None:308 attempted_remote_target_id = self._infer_remote_read_attempt_target(309 state,310 source_host_id,311 target_path,312 )313 if attempted_remote_target_id is None:314 return ExecutionResult(315 state=state,316 raw_output=f"cat: {target_path}: No such file or directory",317 success=False,318 action_risk_kind="read",319 metadata={320 "execution_status": "failure",321 "internal_action": action.kind,322 },323 )324 325 remote_host = state.hosts[attempted_remote_target_id]326 if remote_host.decoy:327 return ExecutionResult(328 state=state,329 raw_output=f"Permission denied: {target_path}",330 success=False,331 internal_exploit=False,332 exploit_outcome="failed",333 risk_events=["decoy_interaction"],334 metadata={335 "execution_status": "failure",336 "internal_action": action.kind,337 "target_host": attempted_remote_target_id,338 },339 )340 341 vulnerability = next(342 (343 record344 for record in remote_host.vulnerabilities345 if record.trigger_type == "file_read"346 and record.trigger_value == target_path347 and self._privilege_allows(348 state.active_privileges,349 record.requires_privilege,350 )351 ),352 None,353 )354 if vulnerability is None or remote_target_id is None:355 return ExecutionResult(356 state=state,357 raw_output=f"Permission denied: {target_path}",358 success=False,359 internal_exploit=False,360 exploit_outcome="failed",361 risk_events=["failed_exploit"],362 metadata={363 "execution_status": "failure",364 "internal_action": action.kind,365 "target_host": attempted_remote_target_id,366 },367 )368 369 remote_file = remote_host.files.get(target_path)370 remote_host.compromised = True371 remote_host.active_privileges = "user"372 self._remember_public_compromise(state, attempted_remote_target_id)373 self._remember_public_file(state, attempted_remote_target_id, target_path)374 375 return ExecutionResult(376 state=state,377 raw_output=(378 remote_file.content379 if remote_file is not None380 else f"Exploit triggered via {target_path}"381 ),382 success=True,383 internal_exploit=True,384 exploit_outcome="successful",385 risk_events=["successful_exploit"],386 metadata={387 "execution_status": "success",388 "internal_action": action.kind,389 "target_host": attempted_remote_target_id,390 "exploit_trigger": target_path,391 },392 )393 394 def _execute_pivot(395 self,396 state: SimulationState,397 action: InternalAction,398 source_host_id: str,399 ) -> ExecutionResult:400 target_host_id = action.target_host401 target_user = action.target_user402 assert target_host_id is not None403 assert target_user is not None404 405 if target_host_id == source_host_id:406 return ExecutionResult(407 state=state,408 raw_output=(409 f"Access denied: target {target_host_id} is already the "410 "current foothold."411 ),412 success=False,413 action_risk_kind="pivot",414 metadata={415 "execution_status": "failure",416 "internal_action": action.kind,417 },418 )419 420 if target_host_id not in self._reachable_from(state, source_host_id):421 return ExecutionResult(422 state=state,423 raw_output=(424 f"Access denied: target {target_host_id} is unreachable from "425 f"{source_host_id}."426 ),427 success=False,428 action_risk_kind="pivot",429 metadata={430 "execution_status": "failure",431 "internal_action": action.kind,432 },433 )434 435 target_host = state.hosts[target_host_id]436 source_subnet = state.hosts[source_host_id].subnet437 public_credentials = state.public_known_credentials.get(target_host_id, [])438 matching_ssh_vulns = [439 vuln440 for vuln in target_host.vulnerabilities441 if vuln.trigger_type == "ssh_user" and vuln.trigger_value == target_user442 ]443 matching_credentials = [444 credential445 for credential in public_credentials446 if credential.username == target_user447 ]448 matching_ssh_vuln = None449 credential_match = None450 exact_source_pair = next(451 (452 (vuln, credential)453 for vuln in matching_ssh_vulns454 for credential in matching_credentials455 if vuln.source_file is not None456 and credential.source_file == vuln.source_file457 ),458 None,459 )460 if exact_source_pair is not None:461 matching_ssh_vuln, credential_match = exact_source_pair462 else:463 password_credential = next(464 (465 credential466 for credential in matching_credentials467 if credential.password is not None468 ),469 None,470 )471 if password_credential is not None:472 credential_match = password_credential473 else:474 credential_match = next(iter(matching_credentials), None)475 if matching_ssh_vulns:476 matching_ssh_vuln = matching_ssh_vulns[0]477 478 if target_host.decoy:479 return ExecutionResult(480 state=state,481 raw_output="Connection refused",482 success=False,483 internal_exploit=False,484 exploit_outcome="failed",485 risk_events=["decoy_interaction"],486 metadata={487 "execution_status": "failure",488 "internal_action": action.kind,489 "target_host": target_host_id,490 },491 )492 493 def _pivot_success_result(494 *,495 risk_events: list,496 internal_exploit: bool,497 exploit_outcome: Literal["none", "successful", "failed"],498 ) -> ExecutionResult:499 prior_compromised = target_host.compromised500 prior_privilege = target_host.active_privileges501 next_privilege = (502 prior_privilege503 if prior_compromised and prior_privilege is not None504 else "user"505 )506 target_host.compromised = True507 target_host.active_privileges = next_privilege508 self._remember_public_compromise(state, target_host_id)509 state.clear_public_privilege(target_host_id)510 object.__setattr__(state, "current_foothold", target_host_id)511 object.__setattr__(state, "active_privileges", next_privilege)512 state.public_current_privilege = None513 return ExecutionResult(514 state=state,515 raw_output=f"Access granted. Current foothold: {target_host_id}",516 success=True,517 pivot_host_id=target_host_id,518 pivot_source_subnet=source_subnet,519 risk_events=risk_events,520 internal_exploit=internal_exploit,521 exploit_outcome=exploit_outcome,522 metadata={523 "execution_status": "success",524 "internal_action": action.kind,525 "target_host": target_host_id,526 },527 )528 529 if matching_ssh_vuln is not None:530 if credential_match is None:531 pass532 elif not self._privilege_allows(533 state.active_privileges,534 matching_ssh_vuln.requires_privilege,535 ):536 return ExecutionResult(537 state=state,538 raw_output="Permission denied (publickey)",539 success=False,540 internal_exploit=False,541 exploit_outcome="failed",542 risk_events=["failed_exploit"],543 metadata={544 "execution_status": "failure",545 "internal_action": action.kind,546 "target_host": target_host_id,547 },548 )549 elif (550 matching_ssh_vuln.source_file is not None551 and credential_match.source_file != matching_ssh_vuln.source_file552 ):553 return ExecutionResult(554 state=state,555 raw_output="Permission denied (publickey)",556 success=False,557 internal_exploit=False,558 exploit_outcome="failed",559 risk_events=["failed_exploit"],560 metadata={561 "execution_status": "failure",562 "internal_action": action.kind,563 "target_host": target_host_id,564 },565 )566 else:567 exploit_backed = credential_match.password is None568 return _pivot_success_result(569 risk_events=(570 ["successful_exploit", "pivot"]571 if exploit_backed572 else ["pivot"]573 ),574 internal_exploit=exploit_backed,575 exploit_outcome="successful" if exploit_backed else "none",576 )577 578 passwordless_file_read_hint = (579 credential_match is not None580 and credential_match.password is None581 and any(582 vuln.trigger_type == "file_read"583 and vuln.source_file is not None584 and vuln.source_file == credential_match.source_file585 for vuln in target_host.vulnerabilities586 )587 )588 passwordless_ssh_hint = (589 credential_match is not None590 and credential_match.password is None591 and any(592 vuln.trigger_type == "ssh_user"593 and vuln.source_file is not None594 and vuln.source_file == credential_match.source_file595 for vuln in target_host.vulnerabilities596 )597 )598 599 if credential_match is not None and credential_match.password is not None:600 return _pivot_success_result(601 risk_events=["pivot"],602 internal_exploit=False,603 exploit_outcome="none",604 )605 606 if credential_match is not None and target_host.compromised and passwordless_file_read_hint:607 return _pivot_success_result(608 risk_events=["pivot"],609 internal_exploit=False,610 exploit_outcome="none",611 )612 613 if credential_match is not None and not target_host.compromised and passwordless_ssh_hint:614 return ExecutionResult(615 state=state,616 raw_output="Permission denied (publickey)",617 success=False,618 internal_exploit=False,619 exploit_outcome="failed",620 risk_events=["failed_exploit"],621 metadata={622 "execution_status": "failure",623 "internal_action": action.kind,624 "target_host": target_host_id,625 },626 )627 628 if credential_match is not None and not passwordless_file_read_hint:629 return _pivot_success_result(630 risk_events=["pivot"],631 internal_exploit=False,632 exploit_outcome="none",633 )634 635 if credential_match is not None:636 return ExecutionResult(637 state=state,638 raw_output="Permission denied (publickey)",639 success=False,640 action_risk_kind="pivot",641 metadata={642 "execution_status": "failure",643 "internal_action": action.kind,644 "target_host": target_host_id,645 },646 )647 648 if target_host.compromised:649 return ExecutionResult(650 state=state,651 raw_output="Permission denied (publickey)",652 success=False,653 action_risk_kind="pivot",654 metadata={655 "execution_status": "failure",656 "internal_action": action.kind,657 "target_host": target_host_id,658 },659 )660 661 return ExecutionResult(662 state=state,663 raw_output="Connection refused",664 success=False,665 internal_exploit=False,666 exploit_outcome="failed",667 risk_events=["failed_exploit"],668 metadata={669 "execution_status": "failure",670 "internal_action": action.kind,671 "target_host": target_host_id,672 },673 )674 675 def _execute_exfiltrate(676 self,677 state: SimulationState,678 action: InternalAction,679 source_host_id: str,680 ) -> ExecutionResult:681 del source_host_id682 target = action.target_path683 assert target is not None684 685 matching_flag = next(686 (687 (path, file_record)688 for host in state.hosts.values()689 for path, file_record in host.files.items()690 if file_record.is_flag691 and file_record.flag_id is not None692 and target == file_record.flag_id693 ),694 None,695 )696 697 if matching_flag is None:698 return ExecutionResult(699 state=state,700 raw_output="Requested flag is not in the current buffer.",701 success=False,702 action_risk_kind=None,703 metadata={704 "execution_status": "failure",705 "internal_action": action.kind,706 },707 )708 709 flag_path, flag_record = matching_flag710 flag_id = flag_record.flag_id711 if flag_id is None or flag_id not in state.exfiltration_buffer:712 return ExecutionResult(713 state=state,714 raw_output="Requested flag is not in the current buffer.",715 success=False,716 action_risk_kind=None,717 metadata={718 "execution_status": "failure",719 "internal_action": action.kind,720 },721 )722 723 return ExecutionResult(724 state=state,725 raw_output=f"Flag exfiltration successful: {flag_id}",726 success=True,727 done=True,728 terminal_reason="exfiltration",729 action_risk_kind=None,730 metadata={731 "execution_status": "success",732 "internal_action": action.kind,733 "terminal_reason": "exfiltration",734 },735 )736 737 def _reachable_from(self, state: SimulationState, source_host_id: str) -> set[str]:738 return {source_host_id, *state.network_graph.adjacency.get(source_host_id, [])}739 740 def _resolve_scan_targets(741 self,742 state: SimulationState,743 source_host_id: str,744 target_spec: str,745 ) -> list[str]:746 reachable = self._reachable_from(state, source_host_id)747 748 if "/" not in target_spec:749 if target_spec in state.hosts and target_spec in reachable:750 return [target_spec]751 return []752 753 try:754 network = ipaddress.ip_network(target_spec, strict=False)755 except ValueError:756 return []757 758 matched_hosts: list[str] = []759 for host_id in sorted(reachable):760 if host_id not in state.hosts:761 continue762 try:763 if ipaddress.ip_address(host_id) in network:764 matched_hosts.append(host_id)765 except ValueError:766 continue767 768 return matched_hosts769 770 def _mark_discovered(self, state: SimulationState, host_id: str) -> bool:771 if host_id in state.discovered_host_ids:772 return False773 state.discovered_host_ids.append(host_id)774 return True775 776 def _privilege_allows(self, active: str | None, required: str) -> bool:777 active_key = active if active is not None else "none"778 return PRIVILEGE_ORDER.get(active_key, 0) >= PRIVILEGE_ORDER.get(required, 0)779 780 def _remember_public_compromise(self, state: SimulationState, host_id: str) -> None:781 if host_id not in state.public_compromised_host_ids:782 state.public_compromised_host_ids.append(host_id)783 state.public_compromised_host_ids.sort()784 785 def _remember_public_services(self, state: SimulationState, host_id: str) -> None:786 state.record_public_services(host_id)787 788 def _execute_local_read(789 self,790 state: SimulationState,791 source_host_id: str,792 action: InternalAction,793 resolved_path: str,794 file_record,795 ) -> ExecutionResult:796 host = state.hosts[source_host_id]797 if not self._privilege_allows(state.active_privileges, file_record.required_privilege):798 return ExecutionResult(799 state=state,800 raw_output=f"Permission denied: {action.target_path}",801 success=False,802 action_risk_kind="read",803 metadata={804 "execution_status": "failure",805 "internal_action": action.kind,806 },807 )808 809 self._remember_public_file(state, source_host_id, resolved_path)810 object.__setattr__(state, "current_buffer", file_record.content)811 object.__setattr__(state, "current_buffer_path", resolved_path)812 813 revealed_targets, newly_visible_host_ids = self._reveal_credentials_from_source_file(814 state,815 source_host_id,816 resolved_path,817 )818 self._reveal_local_paths_from_content(819 state,820 source_host_id=source_host_id,821 source_path=resolved_path,822 raw_output=file_record.content,823 )824 825 if (826 host.privesc_requirement is not None827 and state.active_privileges == "user"828 and resolved_path == host.privesc_requirement.trigger_path829 ):830 host.active_privileges = "root"831 object.__setattr__(state, "active_privileges", "root")832 state.record_public_privilege(source_host_id, "root")833 return ExecutionResult(834 state=state,835 raw_output=f"{file_record.content}\n[+] You are now root.",836 success=True,837 risk_events=["read", "privesc_success"],838 discovered_host_ids_added=[839 host_id840 for host_id in newly_visible_host_ids841 if not state.hosts[host_id].decoy842 ],843 discovered_decoy_ids_added=[844 host_id845 for host_id in newly_visible_host_ids846 if state.hosts[host_id].decoy847 ],848 metadata={849 "execution_status": "success",850 "internal_action": action.kind,851 "privilege_change": "root",852 },853 )854 855 if file_record.is_flag and file_record.flag_id is not None:856 if file_record.flag_id not in state.exfiltration_buffer:857 state.exfiltration_buffer.append(file_record.flag_id)858 859 return ExecutionResult(860 state=state,861 raw_output=file_record.content,862 success=True,863 risk_events=["read"],864 discovered_host_ids_added=[865 host_id866 for host_id in newly_visible_host_ids867 if not state.hosts[host_id].decoy868 ],869 discovered_decoy_ids_added=[870 host_id871 for host_id in newly_visible_host_ids872 if state.hosts[host_id].decoy873 ],874 metadata={875 "execution_status": "success",876 "internal_action": action.kind,877 **({} if not file_record.is_flag else {"flag_id": file_record.flag_id}),878 **(879 {}880 if not revealed_targets881 else {"revealed_credential_targets": revealed_targets}882 ),883 },884 )885 886 def _reveal_local_paths_from_content(887 self,888 state: SimulationState,889 *,890 source_host_id: str,891 source_path: str,892 raw_output: str,893 ) -> None:894 host = state.hosts[source_host_id]895 for disclosed_path in extract_absolute_posix_paths(raw_output):896 if disclosed_path == source_path:897 continue898 if disclosed_path not in host.files:899 continue900 state.record_public_file(901 source_host_id,902 disclosed_path,903 kind="content_disclosure",904 source_host_id=source_host_id,905 source_path=source_path,906 )907 908 def _resolve_remote_read_target(909 self,910 state: SimulationState,911 source_host_id: str,912 target_path: str,913 ) -> str | None:914 candidates = self._remote_read_candidates(state, source_host_id)915 visible_candidates = [916 host_id917 for host_id in candidates918 if target_path in state.public_known_files.get(host_id, [])919 ]920 if len(visible_candidates) == 1:921 return visible_candidates[0]922 return None923 924 def _infer_remote_read_attempt_target(925 self,926 state: SimulationState,927 source_host_id: str,928 target_path: str,929 ) -> str | None:930 candidates = self._remote_read_candidates(state, source_host_id)931 visible_candidates = [932 host_id933 for host_id in candidates934 if target_path in state.public_known_files.get(host_id, [])935 ]936 if len(visible_candidates) == 1:937 return visible_candidates[0]938 if len(visible_candidates) > 1:939 return None940 public_candidates = [941 host_id942 for host_id in candidates943 if self._host_is_publicly_visible(state, host_id)944 ]945 if target_path.startswith("/") and len(public_candidates) == 1:946 return public_candidates[0]947 return None948 949 def _remote_read_candidates(950 self,951 state: SimulationState,952 source_host_id: str,953 ) -> list[str]:954 reachable = self._reachable_from(state, source_host_id)955 return [956 host_id957 for host_id in sorted(state.hosts)958 if host_id != source_host_id959 and host_id in reachable960 and not state.hosts[host_id].compromised961 ]962 963 def _reveal_credentials_from_source_file(964 self,965 state: SimulationState,966 source_host_id: str,967 resolved_path: str,968 ) -> tuple[list[str], list[str]]:969 source_host = state.hosts[source_host_id]970 revealed_targets: list[str] = []971 newly_visible_host_ids: list[str] = []972 for credential in source_host.credentials:973 if credential.source_file != resolved_path:974 continue975 was_visible = self._host_is_publicly_visible(state, credential.target)976 credential.exposed = True977 state.public_known_credentials.setdefault(credential.target, [])978 self._remember_public_credential(state, credential)979 if not was_visible and self._host_is_publicly_visible(state, credential.target):980 newly_visible_host_ids.append(credential.target)981 revealed_targets.append(credential.target)982 for target_host_id, target_host in state.hosts.items():983 for vulnerability in target_host.vulnerabilities:984 if (985 vulnerability.trigger_type == "file_read"986 and vulnerability.source_file == resolved_path987 ):988 was_visible = self._host_is_publicly_visible(state, target_host_id)989 self._remember_public_file(990 state,991 target_host_id,992 vulnerability.trigger_value,993 kind="clue_disclosure",994 source_host_id=source_host_id,995 source_path=resolved_path,996 )997 if not was_visible and self._host_is_publicly_visible(state, target_host_id):998 newly_visible_host_ids.append(target_host_id)999 revealed_targets.append(target_host_id)1000 return sorted(set(revealed_targets)), sorted(set(newly_visible_host_ids))1001 1002 def _remember_public_file(1003 self,1004 state: SimulationState,1005 host_id: str,1006 path: str,1007 *,1008 kind: Literal["local_read", "clue_disclosure"] = "local_read",1009 source_host_id: str | None = None,1010 source_path: str | None = None,1011 ) -> None:1012 state.record_public_file(1013 host_id,1014 path,1015 kind=kind,1016 source_host_id=source_host_id,1017 source_path=source_path,1018 )1019 1020 def _remember_public_credential(1021 self,1022 state: SimulationState,1023 credential,1024 ) -> None:1025 visible = state.public_known_credentials.setdefault(credential.target, [])1026 if any(1027 existing.username == credential.username1028 and existing.password == credential.password1029 and existing.target == credential.target1030 and existing.source_file == credential.source_file1031 for existing in visible1032 ):1033 return1034 visible.append(credential.model_copy(deep=True))1035 visible.sort(1036 key=lambda record: (1037 record.target or "",1038 record.username,1039 record.password or "",1040 record.source_file or "",1041 )1042 )1043 1044 def _host_is_publicly_visible(self, state: SimulationState, host_id: str | None) -> bool:1045 if host_id is None:1046 return False1047 return host_id in {1048 *state.discovered_host_ids,1049 *state.public_known_files.keys(),1050 *state.public_known_credentials.keys(),1051 *state.public_compromised_host_ids,1052 *state.public_known_privileges.keys(),1053 }1054 1055 def _resolve_directory_path(self, target_path: str) -> str:1056 if target_path in {".", "./"}:1057 return DEFAULT_WORKING_DIRECTORY1058 if target_path.startswith("/"):1059 return target_path.rstrip("/") or "/"1060 return f"{DEFAULT_WORKING_DIRECTORY.rstrip('/')}/{target_path}".rstrip("/")1061 1062 def _listed_file_paths(1063 self,1064 files: dict[str, object],1065 directory_path: str,1066 ) -> list[str]:1067 normalized = directory_path.rstrip("/") or "/"1068 listed_paths: list[str] = []1069 for path in sorted(files):1070 if normalized == "/":1071 if path.count("/") == 1:1072 listed_paths.append(path)1073 continue1074 prefix = f"{normalized}/"1075 if not path.startswith(prefix):1076 continue1077 relative = path[len(prefix):]1078 if relative and "/" not in relative:1079 listed_paths.append(path)1080 return listed_paths1081 1082 def _list_directory_entries(1083 self,1084 files: dict[str, object],1085 directory_path: str,1086 ) -> list[str]:1087 normalized = directory_path.rstrip("/") or "/"1088 entries: set[str] = set()1089 1090 for path in sorted(files):1091 if path == normalized:1092 entries.add(path.split("/")[-1])1093 continue1094 1095 if normalized == "/":1096 if not path.startswith("/"):1097 continue1098 relative = path[1:]1099 else:1100 prefix = f"{normalized}/"1101 if not path.startswith(prefix):1102 continue1103 relative = path[len(prefix) :]1104 1105 if not relative:1106 continue1107 entries.add(relative.split("/", maxsplit=1)[0])1108 1109 return sorted(entries)1110 1111 def _resolve_file_path(1112 self,1113 files: dict[str, object],1114 target_path: str,1115 ) -> str | None:1116 candidates: list[str] = []1117 1118 if target_path.startswith("/"):1119 candidates.append(posixpath.normpath(target_path))1120 elif target_path.startswith("./"):1121 candidates.append(1122 posixpath.normpath(1123 f"{DEFAULT_WORKING_DIRECTORY.rstrip('/')}/{target_path[2:]}"1124 )1125 )1126 else:1127 candidates.append(1128 posixpath.normpath(1129 f"{DEFAULT_WORKING_DIRECTORY.rstrip('/')}/{target_path}"1130 )1131 )1132 1133 for candidate in candidates:1134 if candidate in files:1135 return candidate1136 1137 return None1138 1139 def _record_public_turn(1140 self,1141 state: SimulationState,1142 command: str,1143 raw_output: str,1144 ) -> None:1145 if state.last_command is not None:1146 object.__setattr__(1147 state,1148 "recent_turns",1149 (1150 state.recent_turns1151 + [1152 InteractionTurn(1153 command=state.last_command,1154 raw_output=state.current_raw_output,1155 step=max(0, state.step_count - 1),1156 )1157 ]1158 )[-3:],1159 )1160 object.__setattr__(state, "last_command", command)1161 object.__setattr__(state, "current_raw_output", raw_output)1162 state.validate_public_surface_state()1163 1164 def _privilege_allows(self, active_privileges: str, required_privilege: str) -> bool:1165 return PRIVILEGE_ORDER.get(active_privileges, 0) >= PRIVILEGE_ORDER.get(1166 required_privilege,1167 0,1168 )1169