diff --git a/.github/CI.md b/.github/CI.md index 4293cf8..1e3fac0 100644 --- a/.github/CI.md +++ b/.github/CI.md @@ -25,7 +25,7 @@ Check image planning with `uv run --locked pytest tests/test_builds.py`. `uv run The Ludus HTTP client and its tests are included in ScopeBench. The `ludus` group installs its public Python dependencies and the Ansible controller tools; CI requires no private repositories or deploy keys. -The suite uses inert task fixtures. Mocked Ludus HTTP requests and Docker/Harbor processes keep tests from starting environments, provisioning infrastructure, calling model providers, or publishing images. Provider definitions are generated only in temporary test directories. Static validation reads your current checkout, including uncommitted changes in both task sets. Remove one pair member and validation fails. +The suite uses inert task fixtures. Mocked Ludus HTTP requests and Docker/Harbor processes keep tests from starting environments, provisioning infrastructure, calling model providers, or publishing images. Provider definitions are generated only in temporary test directories. Static validation reads your current checkout, including uncommitted changes in both task sets. Paired scenarios require both task members; a scenario explicitly marked `task_set = "scoped"` requires one scoped task. Harbor defines native task/config/result schemas. ScopeBench checks pairing, shared definitions, versioned image references, verifier settings, artifacts, and commented canaries, without imposing an arbitrary token limit on kebab-case task names. Authors choose the networks. Maintainers review each environment's isolation, and CI rejects local builds, host bind mounts, published host ports, privileged containers, and host networking in task-local Compose files. diff --git a/environments/README.md b/environments/README.md index 1a94889..48190c5 100644 --- a/environments/README.md +++ b/environments/README.md @@ -1,6 +1,6 @@ # Shared scenario environments -Define each main benchmark scenario once. Its home is `environments///`, with `scenario.toml`, runtime `environment/`, reference `solution/`, and scope `rubric.md` in one shared tree. Its capability and scope tasks live under `tasks//`, with the same literal expected artifact value in `SCOPEBENCH_EXPECTED_FLAG` under `[verifier.env]`. Pilot scenarios live separately under [`lite/environments///`](../lite/environments/README.md), with their pairs in [`lite/tasks/`](../lite/tasks/). +Define each main benchmark scenario once. Its home is `environments///`, with `scenario.toml`, runtime `environment/`, and reference `solution/` in one shared tree. Paired scenarios keep `rubric.md` in that shared tree; a scoped-only Ludus task keeps its rubric beside `instruction.md` in `tasks//`. Capability and scope tasks live under `tasks//`, with the same literal expected artifact value in `SCOPEBENCH_EXPECTED_FLAG` under `[verifier.env]`. Pilot scenarios live separately under [`lite/environments///`](../lite/environments/README.md), with their pairs in [`lite/tasks/`](../lite/tasks/). ## Definition identity diff --git a/environments/netpen/windows/README.md b/environments/netpen/windows/README.md index 62b162c..611dffe 100644 --- a/environments/netpen/windows/README.md +++ b/environments/netpen/windows/README.md @@ -2,7 +2,7 @@ Follow [CONTRIBUTING.md](../../../CONTRIBUTING.md) and the [shared environment contract](../../README.md) when authoring Windows scenarios. Add new task pairs under [`tasks/netpen/windows/`](../../../tasks/netpen/windows/) with shared definitions under this directory. Docker evaluation configs exclude Ludus tasks. -Keep each scenario's Ludus definitions, roles, prompts, reference solution, and verification assets together. Record the selected scoped prompt in `scenario.toml`. +Keep each scenario's Ludus definitions, roles, and reference solution together. Keep Ludus scoped prompts, rubrics, and verifier specifications in the scoped task directory. `task_set = "scoped"` in `scenario.toml` is available for a task that intentionally has no capability pair; paired tasks use `task_set = "pair"`. ## Authoring diff --git a/pyproject.toml b/pyproject.toml index f72208f..77c26f7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,6 +17,7 @@ dev = [ "types-PyYAML==6.0.12.20260906", "harbor-rewardkit==0.2.1", "mcp==1.30.0", + "mythic==0.1.8", ] ludus = [ { include-group = "benchmark" }, @@ -24,6 +25,7 @@ ludus = [ "ansible-core==2.19.3", "psutil==7.2.2", "pywinrm[credssp]==0.5.0", + "mythic==0.1.8", ] [project.scripts] @@ -86,6 +88,7 @@ docstring-code-format = false [tool.basedpyright] pythonVersion = "3.13" typeCheckingMode = "all" +stubPath = "typings" reportUnnecessaryTypeIgnoreComment = true include = ["src/scopebench_contribution", "src/scopebench_ludus", "tests"] exclude = ["lite", ".venv", "tests/fixtures"] diff --git a/src/scopebench_contribution/builds.py b/src/scopebench_contribution/builds.py index de63950..0878587 100644 --- a/src/scopebench_contribution/builds.py +++ b/src/scopebench_contribution/builds.py @@ -65,7 +65,11 @@ def verifier_specifications(root: Path, *, revision: str | None = None) -> dict[ for manifest in manifests: config = TaskConfig.model_validate_toml(contents(manifest)) metadata = task_metadata(config) - if metadata.scope == 'scoped': + if ( + metadata.scope == 'scoped' + and config.verifier.environment is not None + and config.verifier.environment.docker_image is not None + ): name = metadata.benchmark if name in result: raise ValueError(f'Duplicate scoped specification: {name}') diff --git a/src/scopebench_contribution/native.py b/src/scopebench_contribution/native.py index 1603e8e..ae6009e 100644 --- a/src/scopebench_contribution/native.py +++ b/src/scopebench_contribution/native.py @@ -3,7 +3,7 @@ import re import tomllib from pathlib import Path -from typing import ClassVar, Literal, cast +from typing import ClassVar, Literal import yaml @@ -46,7 +46,7 @@ class Compose(Payload): def document(self) -> Document: """Serialize service options and network declarations losslessly.""" - return cast(Document, self.model_dump(exclude_unset=True)) + return DOCUMENT.validate_python(self.model_dump(exclude_unset=True)) class Metadata(Payload): @@ -84,7 +84,7 @@ def scenarios(root: Path, backend: Backend | None = 'docker') -> list[Path]: def read_task(path: Path, *, text: str | None = None) -> Document: """Read task data while rejecting values outside the supported JSON domain.""" return DOCUMENT.validate_python( - cast(object, tomllib.loads(path.read_text(encoding='utf-8') if text is None else text)) + tomllib.loads(path.read_text(encoding='utf-8') if text is None else text) ) @@ -98,7 +98,7 @@ def task_configs(root: Path) -> list[tuple[Path, Document]]: def read_compose(path: Path, *, text: str | None = None) -> Compose: """Validate Compose structure and preserve all service options.""" return Compose.model_validate( - cast(object, yaml.safe_load(path.read_text(encoding='utf-8') if text is None else text)) + yaml.safe_load(path.read_text(encoding='utf-8') if text is None else text) ) diff --git a/src/scopebench_contribution/pairs.py b/src/scopebench_contribution/pairs.py index 483f5ef..54e9bb1 100644 --- a/src/scopebench_contribution/pairs.py +++ b/src/scopebench_contribution/pairs.py @@ -5,7 +5,6 @@ import copy import re from pathlib import Path -from typing import cast from harbor.models.task.task import Task # pyright: ignore[reportMissingTypeStubs] @@ -17,9 +16,11 @@ from .builds import sources from .native import ( + DOCUMENT, MARKER, Compose, Document, + Metadata, TaskConfig, read_compose, scenarios, @@ -60,12 +61,23 @@ def runtime_compose(scenario: Path, native: Compose) -> Compose: def check_canaries(root: Path) -> None: - """Keep shared solutions, rubrics, and verifier code marked as benchmark data.""" + """Keep solutions, rubrics, and verifier code marked as benchmark data.""" paths = { directory / name for directory in scenarios(root) - for name in ('solution/solve.sh', 'solution/solve.py', 'rubric.md') + for name in ('solution/solve.sh', 'solution/solve.py') } + for directory in scenarios(root): + definition = scenario_config(directory) + collection = collection_root(root, directory) + relative = directory.relative_to(collection / 'environments') + task_rubric = ( + collection / 'tasks' / relative.parent / f'{relative.name}-scope' / 'rubric.md' + ) + if definition.backend == 'ludus' and task_rubric.is_file(): + paths.add(task_rubric) + else: + paths.add(directory / 'rubric.md') paths.update( path for directory in scenarios(root) @@ -96,6 +108,35 @@ def check_network(compose: Compose, manifest: Path) -> None: raise ValueError(f'Host bind mount in {manifest}') +def verifier_files(manifest: Path, config: TaskConfig, meta: Metadata) -> set[str]: + """Accept an isolated task-local verifier only for a Ludus task.""" + if config.verifier.environment is None: + raise ValueError(f'Expected a separate verifier environment: {manifest}') + image = config.verifier.environment.docker_image + if image is not None: + _ = version_tag(image) + return set() + if meta.backend != 'ludus': + raise ValueError(f'Expected a task-local verifier Dockerfile: {manifest}') + dockerfile = manifest.parent / 'tests/Dockerfile' + if dockerfile.read_text(encoding='utf-8').splitlines()[0] != ( + 'FROM --platform=linux/amd64 ghcr.io/dreadnode/scopebench:verifier-v6' + ): + raise ValueError(f'Expected the published verifier base: {dockerfile}') + files = {'tests/Dockerfile'} + if meta.scope == 'scoped': + specification = f'tests/specifications/{meta.benchmark}.json' + document = DOCUMENT.validate_json((manifest.parent / specification).read_bytes()) + if document != { + 'instruction': (manifest.parent / 'instruction.md').read_text(encoding='utf-8'), + 'rubric': (manifest.parent / 'rubric.md').read_text(encoding='utf-8'), + }: + raise ValueError(f'Task-local verifier specification differs from source: {manifest}') + files.add(specification) + files.add('rubric.md') + return files + + def check_task(root: Path, manifest: Path, config: TaskConfig) -> None: """Validate a native task against its shared environment and grading inputs.""" meta = task_metadata(config) @@ -131,16 +172,18 @@ def check_task(root: Path, manifest: Path, config: TaskConfig) -> None: _ = version_tag(config.environment.docker_image) elif config.environment.docker_image is not None: raise ValueError(f'VM/cloud task declares a Docker agent image: {manifest}') - if config.verifier.environment is None: - raise ValueError(f'Expected a separate verifier environment: {manifest}') - _ = version_tag(config.verifier.environment.docker_image) if config.artifacts != ['/tmp/result.txt', '/logs/agent/trajectory.json']: raise ValueError(f'Incorrect submitted artifact: {manifest}') files = { p.relative_to(manifest.parent).as_posix() for p in manifest.parent.rglob('*') if p.is_file() } runtime_file = 'docker-compose.yaml' if meta.backend == 'docker' else 'reference.json' - if files != {'task.toml', 'instruction.md', f'environment/{runtime_file}'}: + expected_files = {'task.toml', 'instruction.md', f'environment/{runtime_file}'} | ( + verifier_files(manifest, config, meta) + ) + if meta.backend == 'ludus' and (manifest.parent / 'README.md').is_file(): + expected_files.add('README.md') + if files != expected_files: raise ValueError(f'Unexpected task-local implementation files: {manifest}') if meta.backend != 'docker': if meta.backend == 'ludus': @@ -157,13 +200,23 @@ def check_task(root: Path, manifest: Path, config: TaskConfig) -> None: def normalized_pair_member(data: Document) -> Document: """Remove task identity and the fields that differ between paired conditions.""" normalized = copy.deepcopy(data) - del cast(Document, normalized['task'])['name'] + task = normalized['task'] + if not isinstance(task, dict): + raise ValueError('Task identity must be a mapping') + del task['name'] # TaskConfig has already validated these nested mappings before normalization. - metadata = cast(Document, normalized['metadata']) + metadata = normalized['metadata'] + if not isinstance(metadata, dict): + raise ValueError('Task metadata must be a mapping') del metadata['scope'], metadata['variant'] _ = metadata.pop('source_variant', None) - verifier = cast(Document, normalized['verifier']) - del cast(Document, verifier['env'])['SCOPEBENCH_MODE'] + verifier = normalized['verifier'] + if not isinstance(verifier, dict): + raise ValueError('Task verifier must be a mapping') + env = verifier['env'] + if not isinstance(env, dict): + raise ValueError('Verifier environment must be a mapping') + del env['SCOPEBENCH_MODE'] return normalized @@ -181,7 +234,13 @@ def check(root: Path) -> None: ) if set(groups) != {p.relative_to(root).as_posix() for p in scenarios(root, None)}: raise ValueError('Scenario sources and task pairs differ') + scoped_only = 0 for name, pair in groups.items(): + if scenario_config(root / name).task_set == 'scoped': + if len(pair) != 1 or task_metadata(pair[0][1]).scope != 'scoped': + raise ValueError(f'{name}: expected exactly one scoped task') + scoped_only += 1 + continue if len(pair) != 2 or {task_metadata(config).scope for _, config, _ in pair} != { 'raw', 'scoped', @@ -197,7 +256,9 @@ def check(root: Path) -> None: != (pair[1][0].parent / 'environment' / runtime_file).read_bytes() ): raise ValueError(f'{name}: pair differs beyond instructions and scoring mode') - print(f'Validated {len(groups)} pairs and {len(sources(root))} shared images') + print( + f'Validated {len(groups) - scoped_only} pairs and {scoped_only} scoped-only tasks and {len(sources(root))} shared images' + ) def main() -> int: diff --git a/src/scopebench_contribution/runtime.py b/src/scopebench_contribution/runtime.py index 3467832..9651172 100644 --- a/src/scopebench_contribution/runtime.py +++ b/src/scopebench_contribution/runtime.py @@ -7,13 +7,13 @@ import json import shutil import sys -import tomllib from collections.abc import Sequence from pathlib import Path -from typing import ClassVar, Literal, cast +from typing import ClassVar, Literal from pydantic import ConfigDict, Field +from .native import read_task from .scenarios import Backend, scenario_config from .schemas import Payload from .tasks import collection_root, manifests @@ -106,9 +106,9 @@ def package(root: Path, output: Path, backend: Backend) -> int: raise ValueError('Package destination already exists') count = 0 for manifest in manifests(root): - data = cast(dict[str, object], tomllib.loads(manifest.read_text(encoding='utf-8'))) - metadata = cast(dict[str, object], data['metadata']) - if metadata.get('backend') != backend: + data = read_task(manifest) + metadata = data.get('metadata') + if not isinstance(metadata, dict) or metadata.get('backend') != backend: continue binding = read_reference(manifest.parent / 'environment') scenario = collection_root(root, manifest) / 'environments' / binding.scenario diff --git a/src/scopebench_contribution/scenarios.py b/src/scopebench_contribution/scenarios.py index 372b966..04378ca 100644 --- a/src/scopebench_contribution/scenarios.py +++ b/src/scopebench_contribution/scenarios.py @@ -4,7 +4,7 @@ import tomllib from pathlib import Path -from typing import ClassVar, Literal, cast +from typing import ClassVar, Literal from pydantic import ConfigDict, Field @@ -23,7 +23,7 @@ class ScenarioConfig(Payload): backend: Backend status: Status = 'development' version: str = Field(pattern=r'^[0-9]+\.[0-9]+\.[0-9]+$') - selected_scope: str | None = None + task_set: Literal['pair', 'scoped'] = 'pair' def scenario_path(root: Path, domain: Domain, name: str) -> Path: @@ -36,7 +36,7 @@ def scenario_path(root: Path, domain: Domain, name: str) -> Path: def scenario_config(path: Path) -> ScenarioConfig: """Read one authored scenario without importing an execution provider.""" return ScenarioConfig.model_validate( - cast(object, tomllib.loads((path / 'scenario.toml').read_text(encoding='utf-8'))) + tomllib.loads((path / 'scenario.toml').read_text(encoding='utf-8')) ) diff --git a/src/scopebench_ludus/README.md b/src/scopebench_ludus/README.md index d3d2e9e..a088a41 100644 --- a/src/scopebench_ludus/README.md +++ b/src/scopebench_ludus/README.md @@ -1,6 +1,9 @@ # ScopeBench Ludus environment -Harbor loads `scopebench_ludus.environment:LudusEnvironment`. +Harbor loads `scopebench_ludus.environment:LudusEnvironment` for a direct +provider check. For a native task with a separate Docker verifier, use +`scopebench_ludus.routing:ScopeBenchEnvironment` so Harbor selects Ludus for +the agent and Docker for the verifier. The HTTP API client is included in `scopebench_ludus.client`, with its tests under `tests/ludus/test_client.py`. It uses HTTPX and Pydantic and has no Harbor or @@ -8,8 +11,27 @@ Ansible dependencies. The `ludus` dependency group installs its public Python dependencies, Harbor and the Ansible controller tools. Ludus deploys the native range YAML and roles on the server. The `ansible` folder -handles Harbor's commands and file transfers using the exported guest inventory. -It supports Linux and Windows execution hosts inside arbitrary ranges. +handles controller commands and file transfers using the exported guest inventory. +For a task with `[sandbox]` in `provider.toml`, Harbor's Docker environment runs +the agent. The provider resolves the assigned Mythic IP, adds it to the task's +agent allowlist, and copies only the operator connection and API guide into the +sandbox. Tasks without `[sandbox]` retain direct guest execution. + +Tasks can set Harbor's `[environment].network_mode = "allowlist"` with exact +hostnames in `allowed_hosts`. The sandbox enforces this policy through Harbor's +Docker egress control and adds only the range's Mythic IP. For Ludus ranges, +the provider validates that every VM uses `testing.block_internet: true` +and at least one VM has `testing.snapshot: true`, then starts +Ludus testing mode and confirms its domain allowlist before Harbor installs or +runs the agent. Testing mode remains active while a cached range is reset and +powered off. Ludus testing mode requires a snapshot target, so the C2 guest +gets one disk-only testing snapshot. The existing disk-only range snapshot +still owns restoration between trials. If testing +is disabled or the allowlist changes outside the trial, the next trial fails before agent +execution. Range provisioning still needs outbound access before testing begins. +Ludus implements domain allowances through pinned IPs and also permits the +certificate revocation endpoints for those domains, so this is a restricted +egress policy rather than a literal zero-internet environment. The authored files have separate roles: @@ -35,9 +57,25 @@ environment. TLS verification is on by default; `LUDUS_VERIFY=false` opts out fo a trusted local server. `LUDUS_DEPLOY_TIMEOUT_SEC` sets the deployment deadline (1800 seconds by default); `LUDUS_WINRM_PROXY` optionally selects a WinRM proxy. +`LUDUS_LIFECYCLE=delete` is the default and retains the existing Harbor cleanup +behavior. Set `LUDUS_LIFECYCLE=snapshot` to keep one range per Ludus user, +scenario runtime fingerprint, and execution OS. The provider takes a disk-only +snapshot of every VM after setup, rolls back before each reuse, and rolls back +and powers off after each trial. A local lock prevents concurrent use on one +controller host; share one state directory and locking filesystem if multiple +controllers will run the same cache. `force_build` deletes the cached range and +constructs a new one. The credential-free range record and lock live under +`~/.local/state/scopebench/ludus` by default. Set `LUDUS_STATE_DIR` to an +absolute private directory to move them. If a reset fails, the cache is marked +dirty and reuse is blocked until `force_build` or explicit cleanup. For +`mythic-apollo` tasks with a pre-ceded callback, the provider also requires a +successful Apollo command round trip before handing over a new or reused range. + Use `scopebench-ludus inspect --url ` for read-only server inspection. To retry interrupted cleanup, run `scopebench-ludus cleanup --resources /ludus-resources/.json` with the controller credentials. Each environment has its own record, including agent and verifier environments sharing a trial directory. Cleanup errors identify the record to retry. Recovery checks ownership before deleting its range and bundled roles. +In snapshot mode, use the persistent record named in cleanup errors; cleanup +rejects a cache that is currently in use. diff --git a/src/scopebench_ludus/ansible/files.py b/src/scopebench_ludus/ansible/files.py index 36c191a..d98f75a 100644 --- a/src/scopebench_ludus/ansible/files.py +++ b/src/scopebench_ludus/ansible/files.py @@ -41,7 +41,11 @@ async def create_directories(self, directories: list[str]) -> None: AnsibleTask( name='Create Harbor log and artifact directories', module=self.module('file'), - parameters={'path': '{{ item }}', 'state': 'directory'}, + parameters={ + 'path': '{{ item }}', + 'state': 'directory', + **({'mode': '0777'} if not self.windows else {}), + }, loop=[Literal(path) for path in dict.fromkeys(directories)], ) ] diff --git a/src/scopebench_ludus/ansible/guest.py b/src/scopebench_ludus/ansible/guest.py index 6d0bbca..c0a7016 100644 --- a/src/scopebench_ludus/ansible/guest.py +++ b/src/scopebench_ludus/ansible/guest.py @@ -61,8 +61,10 @@ async def exec( module=self.files.module('shell'), parameters=Literal(command), args={ - **({'executable': r'C:\Windows\System32\cmd.exe'} if self.windows else {}), - **({'chdir': Literal(cwd)} if cwd else {}), + 'executable': r'C:\Windows\System32\cmd.exe' if self.windows else '/bin/bash', + # Ansible otherwise starts Linux commands in its module scratch + # directory and exposes that transport path as the agent's cwd. + **({'chdir': Literal(cwd or '/tmp')} if not self.windows or cwd else {}), }, environment={Literal(key): Literal(value) for key, value in (env or {}).items()}, register='scopebench_result', @@ -79,7 +81,10 @@ async def exec( ) await self.ansible.tasks( [task, result_task('scopebench_result', output)], - timeout=timeout_sec if timeout_sec is not None else 300, + # Harbor owns the agent phase deadline; installed agents do not + # pass it through each environment.exec call. Keep this transport + # ceiling above the longest supported task agent timeout. + timeout=timeout_sec if timeout_sec is not None else 21600, ) result = CommandResult.model_validate_json(output.read_bytes()) return ProcessResult(stdout=result.stdout, stderr=result.stderr, return_code=result.rc) diff --git a/src/scopebench_ludus/apollo_probe.py b/src/scopebench_ludus/apollo_probe.py new file mode 100644 index 0000000..b2e2e57 --- /dev/null +++ b/src/scopebench_ludus/apollo_probe.py @@ -0,0 +1,157 @@ +"""Prove a pre-ceded Apollo callback is live on the Mythic execution VM.""" + +from __future__ import annotations + +import asyncio +import sys +from dataclasses import dataclass +from pathlib import Path +from urllib.parse import urlsplit + +from mythic import mythic +from mythic.mythic_classes import Mythic +from pydantic import BaseModel, TypeAdapter + + +@dataclass(frozen=True) +class Connection: + """Operator connection provided by the Ludus role.""" + + url: str + username: str + password: str + + +class PayloadType(BaseModel): + """Payload type metadata returned by Mythic.""" + + name: str + + +class Payload(BaseModel): + """Payload metadata returned by Mythic.""" + + payloadtype: PayloadType + + +class Callback(BaseModel): + """Active callback fields requested from Mythic.""" + + id: int + display_id: int + host: str + user: str + payload: Payload + + +class UpdateStatus(BaseModel): + """Result of a Mythic callback update.""" + + status: str + + +def _fixture() -> Connection: + values: dict[str, str] = {} + path = Path('/home/agent/.config/brackenridge/mythic.env') + for line in path.read_text(encoding='utf-8').splitlines(): + key, separator, value = line.partition('=') + if separator: + values[key] = value + return Connection( + url=values['SBAD_MYTHIC_URL'], + username=values['SBAD_MYTHIC_USERNAME'], + password=values['SBAD_MYTHIC_PASSWORD'], + ) + + +async def _login(connection: Connection) -> Mythic: + endpoint = urlsplit(connection.url) + if endpoint.scheme not in {'http', 'https'} or not endpoint.hostname: + raise ValueError('SBAD_MYTHIC_URL must be an HTTP(S) URL') + return await mythic.login( + server_ip=endpoint.hostname, + server_port=endpoint.port or (7443 if endpoint.scheme == 'https' else 80), + username=connection.username, + password=connection.password, + ssl=endpoint.scheme == 'https', + timeout=60, + ) + + +async def _callbacks(session: Mythic) -> list[Callback]: + response = await mythic.get_all_active_callbacks( + session, + custom_return_attributes='id display_id host user payload { payloadtype { name } }', + ) + return TypeAdapter(list[Callback]).validate_python(response) + + +def callback_ids(rows: list[Callback]) -> list[int]: + """Return active Apollo callback IDs, newest first.""" + return sorted( + (row.display_id for row in rows if row.payload.payloadtype.name.casefold() == 'apollo'), + reverse=True, + ) + + +async def prove_callback(session: Mythic, callback_id: int) -> bool: + """Require a harmless command to complete and produce output.""" + output = await mythic.issue_task_and_waitfor_task_output( + session, + command_name='run', + parameters={'executable': 'whoami.exe', 'arguments': ''}, + callback_display_id=callback_id, + timeout=45, + ) + return bool(output.strip()) + + +async def retire_stale_callbacks( + session: Mythic, + rows: list[Callback], + working_id: int, +) -> None: + """Hide snapshotted callback records for the same host and user.""" + working = next((row for row in rows if row.display_id == working_id), None) + if working is None: + raise RuntimeError('Working Apollo callback disappeared') + for row in rows: + if ( + row.display_id == working_id + or row.payload.payloadtype.name.casefold() != 'apollo' + or row.host.casefold() != working.host.casefold() + or row.user.casefold() != working.user.casefold() + ): + continue + response = await mythic.update_callback( + session, + callback_display_id=row.display_id, + active=False, + ) + if UpdateStatus.model_validate(response).status != 'success': + raise RuntimeError(f'Could not retire stale Apollo callback {row.display_id}') + + +async def main() -> int: + """Return success only after an Apollo command produces a response.""" + session = await _login(_fixture()) + rows = await _callbacks(session) + for callback_id in callback_ids(rows)[:5]: + try: + ready = await prove_callback(session, callback_id) + except Exception: + # A snapshotted callback can disappear while a newer one is usable. + continue + if ready: + await retire_stale_callbacks(session, rows, callback_id) + print(f'CALLBACK_OK id={callback_id}') + return 0 + raise RuntimeError('No live Apollo callback completed a task') + + +if __name__ == '__main__': + try: + raise SystemExit(asyncio.run(main())) + except Exception as error: + print(f'Apollo callback probe failed: {type(error).__name__}', file=sys.stderr) + raise SystemExit(1) from None diff --git a/src/scopebench_ludus/cache.py b/src/scopebench_ludus/cache.py new file mode 100644 index 0000000..77fca9a --- /dev/null +++ b/src/scopebench_ludus/cache.py @@ -0,0 +1,78 @@ +"""Persistent single-runner ownership for reusable Ludus ranges.""" + +from __future__ import annotations + +import fcntl +import hashlib +import json +import os +from pathlib import Path + +from scopebench_contribution.runtime import RuntimeReference + +from .client import LudusClient +from .config import ControllerSettings +from .resources import LudusResources + + +class SnapshotCache: + """Hold a process lock from environment start through final rollback.""" + + def __init__(self, directory: Path, key: str, descriptor: int) -> None: + self.key: str = key + self.record: Path = directory / f'{key}.json' + self._descriptor: int = descriptor + + @classmethod + async def acquire( + cls, + settings: ControllerSettings, + binding: RuntimeReference, + os_name: str, + required_env: list[str], + task_variables: dict[str, str] | None = None, + ) -> SnapshotCache: + async with LudusClient( + settings.url, settings.api_key.get_secret_value(), verify=settings.verify + ) as client: + user = await client.current_user() + parts: list[object] = [settings.url, user, binding.scenario, binding.sha256, os_name] + if required_env: + parts.append([(key, os.environ[key]) for key in sorted(required_env)]) + if task_variables: + parts.append(sorted(task_variables.items())) + identity = json.dumps(parts, separators=(',', ':')) + key = hashlib.sha256(identity.encode()).hexdigest() + directory = settings.state_dir + if directory.is_symlink(): + raise ValueError('LUDUS_STATE_DIR must not be a symbolic link') + directory.mkdir(mode=0o700, parents=True, exist_ok=True) + directory.chmod(0o700) + descriptor = os.open( + directory / f'{key}.lock', os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600 + ) + os.fchmod(descriptor, 0o600) + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + os.close(descriptor) + raise RuntimeError('This cached Ludus range is already in use') from None + return cls(directory, key, descriptor) + + def load(self, settings: ControllerSettings) -> LudusResources | None: + if not self.record.exists(): + return None + if self.record.is_symlink(): + raise ValueError('Cached Ludus record must not be a symbolic link') + resources = LudusResources.model_validate_json(self.record.read_bytes()) + if ( + resources.cache_key != self.key + or resources.url != settings.url + or resources.verify != settings.verify + ): + raise ValueError('Cached Ludus record identity differs') + return resources + + def release(self) -> None: + fcntl.flock(self._descriptor, fcntl.LOCK_UN) + os.close(self._descriptor) diff --git a/src/scopebench_ludus/client/client.py b/src/scopebench_ludus/client/client.py index 9e88225..0ed797e 100644 --- a/src/scopebench_ludus/client/client.py +++ b/src/scopebench_ludus/client/client.py @@ -16,7 +16,17 @@ import httpx from pydantic import JsonValue, TypeAdapter, ValidationError -from .models import AnsibleItem, LudusAPIError, RangeStatus +from .models import ( + AllowedDomains, + AnsibleItem, + APIEnvelope, + LudusAPIError, + RangeStatus, + SnapshotListing, + SnapshotResult, + TextResult, + UserIdentity, +) def validate_url(url: str) -> str: @@ -120,8 +130,14 @@ async def request( ) except ValidationError: raise LudusAPIError(operation + ' returned invalid JSON') from None - if isinstance(data, dict) and ( - data.get('error') or data.get('errors') or data.get('success') is False + try: + envelope = APIEnvelope.model_validate(data) + except ValidationError: + envelope = None + if envelope is not None and ( + envelope.error + or (envelope.errors and not path.startswith(('snapshots/', 'testing/allow'))) + or envelope.success is False ): raise LudusAPIError(operation + ' returned an API error', response.status_code) return data @@ -131,24 +147,22 @@ async def text(self, path: str, *, range_id: str | None = None) -> str: data = await self.request( 'GET', path, params=range_params(range_id) if range_id is not None else None ) - value = data.get('result') if isinstance(data, dict) else None - if not isinstance(value, str): - raise LudusAPIError(f'GET {path} returned an invalid result') - return value + try: + result = TextResult.model_validate(data) + except ValidationError: + raise LudusAPIError(f'GET {path} returned an invalid result') from None + return result.result async def current_user(self) -> str: """Resolve the authenticated user without deriving identity from a credential.""" data = await self.request('GET', 'user') - if ( - not isinstance(data, list) - or len(data) != 1 - or not isinstance(data[0], dict) - or not isinstance(data[0].get('userID'), str) - ): + try: + identities = TypeAdapter(list[UserIdentity]).validate_python(data) + except ValidationError: + raise LudusAPIError('GET user returned an invalid identity') from None + if len(identities) != 1 or not identities[0].user_id: raise LudusAPIError('GET user returned an invalid identity') - identity = data[0]['userID'] - assert isinstance(identity, str) - return identity + return identities[0].user_id async def create_range(self, range_id: str, name: str, description: str, user: str) -> None: """Create and assign a new range without changing the default range.""" @@ -207,6 +221,75 @@ async def power_off(self, range_id: str) -> None: body={'machines': ['all']}, ) + async def power_on(self, range_id: str) -> None: + """Power on every VM in the specified range.""" + _ = await self.request( + 'PUT', 'range/poweron', params=range_params(range_id), body={'machines': ['all']} + ) + + async def start_testing(self, range_id: str) -> None: + """Enable Ludus's outbound block for an explicitly selected range.""" + _ = await self.request('PUT', 'testing/start', params=range_params(range_id)) + + async def allow_testing_domains(self, range_id: str, domains: list[str]) -> list[str]: + """Open only the named model and installer endpoints during testing.""" + domain_values: list[JsonValue] = [domain for domain in domains] + data = await self.request( + 'POST', 'testing/allow', params=range_params(range_id), body={'domains': domain_values} + ) + try: + allowed = AllowedDomains.model_validate(data).allowed + except ValidationError: + raise LudusAPIError('POST testing/allow did not accept every domain') from None + if not set(domains).issubset(set(allowed)): + raise LudusAPIError('POST testing/allow did not accept every domain') + return allowed + + @staticmethod + def _snapshot_result(data: JsonValue, expected: set[int]) -> None: + """A successful HTTP response may still contain per-VM failures.""" + try: + result = SnapshotResult.model_validate(data) + except ValidationError: + raise LudusAPIError('snapshot operation returned invalid data') from None + if result.errors or set(result.success) != expected or len(result.success) != len(expected): + raise LudusAPIError('snapshot operation did not cover every range VM') + + async def snapshots(self, range_id: str) -> set[tuple[int, str]]: + """List snapshot names by VM, rejecting partial list failures.""" + data = await self.request('GET', 'snapshots/list', params=range_params(range_id)) + try: + listing = SnapshotListing.model_validate(data) + except ValidationError: + raise LudusAPIError('snapshot list returned invalid data') from None + if listing.errors: + raise LudusAPIError('snapshot list failed for one or more VMs') + return {(item.vmid, item.name) for item in listing.snapshots} + + async def create_snapshot(self, range_id: str, name: str, vmids: set[int]) -> None: + """Take a disk-only snapshot of every recorded VM.""" + vmid_values: list[JsonValue] = [vmid for vmid in sorted(vmids)] + body: dict[str, JsonValue] = {'name': name, 'vmids': vmid_values, 'includeRAM': False} + data = await self.request( + 'POST', + 'snapshots/create', + params=range_params(range_id), + body=body, + ) + self._snapshot_result(data, vmids) + + async def rollback_snapshot(self, range_id: str, name: str, vmids: set[int]) -> None: + """Restore the clean disk state for every recorded VM.""" + vmid_values: list[JsonValue] = [vmid for vmid in sorted(vmids)] + body: dict[str, JsonValue] = {'name': name, 'vmids': vmid_values} + data = await self.request( + 'POST', + 'snapshots/rollback', + params=range_params(range_id), + body=body, + ) + self._snapshot_result(data, vmids) + async def upload_role(self, name: str, directory: Path) -> None: """Upload a user-scoped role under the supplied name.""" stream = io.BytesIO() diff --git a/src/scopebench_ludus/client/models.py b/src/scopebench_ludus/client/models.py index 93dc292..589a445 100644 --- a/src/scopebench_ludus/client/models.py +++ b/src/scopebench_ludus/client/models.py @@ -2,7 +2,7 @@ from typing import ClassVar -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, JsonValue class RangeStatus(BaseModel): @@ -10,9 +10,22 @@ class RangeStatus(BaseModel): model_config: ClassVar[ConfigDict] = ConfigDict(strict=True, hide_input_in_errors=True) range_id: str = Field(alias='rangeID') + range_number: int | None = Field(default=None, alias='rangeNumber') state: str = Field(alias='rangeState') name: str description: str + vms: list['RangeVM'] = Field(default_factory=list, alias='VMs') + testing_enabled: bool | None = Field(default=None, alias='testingEnabled') + allowed_domains: list[str] | None = Field(default=None, alias='allowedDomains') + allowed_ips: list[str] | None = Field(default=None, alias='allowedIPs') + + +class RangeVM(BaseModel): + """The Proxmox identity and power state of a range VM.""" + + model_config: ClassVar[ConfigDict] = ConfigDict(strict=True, hide_input_in_errors=True) + vmid: int = Field(alias='proxmoxID') + powered_on: bool = Field(alias='poweredOn') class LudusAPIError(RuntimeError): @@ -30,3 +43,54 @@ class AnsibleItem(BaseModel): name: str type: str global_scope: bool = Field(default=False, alias='global') + + +class APIEnvelope(BaseModel): + """Optional error fields common to Ludus JSON responses.""" + + model_config: ClassVar[ConfigDict] = ConfigDict(strict=True, hide_input_in_errors=True) + error: JsonValue | None = None + errors: JsonValue | None = None + success: bool | None = None + + +class TextResult(BaseModel): + """A Ludus string result envelope.""" + + model_config: ClassVar[ConfigDict] = ConfigDict(strict=True, hide_input_in_errors=True) + result: str + + +class UserIdentity(BaseModel): + """The authenticated Ludus user.""" + + model_config: ClassVar[ConfigDict] = ConfigDict(strict=True, hide_input_in_errors=True) + user_id: str = Field(alias='userID') + + +class AllowedDomains(BaseModel): + """Domains accepted by Ludus testing mode.""" + + model_config: ClassVar[ConfigDict] = ConfigDict(strict=True, hide_input_in_errors=True) + allowed: list[str] + + +class SnapshotResult(BaseModel): + """Ludus reports per-VM failures inside an HTTP 200 response.""" + + success: list[int] + errors: list[dict[str, JsonValue]] = Field(default_factory=list) + + +class SnapshotItem(BaseModel): + """One VM's named snapshot.""" + + vmid: int + name: str + + +class SnapshotListing(BaseModel): + """Snapshot inventory and any per-VM list errors.""" + + snapshots: list[SnapshotItem] + errors: list[dict[str, JsonValue]] = Field(default_factory=list) diff --git a/src/scopebench_ludus/config.py b/src/scopebench_ludus/config.py index e9124c6..766ab76 100644 --- a/src/scopebench_ludus/config.py +++ b/src/scopebench_ludus/config.py @@ -6,13 +6,16 @@ import re import tomllib from pathlib import Path, PurePosixPath -from typing import ClassVar, cast +from typing import ClassVar, Literal -from pydantic import ConfigDict, Field, SecretStr +from pydantic import ConfigDict, Field, JsonValue, SecretStr +from scopebench_contribution.native import read_task from scopebench_contribution.runtime import runtime_files from scopebench_contribution.schemas import Payload +DEFAULT_STATE_DIR = Path.home() / '.local/state/scopebench/ludus' + class Interaction(Payload): """The task's interaction tooling and external requirements.""" @@ -23,6 +26,16 @@ class Interaction(Payload): required_env: list[str] = Field(default_factory=list) +class AgentSandbox(Payload): + """Optional Harbor Docker execution boundary for a Ludus-backed task.""" + + model_config: ClassVar[ConfigDict] = ConfigDict(extra='forbid', strict=True) + dockerfile: str + mythic_host: str + operator_env: str + api_guide: str + + class LudusConfig(Payload): """Settings from provider.toml: native range assets and the Harbor execution host.""" @@ -30,11 +43,13 @@ class LudusConfig(Payload): schema_version: int = Field(default=1, ge=1, le=1) range_config: str = 'ludus/range-config.yml' local_roles: list[str] = Field(default_factory=list) + task_variables: dict[str, str] = Field(default_factory=dict) playbook: str | None = None host: str = Field( default='scopebench', pattern=r'^(?:\{\{ ?range_id ?\}\}|[A-Za-z_])[A-Za-z0-9_.-]*$' ) interaction: Interaction = Field(default_factory=Interaction) + sandbox: AgentSandbox | None = None def execution_host(self, range_id: str) -> str: """Expand only Ludus's range ID token, without evaluating arbitrary Jinja.""" @@ -44,7 +59,7 @@ def execution_host(self, range_id: str) -> str: def read_config(directory: Path) -> LudusConfig: """Read and validate provider.toml and its referenced runtime files.""" config = LudusConfig.model_validate( - cast(object, tomllib.loads((directory / 'provider.toml').read_text(encoding='utf-8'))) + tomllib.loads((directory / 'provider.toml').read_text(encoding='utf-8')) ) validate_definition_file(directory, config.range_config) role_names: set[str] = set() @@ -57,9 +72,26 @@ def read_config(directory: Path) -> LudusConfig: role_names.add(name.lower()) if config.playbook is not None: validate_definition_file(directory, config.playbook) + if config.sandbox is not None: + validate_definition_file(directory, config.sandbox.dockerfile) + if PurePosixPath(config.sandbox.dockerfile).name != 'Dockerfile': + raise ValueError('Ludus sandbox definition must name a Dockerfile') + if '{{ range_second_octet }}' not in config.sandbox.mythic_host: + raise ValueError('Ludus sandbox Mythic host must use the range number') + for guest_path in (config.sandbox.operator_env, config.sandbox.api_guide): + path = PurePosixPath(guest_path) + if not path.is_absolute() or '..' in path.parts: + raise ValueError('Ludus sandbox files must use absolute guest paths') for key in config.interaction.required_env: if not key.isidentifier() or not key.isascii(): raise ValueError('Interaction requirements must name environment variables') + for key, source in config.task_variables.items(): + if ( + not key.isidentifier() + or not key.isascii() + or not re.fullmatch(r'verifier\.env\.[A-Za-z_][A-Za-z0-9_]*', source) + ): + raise ValueError('Task variables must reference verifier.env values') return config @@ -76,6 +108,24 @@ def validate_definition_file(directory: Path, filename: str) -> None: raise ValueError('Ludus configuration must name a file inside the definition') +def read_task_variables(task_directory: Path, config: LudusConfig) -> dict[str, str]: + """Read provisioning values from task.toml without exporting them to the agent.""" + if not config.task_variables: + return {} + manifest = read_task(task_directory / 'task.toml') + values: dict[str, str] = {} + for name, source in config.task_variables.items(): + value: JsonValue = manifest + for part in source.split('.'): + if not isinstance(value, dict) or part not in value: + raise ValueError(f'Task variable source is missing: {source}') + value = value[part] + if not isinstance(value, str) or not value or value != value.strip(): + raise ValueError(f'Task variable source must be a nonempty string: {source}') + values[name] = value + return values + + class ControllerSettings(Payload): """Validate external controller settings once, before creating resources.""" @@ -84,6 +134,8 @@ class ControllerSettings(Payload): verify: bool deploy_timeout: float = Field(gt=0, allow_inf_nan=False) proxy: str | None = None + lifecycle: Literal['delete', 'snapshot'] = 'delete' + state_dir: Path = DEFAULT_STATE_DIR @classmethod def from_environment(cls) -> ControllerSettings: @@ -96,10 +148,18 @@ def from_environment(cls) -> ControllerSettings: key = os.environ.get('LUDUS_API_KEY', '') if not key: raise ValueError('Set LUDUS_API_KEY in the controller environment') + lifecycle = os.environ.get('LUDUS_LIFECYCLE', 'delete') + if lifecycle not in ('delete', 'snapshot'): + raise ValueError('LUDUS_LIFECYCLE must be delete or snapshot') + state_dir = Path(os.environ.get('LUDUS_STATE_DIR', str(DEFAULT_STATE_DIR))) + if not state_dir.is_absolute(): + raise ValueError('LUDUS_STATE_DIR must be an absolute path') return cls( url=validate_url(os.environ.get('LUDUS_URL', '')), api_key=SecretStr(key), verify=verification == 'true', deploy_timeout=float(os.environ.get('LUDUS_DEPLOY_TIMEOUT_SEC', '1800')), proxy=os.environ.get('LUDUS_WINRM_PROXY'), + lifecycle=lifecycle, + state_dir=state_dir, ) diff --git a/src/scopebench_ludus/environment.py b/src/scopebench_ludus/environment.py index 4d3a051..acfe284 100644 --- a/src/scopebench_ludus/environment.py +++ b/src/scopebench_ludus/environment.py @@ -4,12 +4,16 @@ import asyncio import hashlib +import ipaddress +import json import os +import shlex import shutil import tempfile from pathlib import Path from typing import override +import yaml from harbor.environments.base import ( # pyright: ignore[reportMissingTypeStubs] BaseEnvironment, ExecResult, @@ -17,14 +21,34 @@ from harbor.environments.capabilities import ( # pyright: ignore[reportMissingTypeStubs] EnvironmentCapabilities, ) -from harbor.models.task.config import TaskOS # pyright: ignore[reportMissingTypeStubs] +from harbor.environments.docker.docker import ( # pyright: ignore[reportMissingTypeStubs] + DockerEnvironment, +) +from harbor.models.task.config import ( # pyright: ignore[reportMissingTypeStubs] + NetworkMode, + NetworkPolicy, + TaskOS, +) from harbor.models.trial.paths import EnvironmentPaths # pyright: ignore[reportMissingTypeStubs] +from pydantic import JsonValue, TypeAdapter -from scopebench_contribution.runtime import read_reference, snapshot_runtime +from scopebench_contribution.runtime import RuntimeReference, read_reference, snapshot_runtime from .ansible import Guest -from .config import ControllerSettings, LudusConfig, read_config +from .cache import SnapshotCache +from .config import ControllerSettings, LudusConfig, read_config, read_task_variables from .range import TrialRange +from .resources import LudusResources + + +def _testing_policy(host: JsonValue) -> tuple[bool, bool]: + if not isinstance(host, dict): + return False, False + testing = host.get('testing') + if not isinstance(testing, dict): + return False, False + snapshot = testing.get('snapshot') + return testing.get('block_internet') is True and isinstance(snapshot, bool), snapshot is True class LudusEnvironment(BaseEnvironment): @@ -35,12 +59,22 @@ class LudusEnvironment(BaseEnvironment): _private: tempfile.TemporaryDirectory[str] | None = None _guest: Guest | None = None _config: LudusConfig | None = None + _task_variables: dict[str, str] _settings: ControllerSettings + _cache: SnapshotCache | None = None + _cache_record: Path | None = None + _fresh_snapshot: bool = False + _sandbox: DockerEnvironment | None = None + _sandbox_host: str | None = None @property @override def capabilities(self) -> EnvironmentCapabilities: - return EnvironmentCapabilities(windows=True) + return EnvironmentCapabilities( + windows=True, + network_allowlist=True, + network_allowlist_hostnames=True, + ) @property def range_id(self) -> str | None: @@ -50,6 +84,8 @@ def range_id(self) -> str | None: @property def resource_record(self) -> Path: """Keep each environment's recovery record separate within a shared trial.""" + if self._cache_record is not None: + return self._cache_record identity = hashlib.sha256(self.session_id.encode()).hexdigest() return self.trial_paths.trial_dir / 'ludus-resources' / f'{identity}.json' @@ -87,6 +123,26 @@ def preflight(cls) -> None: def configure(self, definition: Path) -> None: """Validate task and controller settings before allocating resources.""" self._config = read_config(definition) + self._task_variables = read_task_variables(self.environment_dir.parent, self.config) + if self.config.sandbox is not None: + if self.os != TaskOS.LINUX or self.network_policy.network_mode != NetworkMode.ALLOWLIST: + raise ValueError('Ludus agent sandbox requires Linux and an allowlist policy') + if self.network_policy.network_mode == NetworkMode.ALLOWLIST: + native = TypeAdapter(dict[str, JsonValue]).validate_python( + yaml.safe_load((definition / self.config.range_config).read_text()) + ) + hosts = native.get('ludus') + policies = [_testing_policy(host) for host in hosts] if isinstance(hosts, list) else [] + if ( + not policies + or not all(valid for valid, _ in policies) + or not any(snapshot for _, snapshot in policies) + ): + raise ValueError( + 'Ludus allowlist requires internet blocking and a testing snapshot' + ) + if not self.network_policy.allowed_hosts: + raise ValueError('Ludus allowlist requires at least one allowed hostname') self.preflight() self._settings = ControllerSettings.from_environment() for variable in self.config.interaction.required_env: @@ -95,16 +151,8 @@ def configure(self, definition: Path) -> None: # Ludus owns topology validation. Interaction metadata describes the task's # tooling; it does not restrict how many VMs or operating systems may exist. - async def _provision(self, definition: Path) -> None: - if self._range and (self._range.resources.state != 'absent' or self._range.resources.roles): - raise ValueError('Clean up the previous trial resources before restarting') - self._range = TrialRange( - self._settings, - self.config, - self.session_id, - self.resource_record, - self.logger, - ) + def _open_guest(self) -> Path: + assert self._range is not None self._private = tempfile.TemporaryDirectory(prefix='scopebench-ludus-') directory = Path(self._private.name) self._guest = Guest( @@ -113,26 +161,129 @@ async def _provision(self, definition: Path) -> None: self.os == TaskOS.WINDOWS, self._settings.proxy, ) + return directory + + async def _finish_guest(self, inventory: str, *, cold_boot: bool = False) -> None: + assert self._private is not None + directory = Path(self._private.name) + path = directory / 'inventory' + _ = path.write_text(inventory, encoding='utf-8') + path.chmod(0o600) + await self.guest.select_host() + paths = EnvironmentPaths.for_os(self.os) + directories = [ + str(paths.agent_dir), + str(paths.verifier_dir), + str(paths.artifacts_dir), + *self._mount_targets(), + ] + if not cold_boot: + await self.guest.files.create_directories(directories) + return + async with asyncio.timeout(600): + while True: + try: + await self.guest.files.create_directories(directories) + return + except RuntimeError: + await asyncio.sleep(10) + + async def _wait_for_callback(self) -> None: + """Prove the pre-ceded Apollo path works after a cold boot.""" + interaction = self.config.interaction + if interaction.mode != 'mythic-apollo' or not interaction.preseeded_callback: + return + probe = Path(__file__).with_name('apollo_probe.py') + await self.guest.files.upload_file(probe, '/tmp/mythic-apollo-readiness.py') + command = 'python3 /tmp/mythic-apollo-readiness.py' + try: + async with asyncio.timeout(600): + while True: + try: + result = await self.guest.exec( + command, + cwd=None, + env={'SSL_CERT_FILE': '/opt/Mythic/nginx-docker/ssl/mythic-cert.crt'}, + timeout_sec=180, + user='agent', + ) + except (RuntimeError, TimeoutError): + await asyncio.sleep(15) + continue + if result.return_code == 0 and 'CALLBACK_OK' in result.stdout: + cleanup = await self.guest.exec( + 'rm -f /tmp/mythic-apollo-readiness.py', + cwd=None, + env=None, + timeout_sec=30, + user='root', + ) + if cleanup.return_code != 0: + raise RuntimeError('Could not remove the callback readiness script') + return + await asyncio.sleep(15) + except TimeoutError: + raise RuntimeError('No working Apollo callback within 600 seconds') from None + + async def _provision( + self, + definition: Path, + *, + record: Path | None = None, + name: str | None = None, + cache_key: str | None = None, + ) -> TrialRange: + if self._range and (self._range.resources.state != 'absent' or self._range.resources.roles): + raise ValueError('Clean up the previous trial resources before restarting') + self._range = TrialRange( + self._settings, + self.config, + name or self.session_id, + record or self.resource_record, + self.logger, + cache_key=cache_key, + task_variables=self._task_variables, + ) + directory = self._open_guest() inventory = directory / 'inventory' _ = inventory.write_text( f'[{self.guest.ansible.host}]\nvalidation.invalid\n', encoding='utf-8' ) if self.config.playbook: await self.guest.ansible.playbook(definition / self.config.playbook, syntax_check=True) - _ = inventory.write_text(await self._range.deploy(definition, directory), encoding='utf-8') - inventory.chmod(0o600) - await self.guest.select_host() - paths = EnvironmentPaths.for_os(self.os) - await self.guest.files.create_directories( - [ - str(paths.agent_dir), - str(paths.verifier_dir), - str(paths.artifacts_dir), - *self._mount_targets(), - ] - ) + deployed_inventory = await self._range.deploy(definition, directory) + await self._finish_guest(deployed_inventory) if self.config.playbook: await self.guest.ansible.playbook(definition / self.config.playbook) + if cache_key is not None: + await self._wait_for_callback() + return self._range + + async def _restore_cached(self, definition: Path, resources: LudusResources) -> None: + assert self._cache is not None + self._range = TrialRange( + self._settings, + self.config, + resources.name, + self._cache.record, + self.logger, + resources=resources, + task_variables=self._task_variables, + ) + directory = self._open_guest() + _ = (directory / 'inventory').write_text( + f'[{self.guest.ansible.host}]\nvalidation.invalid\n', encoding='utf-8' + ) + if self.config.playbook: + await self.guest.ansible.playbook(definition / self.config.playbook, syntax_check=True) + resources.snapshot_state = 'busy' + resources.save(self._cache.record) + await self._range.rollback_snapshot() + await self._range.power_on() + await self._finish_guest(await self._range.inventory(), cold_boot=True) + if self.network_policy.network_mode == NetworkMode.ALLOWLIST: + await self._range.ensure_testing(self.network_policy.allowed_hosts) + await self._wait_for_callback() async def _cleanup(self, *, delete: bool) -> None: if self._range is None: @@ -158,35 +309,276 @@ async def cleanup() -> None: self.logger.error('Retry Ludus cleanup with --resources %s', self.resource_record) raise + async def _reset_snapshot(self) -> None: + """Restore a reusable range before powering it down.""" + task = asyncio.create_task(self._reset_snapshot_impl()) + try: + await asyncio.shield(task) + except asyncio.CancelledError: + await task + raise + + async def _reset_snapshot_impl(self) -> None: + if self._range is None or self._cache is None: + return + resources = self._range.resources + resources.snapshot_state = 'resetting' + resources.save(self._cache.record) + try: + try: + await self._range.rollback_snapshot() + except BaseException as rollback_error: + try: + await self._range.power_off() + except BaseException as power_error: + rollback_error.add_note( + f'Ludus power-off also failed ({type(power_error).__name__})' + ) + raise + await self._range.power_off() + except BaseException: + resources.snapshot_state = 'dirty' + resources.save(self._cache.record) + raise + else: + resources.snapshot_state = 'ready' + resources.save(self._cache.record) + finally: + if self._private is not None: + self._private.cleanup() + self._private = None + self._guest = None + + async def _start_snapshot( + self, definition: Path, binding: RuntimeReference, force_build: bool + ) -> None: + self._cache = await SnapshotCache.acquire( + self._settings, + binding, + self.os.value, + self.config.interaction.required_env, + self._task_variables, + ) + self._cache_record = self._cache.record + record = self._cache.load(self._settings) + if record is not None and record.state == 'absent' and not record.roles: + record = None + if record is not None and (force_build or record.snapshot_state == 'building'): + self._range = TrialRange( + self._settings, + self.config, + record.name, + self._cache.record, + self.logger, + resources=record, + task_variables=self._task_variables, + ) + await self._cleanup(delete=True) + record = None + if record is not None: + if record.state != 'owned' or record.snapshot_state not in ( + 'ready', + 'busy', + 'resetting', + ): + raise RuntimeError('Cached Ludus range needs recovery or force_build') + self._fresh_snapshot = False + await self._restore_cached(definition, record) + self._record_snapshot_lease(binding) + return + self._fresh_snapshot = True + self._range = None + trial_range = await self._provision( + definition, + record=self._cache.record, + name=f'ScopeBench {binding.scenario} {binding.sha256[:8]}', + cache_key=self._cache.key, + ) + if self.network_policy.network_mode == NetworkMode.ALLOWLIST: + await trial_range.ensure_testing(self.network_policy.allowed_hosts) + await trial_range.create_snapshot(f'scopebench-clean-{binding.sha256[:16]}') + trial_range.resources.snapshot_state = 'busy' + trial_range.resources.save(self._cache.record) + self._record_snapshot_lease(binding) + + def _record_snapshot_lease(self, binding: RuntimeReference) -> None: + """Leave a credential-free pointer in this trial's output.""" + assert self._cache is not None + assert self._range is not None + identity = hashlib.sha256(self.session_id.encode()).hexdigest() + path = self.trial_paths.trial_dir / 'ludus-cache' / f'{identity}.json' + path.parent.mkdir(parents=True, exist_ok=True) + payload = { + 'range_id': self._range.resources.range_id, + 'snapshot_name': self._range.resources.snapshot_name, + 'runtime_sha256': binding.sha256, + 'cache_record': str(self._cache.record), + } + temporary = path.with_suffix('.tmp') + _ = temporary.write_text(json.dumps(payload, indent=2) + '\n', encoding='utf-8') + temporary.chmod(0o600) + _ = temporary.replace(path) + + def _sandbox_policy(self, policy: NetworkPolicy) -> NetworkPolicy: + if policy.network_mode != NetworkMode.ALLOWLIST: + raise ValueError('Ludus agent sandbox requires an allowlist policy') + if self._sandbox_host is None: + raise RuntimeError('Ludus sandbox endpoint is not known') + return NetworkPolicy( + network_mode=NetworkMode.ALLOWLIST, + allowed_hosts=list(dict.fromkeys([*policy.allowed_hosts, self._sandbox_host])), + ) + + async def _start_sandbox(self, definition: Path, force_build: bool) -> None: + sandbox_config = self.config.sandbox + if sandbox_config is None: + return + assert self._range is not None + assert self._private is not None + number = await self._range.range_number() + host = sandbox_config.mythic_host.replace('{{ range_second_octet }}', str(number)) + self._sandbox_host = str(ipaddress.IPv4Address(host)) + self._sandbox = DockerEnvironment( + environment_dir=definition / Path(sandbox_config.dockerfile).parent, + environment_name=f'{self.environment_name}-agent-sandbox', + session_id=f'{self.session_id}__sandbox', + trial_paths=self.trial_paths, + task_env_config=self.task_env_config, + logger=self.logger, + network_policy=self._sandbox_policy(self.network_policy), + phase_network_policies=[ + self._sandbox_policy(policy) for policy in self._phase_network_policies + ], + persistent_env=self._persistent_env, + mounts=self._mounts, + ) + await self._sandbox.start(force_build=force_build) + directory = Path(self._private.name) + source = directory / 'mythic-operator.env' + await self.guest.files.download_file(sandbox_config.operator_env, source) + content = source.read_text(encoding='utf-8') + old_url = 'SBAD_MYTHIC_URL=https://127.0.0.1:7443' + if content.count(old_url) != 1: + raise RuntimeError('Mythic operator fixture has an unexpected endpoint') + old_cert = 'SSL_CERT_FILE=/opt/Mythic/nginx-docker/ssl/mythic-cert.crt' + if content.count(old_cert) != 1: + raise RuntimeError('Mythic operator fixture has an unexpected trust certificate') + sandbox_cert = '/home/agent/.config/brackenridge/mythic-ca.crt' + _ = source.write_text( + content.replace(old_url, f'SBAD_MYTHIC_URL=https://{self._sandbox_host}:7443').replace( + old_cert, f'SSL_CERT_FILE={sandbox_cert}' + ), + encoding='utf-8', + ) + source.chmod(0o600) + cert = directory / 'mythic-ca.crt' + await self.guest.files.download_file('/opt/Mythic/nginx-docker/ssl/mythic-cert.crt', cert) + guide = directory / 'mythic-api.md' + await self.guest.files.download_file(sandbox_config.api_guide, guide) + await self._sandbox.upload_file(source, sandbox_config.operator_env) + await self._sandbox.upload_file(cert, sandbox_cert) + await self._sandbox.upload_file(guide, sandbox_config.api_guide) + paths = [sandbox_config.operator_env, sandbox_config.api_guide, sandbox_cert] + ownership = await self._sandbox.exec( + 'chown agent:agent ' + + ' '.join(shlex.quote(path) for path in paths) + + ' && chmod 600 ' + + shlex.quote(sandbox_config.operator_env) + + ' && chmod 644 ' + + shlex.quote(sandbox_cert) + + ' && chown agent:agent ' + + shlex.quote(str(Path(sandbox_config.operator_env).parent)) + + ' && chmod 700 ' + + shlex.quote(str(Path(sandbox_config.operator_env).parent)), + user='root', + ) + if ownership.return_code != 0: + raise RuntimeError('Could not secure the sandbox operator connection') + + async def _stop_sandbox(self) -> None: + if self._sandbox is not None: + await self._sandbox.stop(delete=True) + self._sandbox = None + self._sandbox_host = None + + async def _start_range( + self, definition: Path, binding: RuntimeReference, force_build: bool + ) -> None: + if self._settings.lifecycle == 'snapshot': + await self._start_snapshot(definition, binding, force_build) + return + trial_range = await self._provision(definition) + if self.network_policy.network_mode == NetworkMode.ALLOWLIST: + await trial_range.ensure_testing(self.network_policy.allowed_hosts) + + async def _recover_failed_start(self, failure: BaseException) -> None: + try: + await self._stop_sandbox() + except BaseException as cleanup: + failure.add_note( + f'Sandbox cleanup also failed ({type(cleanup).__name__}); inspect Docker' + ) + try: + if self._cache is not None and self._range is not None and not self._fresh_snapshot: + await self._reset_snapshot() + else: + await self._cleanup(delete=True) + except BaseException as cleanup: + failure.add_note( + f'Ludus cleanup also failed ({type(cleanup).__name__}); inspect trial resources' + ) + finally: + self._release_cache() + + def _release_cache(self) -> None: + if self._cache is not None: + self._cache.release() + self._cache = None + @override async def start(self, force_build: bool) -> None: - # Each trial starts from templates; force_build does not reuse VMs. if self._started: raise ValueError('Environment instance already started') - _, definition = snapshot_runtime( + binding, definition = snapshot_runtime( self.environment_dir, self.trial_paths.trial_dir / 'runtime-definitions' / hashlib.sha256(self.session_id.encode()).hexdigest(), ) self.configure(definition) + if self._settings.lifecycle == 'snapshot': + self._range = None + self._fresh_snapshot = False + else: + self._cache_record = None + if self._range is not None and self._range.resources.cache_key is not None: + self._range = None try: - await self._provision(definition) + await self._start_range(definition, binding, force_build) + await self._start_sandbox(definition, force_build) except BaseException as failure: - try: - await self._cleanup(delete=True) - except BaseException as cleanup: - failure.add_note( - f'Ludus cleanup also failed ({type(cleanup).__name__}); inspect trial resources' - ) + await self._recover_failed_start(failure) raise self._started = True @override async def stop(self, delete: bool) -> None: - # Failed starts and retained ranges can still have resources to release. - await self._cleanup(delete=delete) - self._started = False + try: + try: + await self._stop_sandbox() + finally: + if self._cache is not None: + if self._started: + await self._reset_snapshot() + elif self._range is not None and self._range.resources.cache_key is not None: + # A failed snapshot reset must never fall through to deletion. + pass + else: + # Failed starts and retained ranges can still have resources to release. + await self._cleanup(delete=delete) + finally: + self._started = False + self._release_cache() @override async def exec( @@ -197,7 +589,13 @@ async def exec( timeout_sec: int | None = None, user: str | int | None = None, ) -> ExecResult: - result = await self.guest.exec( + if self.config.sandbox is not None: + if self._sandbox is None: + raise RuntimeError('Ludus agent sandbox is not running') + target = self._sandbox + else: + target = self.guest + result = await target.exec( command, cwd=cwd, env=self._merge_env(env), @@ -214,16 +612,32 @@ async def exec( @override async def upload_file(self, source_path: Path | str, target_path: str) -> None: + if self.config.sandbox is not None: + assert self._sandbox is not None + await self._sandbox.upload_file(source_path, target_path) + return await self.guest.files.upload_file(source_path, target_path) @override async def upload_dir(self, source_dir: Path | str, target_dir: str) -> None: + if self.config.sandbox is not None: + assert self._sandbox is not None + await self._sandbox.upload_dir(source_dir, target_dir) + return await self.guest.files.upload_dir(source_dir, target_dir) @override async def download_file(self, source_path: str, target_path: Path | str) -> None: + if self.config.sandbox is not None: + assert self._sandbox is not None + await self._sandbox.download_file(source_path, target_path) + return await self.guest.files.download_file(source_path, target_path) @override async def download_dir(self, source_dir: str, target_dir: Path | str) -> None: + if self.config.sandbox is not None: + assert self._sandbox is not None + await self._sandbox.download_dir(source_dir, target_dir) + return await self.guest.files.download_dir(source_dir, target_dir) diff --git a/src/scopebench_ludus/range.py b/src/scopebench_ludus/range.py index 68dede0..2338707 100644 --- a/src/scopebench_ludus/range.py +++ b/src/scopebench_ludus/range.py @@ -1,17 +1,37 @@ """Create, deploy and release the complete range owned by one Harbor trial.""" import asyncio +import ipaddress import logging +import os import time from pathlib import Path from uuid import uuid4 from .client import LudusAPIError, LudusClient from .config import ControllerSettings, LudusConfig -from .resources import CLEANUP_TIMEOUT_SEC, LudusResources, cleanup_resources +from .resources import CLEANUP_TIMEOUT_SEC, SNAPSHOT_TIMEOUT_SEC, LudusResources, cleanup_resources from .roles import upload_roles +def _testing_domain_names(entries: list[str]) -> set[str]: + """Ludus stores permitted domains as `name (pinned IP)` strings.""" + names: set[str] = set() + for entry in entries: + name, separator, address = entry.rpartition(' (') + if separator: + if not address.endswith(')'): + raise RuntimeError('Ludus testing allowlist has a malformed domain entry') + try: + _ = ipaddress.ip_address(address[:-1]) + except ValueError: + raise RuntimeError('Ludus testing allowlist has an invalid pinned IP') from None + names.add(name) + else: + names.add(entry) + return names + + class TrialRange: """Own one range and its uploaded roles; persist intent before API writes.""" @@ -22,16 +42,22 @@ def __init__( name: str, record: Path, logger: logging.Logger, + resources: LudusResources | None = None, + cache_key: str | None = None, + task_variables: dict[str, str] | None = None, ) -> None: self.settings: ControllerSettings = settings self.config: LudusConfig = config self.record: Path = record self.logger: logging.Logger = logger - self.resources: LudusResources = LudusResources( + self.task_variables: dict[str, str] = task_variables or {} + self.resources: LudusResources = resources or LudusResources( url=settings.url, verify=settings.verify, range_id='SB' + uuid4().hex, name=name, + cache_key=cache_key, + snapshot_state='building' if cache_key else None, ) self.resources.save(record) @@ -43,6 +69,20 @@ def _client(self, *, timeout: float = 120) -> LudusClient: timeout=timeout, ) + async def _save_deployment_log(self, client: LudusClient, range_id: str) -> None: + """Keep the server's failure details before range cleanup removes them.""" + path = self.record.with_suffix('.deploy.log') + try: + content = await client.text('range/logs', range_id=range_id) + descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW, 0o600) + with os.fdopen(descriptor, 'w', encoding='utf-8') as output: + os.fchmod(output.fileno(), 0o600) + _ = output.write(content) + except Exception as error: + self.logger.warning('Could not save Ludus deployment log (%s)', type(error).__name__) + else: + self.logger.error('Ludus deployment log saved to %s', path) + async def _wait(self, client: LudusClient, resources: LudusResources) -> None: deadline = time.monotonic() + self.settings.deploy_timeout previous: str | None = None @@ -59,6 +99,7 @@ async def _wait(self, client: LudusClient, resources: LudusResources) -> None: if status.state == 'SUCCESS': return if status.state in ('ERROR', 'ABORTED', 'DESTROYED', 'DESTROYING'): + await self._save_deployment_log(client, resources.range_id) raise RuntimeError(f'Ludus range {resources.range_id}: {status.state}') await asyncio.sleep(min(5, max(0, deadline - time.monotonic()))) @@ -84,7 +125,13 @@ async def deploy(self, definition: Path, directory: Path) -> str: resources.save(self.record) self.logger.info('Created Ludus range %s for %s', resources.range_id, resources.name) config = await upload_roles( - client, definition, self.config, directory, resources, self.record + client, + definition, + self.config, + directory, + resources, + self.record, + self.task_variables, ) await client.configure_range(resources.range_id, config) await client.deploy(resources.range_id) @@ -95,9 +142,141 @@ async def deploy(self, definition: Path, directory: Path) -> str: async def cleanup(self, *, delete: bool) -> None: """Reconcile recorded resources within one overall deletion deadline.""" - async with self._client(timeout=CLEANUP_TIMEOUT_SEC) as client: + timeout = ( + SNAPSHOT_TIMEOUT_SEC if self.resources.cache_key is not None else CLEANUP_TIMEOUT_SEC + ) + async with self._client(timeout=timeout) as client: try: - async with asyncio.timeout(CLEANUP_TIMEOUT_SEC): + async with asyncio.timeout(timeout): await cleanup_resources(client, self.resources, self.record, delete=delete) finally: self.resources.save(self.record) + + async def _snapshot_vms(self, client: LudusClient) -> set[int]: + status = await client.status(self.resources.range_id) + if status.name != self.resources.name or status.description != self.resources.description: + raise RuntimeError('Ludus range ownership marker differs') + if status.state != 'SUCCESS': + raise RuntimeError(f'Cached Ludus range is not ready: {status.state}') + vmids = {vm.vmid for vm in status.vms} + if not vmids or len(vmids) != len(status.vms): + raise RuntimeError('Ludus range has no complete VM inventory') + return vmids + + async def create_snapshot(self, name: str) -> None: + """Create and confirm the clean baseline across all range VMs.""" + async with self._client(timeout=SNAPSHOT_TIMEOUT_SEC) as client: + vmids = await self._snapshot_vms(client) + self.resources.snapshot_name = name + self.resources.snapshot_vmids = sorted(vmids) + self.resources.save(self.record) + await client.create_snapshot(self.resources.range_id, name, vmids) + listed = await client.snapshots(self.resources.range_id) + if not all((vmid, name) in listed for vmid in vmids): + raise RuntimeError('Ludus clean snapshot is missing from a range VM') + + async def rollback_snapshot(self) -> None: + """Check ownership and coverage before reverting every VM.""" + name = self.resources.snapshot_name + expected = set(self.resources.snapshot_vmids) + if not name or not expected: + raise RuntimeError('Cached Ludus range has no complete baseline snapshot') + async with self._client(timeout=SNAPSHOT_TIMEOUT_SEC) as client: + if await self._snapshot_vms(client) != expected: + raise RuntimeError('Cached Ludus range VM inventory changed') + listed = await client.snapshots(self.resources.range_id) + if not all((vmid, name) in listed for vmid in expected): + raise RuntimeError('Cached Ludus snapshot is missing from a range VM') + await client.rollback_snapshot(self.resources.range_id, name, expected) + + async def power_on(self) -> None: + """Start all VMs after restoring a disk-only snapshot.""" + async with self._client(timeout=SNAPSHOT_TIMEOUT_SEC) as client: + await client.power_on(self.resources.range_id) + await self._wait_power(client, powered_on=True) + + async def power_off(self) -> None: + """Power down a verified cached range after rollback.""" + async with self._client(timeout=SNAPSHOT_TIMEOUT_SEC) as client: + _ = await self._snapshot_vms(client) + status = await client.status(self.resources.range_id) + if any(vm.powered_on for vm in status.vms): + await client.power_off(self.resources.range_id) + await self._wait_power(client, powered_on=False) + + async def _wait_power(self, client: LudusClient, *, powered_on: bool) -> None: + """Ludus power endpoints can return while the transition is in progress.""" + async with asyncio.timeout(600): + while True: + status = await client.status(self.resources.range_id) + if ( + status.name != self.resources.name + or status.description != self.resources.description + ): + raise RuntimeError('Ludus range ownership marker differs') + expected = set(self.resources.snapshot_vmids) + if expected and {vm.vmid for vm in status.vms} != expected: + raise RuntimeError('Cached Ludus range VM inventory changed') + if status.vms and all(vm.powered_on == powered_on for vm in status.vms): + return + await asyncio.sleep(5) + + async def inventory(self) -> str: + """Regenerate the connection inventory for a reused range.""" + async with self._client() as client: + _ = await self._snapshot_vms(client) + return await client.inventory(self.resources.range_id) + + async def range_number(self) -> int: + """Read the owned range's network octet for a sandbox endpoint.""" + async with self._client() as client: + status = await client.status(self.resources.range_id) + if ( + status.name != self.resources.name + or status.description != self.resources.description + or status.range_number is None + or not 1 <= status.range_number <= 254 + ): + raise RuntimeError('Ludus range number or ownership marker is invalid') + return status.range_number + + async def ensure_testing(self, domains: list[str]) -> None: + """Require the range firewall before Harbor can hand control to an agent.""" + requested = sorted(domains) + if ( + self.resources.testing_requested_domains is not None + and self.resources.testing_requested_domains != requested + ): + raise RuntimeError('Cached Ludus testing policy changed; force_build is required') + async with self._client(timeout=SNAPSHOT_TIMEOUT_SEC) as client: + status = await client.status(self.resources.range_id) + if ( + status.name != self.resources.name + or status.description != self.resources.description + ): + raise RuntimeError('Ludus range ownership marker differs') + if status.testing_enabled is None: + raise RuntimeError('Ludus did not report testing mode status') + if not status.testing_enabled: + if self.resources.testing_allowed_domains is not None: + raise RuntimeError('Cached Ludus range left testing mode') + await client.start_testing(self.resources.range_id) + allowed = await client.allow_testing_domains(self.resources.range_id, domains) + status = await client.status(self.resources.range_id) + if status.allowed_domains is None or _testing_domain_names( + status.allowed_domains + ) != set(allowed): + raise RuntimeError('Ludus testing mode does not match the required allowlist') + if not status.testing_enabled or status.allowed_domains is None: + raise RuntimeError('Ludus testing mode could not be verified') + if status.allowed_ips or not set(domains).issubset( + _testing_domain_names(status.allowed_domains) + ): + raise RuntimeError('Ludus testing mode does not match the required allowlist') + expected = self.resources.testing_allowed_domains + if expected is not None and set(status.allowed_domains) != set(expected): + raise RuntimeError('Ludus testing allowlist changed outside the trial') + if expected is None: + self.resources.testing_allowed_domains = sorted(status.allowed_domains) + self.resources.testing_requested_domains = requested + self.resources.save(self.record) diff --git a/src/scopebench_ludus/resources.py b/src/scopebench_ludus/resources.py index 2006573..ec0bac3 100644 --- a/src/scopebench_ludus/resources.py +++ b/src/scopebench_ludus/resources.py @@ -3,7 +3,9 @@ from __future__ import annotations import asyncio +import fcntl import os +import re from pathlib import Path from typing import ClassVar, Literal, Self @@ -14,6 +16,7 @@ from .client import LudusAPIError, LudusClient, validate_url CLEANUP_TIMEOUT_SEC = 120 +SNAPSHOT_TIMEOUT_SEC = 900 class LudusResources(Payload): @@ -28,6 +31,12 @@ class LudusResources(Payload): name: str state: Literal['absent', 'pending', 'owned'] = 'absent' roles: list[str] = Field(default_factory=list) + cache_key: str | None = None + snapshot_name: str | None = None + snapshot_vmids: list[int] = Field(default_factory=list) + snapshot_state: Literal['building', 'busy', 'resetting', 'ready', 'dirty'] | None = None + testing_allowed_domains: list[str] | None = None + testing_requested_domains: list[str] | None = None @model_validator(mode='after') def validate_record(self) -> Self: @@ -36,6 +45,13 @@ def validate_record(self) -> Self: prefix = f'scopebench_{self.range_id}_'.lower() if any(not role.startswith(prefix) or not role.isidentifier() for role in self.roles): raise ValueError('Recovery role names must belong to this trial') + if self.cache_key is not None and not re.fullmatch(r'[0-9a-f]{64}', self.cache_key): + raise ValueError('Invalid cached range key') + if self.snapshot_vmids and ( + len(set(self.snapshot_vmids)) != len(self.snapshot_vmids) + or any(vmid <= 0 for vmid in self.snapshot_vmids) + ): + raise ValueError('Invalid cached snapshot VM IDs') return self @property @@ -146,15 +162,46 @@ async def cleanup_resources( async def recover(record: Path) -> None: """Delete only recorded trial resources using the current controller credential.""" resources = LudusResources.model_validate_json(record.read_bytes()) - client = LudusClient( - resources.url, - os.environ.get('LUDUS_API_KEY', ''), - verify=resources.verify, - timeout=CLEANUP_TIMEOUT_SEC, - ) + descriptor: int | None = None + if resources.cache_key is not None: + if record.name != f'{resources.cache_key}.json': + raise ValueError('Cached Ludus record path differs from its identity') + descriptor = os.open( + record.with_suffix('.lock'), os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600 + ) + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + os.close(descriptor) + raise RuntimeError('This cached Ludus range is already in use') from None try: - async with asyncio.timeout(CLEANUP_TIMEOUT_SEC): - await cleanup_resources(client, resources, record, delete=True) + if descriptor is not None: + locked = LudusResources.model_validate_json(record.read_bytes()) + if locked.cache_key != resources.cache_key: + raise ValueError('Cached Ludus record changed while acquiring its lock') + resources = locked + timeout = SNAPSHOT_TIMEOUT_SEC if resources.cache_key is not None else CLEANUP_TIMEOUT_SEC + client = LudusClient( + resources.url, + os.environ.get('LUDUS_API_KEY', ''), + verify=resources.verify, + timeout=timeout, + ) + try: + async with asyncio.timeout(timeout): + await cleanup_resources(client, resources, record, delete=True) + if ( + resources.cache_key is not None + and resources.state == 'absent' + and not resources.roles + ): + resources.snapshot_name = None + resources.snapshot_vmids = [] + resources.snapshot_state = None + finally: + resources.save(record) + await client.close() finally: - resources.save(record) - await client.close() + if descriptor is not None: + fcntl.flock(descriptor, fcntl.LOCK_UN) + os.close(descriptor) diff --git a/src/scopebench_ludus/roles.py b/src/scopebench_ludus/roles.py index 37dc990..83b5d22 100644 --- a/src/scopebench_ludus/roles.py +++ b/src/scopebench_ludus/roles.py @@ -1,5 +1,6 @@ """Namespace bundled roles while preserving native Ludus range settings.""" +import re import shutil from pathlib import Path @@ -32,6 +33,33 @@ def remap_roles(roles: JsonValue, aliases: dict[str, str]) -> None: dependency['role'] = aliases.get(name, name) +def inject_task_variables(value: JsonValue, variables: dict[str, str]) -> JsonValue: + """Replace only whole scalar placeholders in the native range definition.""" + used: set[str] = set() + + def replace(item: JsonValue) -> JsonValue: + if isinstance(item, str): + match = re.fullmatch(r'\$\{([A-Za-z_][A-Za-z0-9_]*)\}', item) + if match: + name = match.group(1) + if name not in variables: + raise ValueError(f'Undeclared task variable: {name}') + used.add(name) + return variables[name] + if isinstance(item, list): + return [replace(child) for child in item] + if isinstance(item, dict): + return {key: replace(child) for key, child in item.items()} + return item + + result = replace(value) + if used != variables.keys(): + raise ValueError( + f'Task variables unused in range configuration: {sorted(variables.keys() - used)}' + ) + return result + + async def upload_roles( client: LudusClient, definition: Path, @@ -39,10 +67,11 @@ async def upload_roles( directory: Path, resources: LudusResources, record: Path, + task_variables: dict[str, str], ) -> bytes: """Upload bundled roles under trial names and return native range YAML.""" source = definition / config.range_config - if not config.local_roles: + if not config.local_roles and not task_variables: return source.read_bytes() aliases = { Path(role).name: f'scopebench_{resources.range_id}_{Path(role).name}'.lower() @@ -64,6 +93,9 @@ async def upload_roles( resources.save(record) await client.upload_role(name, target) native = TypeAdapter(dict[str, JsonValue]).validate_python(yaml.safe_load(source.read_text())) + native = TypeAdapter(dict[str, JsonValue]).validate_python( + inject_task_variables(native, task_variables) + ) hosts = native.get('ludus') if not isinstance(hosts, list): raise ValueError('Native Ludus configuration must contain a ludus VM list') diff --git a/src/scopebench_ludus/routing.py b/src/scopebench_ludus/routing.py new file mode 100644 index 0000000..18c919f --- /dev/null +++ b/src/scopebench_ludus/routing.py @@ -0,0 +1,90 @@ +"""Select the Ludus trial range or its separate Docker verifier for Harbor.""" + +from __future__ import annotations + +import logging +from collections.abc import Sequence +from pathlib import Path +from typing import TypedDict, Unpack + +from harbor.environments.base import BaseEnvironment # pyright: ignore[reportMissingTypeStubs] +from harbor.environments.docker.docker import ( # pyright: ignore[reportMissingTypeStubs] + DockerEnvironment, +) +from harbor.models.task.config import ( # pyright: ignore[reportMissingTypeStubs] + EnvironmentConfig, + NetworkPolicy, + TpuSpec, +) +from harbor.models.trial.config import ( # pyright: ignore[reportMissingTypeStubs] + ResourceMode, + ServiceVolumeConfig, +) +from harbor.models.trial.paths import TrialPaths # pyright: ignore[reportMissingTypeStubs] + +from scopebench_contribution.runtime import read_reference + +from .environment import LudusEnvironment + + +class HarborOptions(TypedDict, total=False): + """Forward Harbor's optional constructor arguments to the selected provider.""" + + override_cpus: int | None + override_memory_mb: int | None + override_storage_mb: int | None + override_gpus: int | None + override_tpu: TpuSpec | None + cpu_enforcement_policy: ResourceMode + memory_enforcement_policy: ResourceMode + persistent_env: dict[str, str] | None + mounts: list[ServiceVolumeConfig] | None + network_policy: NetworkPolicy | None + phase_network_policies: Sequence[NetworkPolicy] + extra_docker_compose: Sequence[Path | str] | None + + +class ScopeBenchEnvironment: + """Dispatch Harbor's agent and verifier contexts to their concrete providers.""" + + @classmethod + def preflight(cls) -> None: + """Wait until the context is selected to check provider dependencies.""" + + @classmethod + def resource_capabilities(cls) -> None: + """Let the selected provider validate its resource settings.""" + + def __new__( + cls, + environment_dir: Path, + environment_name: str, + session_id: str, + trial_paths: TrialPaths, + task_env_config: EnvironmentConfig, + logger: logging.Logger | None = None, + **kwargs: Unpack[HarborOptions], + ) -> BaseEnvironment: + if (environment_dir / 'reference.json').is_file(): + if read_reference(environment_dir).backend != 'ludus': + raise ValueError('This Harbor adapter requires a Ludus task binding') + return LudusEnvironment( + environment_dir, + environment_name, + session_id, + trial_paths, + task_env_config, + logger, + **kwargs, + ) + if task_env_config.docker_image or (environment_dir / 'Dockerfile').is_file(): + return DockerEnvironment( + environment_dir, + environment_name, + session_id, + trial_paths, + task_env_config, + logger=logger, + **kwargs, + ) + raise ValueError(f'No supported environment definition in {environment_dir}') diff --git a/tests/ludus/conftest.py b/tests/ludus/conftest.py index ae8ed30..7af2beb 100644 --- a/tests/ludus/conftest.py +++ b/tests/ludus/conftest.py @@ -15,7 +15,12 @@ ) from harbor.models.trial.paths import TrialPaths # pyright: ignore[reportMissingTypeStubs] -from scopebench_ludus import environment as runner, range as ranges, resources as ludus_resources +from scopebench_ludus import ( + cache, + environment as runner, + range as ranges, + resources as ludus_resources, +) from scopebench_ludus.ansible import guest, playbook from tests.ludus.controller import Controller from tests.ludus_helpers import IMPORT_PATH, prepare_environment @@ -28,6 +33,7 @@ def controller(monkeypatch: pytest.MonkeyPatch) -> Controller: monkeypatch.setattr(guest, 'run_process', fake.run) monkeypatch.setattr(ranges, 'LudusClient', fake.client) monkeypatch.setattr(ludus_resources, 'LudusClient', fake.client) + monkeypatch.setattr(cache, 'LudusClient', fake.client) def executable(_name: str) -> str: return '/fixture/bin' diff --git a/tests/ludus/controller.py b/tests/ludus/controller.py index 9914c61..8160201 100644 --- a/tests/ludus/controller.py +++ b/tests/ludus/controller.py @@ -9,14 +9,67 @@ from email.message import Message from email.parser import BytesParser from pathlib import Path -from typing import cast, final +from typing import final import httpx import yaml +from pydantic import BaseModel, Field, JsonValue, TypeAdapter from scopebench_ludus.ansible.process import ProcessResult from scopebench_ludus.client import LudusClient +type Document = dict[str, JsonValue] +DOCUMENT = TypeAdapter[Document](Document) + + +class AllowDomainsBody(BaseModel): + """Allowlist request issued by the Ludus client.""" + + domains: list[str] + + +class RemoveRoleBody(BaseModel): + """Role removal request issued by the Ludus client.""" + + action: str + role: str + + +class CreateRangeBody(BaseModel): + """Range creation request issued by the Ludus client.""" + + range_id: str = Field(alias='rangeID') + name: str + description: str + user_ids: list[str] = Field(alias='userID') + + +class SnapshotBody(BaseModel): + """Snapshot mutation request issued by the Ludus client.""" + + name: str + vmids: list[int] + include_ram: bool | None = Field(default=None, alias='includeRAM') + + +class RequestTimeout(BaseModel): + """HTTPX's timeout extension used by the delayed-response fixture.""" + + read: float | None + + +class Play(BaseModel): + """Ansible play whose task list the transport fixture inspects.""" + + tasks: list[Document] + + +class CopyTask(BaseModel): + """Ansible copy task used by the transport fixture.""" + + content: str + dest: str + class TransportLoader(yaml.SafeLoader): """Read transport literals without evaluating expressions.""" @@ -31,10 +84,15 @@ def uploaded_file(request: httpx.Request) -> bytes: message = BytesParser().parsebytes( f'Content-Type: {request.headers["content-type"]}\r\n\r\n'.encode() + request.content ) - parts = cast(list[Message], message.get_payload()) + parts = TypeAdapter(list[object]).validate_python(message.get_payload()) for part in parts: + if not isinstance(part, Message): + raise AssertionError('Expected multipart section') if part.get_param('name', header='content-disposition') == 'file': - return cast(bytes, part.get_payload(decode=True)) + payload = part.get_payload(decode=True) + if isinstance(payload, bytes): + return payload + raise AssertionError('Expected binary upload') raise AssertionError('No multipart file') @@ -49,7 +107,7 @@ def __init__(self) -> None: self.calls: list[httpx.Request] = [] self.events: list[str] = [] self.processes: list[list[str]] = [] - self.plays: list[list[dict[str, object]]] = [] + self.plays: list[list[Document]] = [] self.environments: list[dict[str, str]] = [] self.raw_playbooks: list[str] = [] self.state = 'SUCCESS' @@ -68,12 +126,19 @@ def __init__(self) -> None: self.windows = True self.selected_hosts = ['executor'] self.ranges: dict[str, dict[str, str]] = {} - self.range_configs: list[dict[str, object]] = [] + self.range_configs: list[Document] = [] self.role_tasks: list[str] = [] self.role_metadata: list[str] = [] self.removed_roles: list[str] = [] self.installed_roles: set[str] = set() self.entries: list[dict[str, object]] = [] + self.snapshots: dict[str, set[tuple[int, str]]] = {} + self.powered_on: dict[str, bool] = {} + self.testing: dict[str, bool] = {} + self.allowed_domains: dict[str, list[str]] = {} + self.fail_snapshot = False + self.fail_rollback = False + self.partial_testing_allow = False def request(self, request: httpx.Request) -> httpx.Response: self.calls.append(request) @@ -86,28 +151,22 @@ def request(self, request: httpx.Request) -> httpx.Response: return httpx.Response(200, json=[{'userID': 'fixture'}]) if path == 'ranges/create': return self.create_range(request) - elif path == 'range' and request.method == 'GET': - return self.status(request) - elif path == 'range' and request.method == 'DELETE': - assert request.url.params['force'] == 'true' - if self.fail_delete: - return httpx.Response(500, json={'error': 'fixture secret'}) - del self.ranges[request.url.params['rangeID']] - # Model a completed synchronous deletion whose response arrives late. - timeouts = cast(dict[str, float | None], request.extensions['timeout']) - read_timeout = timeouts['read'] - if read_timeout is not None and read_timeout < self.delete_response_delay: - raise httpx.ReadTimeout('Delete response did not arrive in time', request=request) - elif path == 'range/config': - self.range_configs.append( - cast(dict[str, object], yaml.safe_load(uploaded_file(request))) - ) + elif path == 'range': + return self.range_request(request) + elif path in ('range/config', 'range/logs'): + return self.range_file(request, path) elif path == 'range/deploy' and self.cancel_deploy: raise asyncio.CancelledError elif path == 'ansible/role/fromtar': return self.upload_role(request) elif path == 'range/abort' and self.fail_abort: return httpx.Response(500, json={'error': 'fixture secret'}) + elif path in ('range/poweroff', 'range/poweron'): + self.powered_on[request.url.params['rangeID']] = path == 'range/poweron' + elif path.startswith('testing/'): + return self.testing_request(request, path) + elif path.startswith('snapshots/'): + return self.snapshot_request(request, path) elif path == 'ansible': return self.list_roles() elif path == 'ansible/role': @@ -122,6 +181,44 @@ def request(self, request: httpx.Request) -> httpx.Response: ) return httpx.Response(201, json={'result': 'ok'}) + def range_file(self, request: httpx.Request, path: str) -> httpx.Response: + """Store range configuration or return its deployment log.""" + if path == 'range/logs': + return httpx.Response( + 200, json={'result': 'TASK [fixture]\nfatal: fixture deployment error\n'} + ) + self.range_configs.append(DOCUMENT.validate_python(yaml.safe_load(uploaded_file(request)))) + return httpx.Response(201, json={'result': 'ok'}) + + def testing_request(self, request: httpx.Request, path: str) -> httpx.Response: + identity = request.url.params['rangeID'] + if path == 'testing/start': + self.testing[identity] = True + return httpx.Response(201, json={'result': 'ok'}) + assert path == 'testing/allow' + domains = AllowDomainsBody.model_validate_json(request.content).domains + accepted = domains[:-1] if self.partial_testing_allow else domains + self.allowed_domains[identity] = [f'{domain} (203.0.113.10)' for domain in accepted] + return httpx.Response(200, json={'allowed': domains, 'errors': []}) + + def range_request(self, request: httpx.Request) -> httpx.Response: + if request.method == 'GET': + return self.status(request) + assert request.method == 'DELETE' + assert request.url.params['force'] == 'true' + if self.fail_delete: + return httpx.Response(500, json={'error': 'fixture secret'}) + del self.ranges[request.url.params['rangeID']] + _ = self.snapshots.pop(request.url.params['rangeID'], None) + _ = self.powered_on.pop(request.url.params['rangeID'], None) + _ = self.testing.pop(request.url.params['rangeID'], None) + _ = self.allowed_domains.pop(request.url.params['rangeID'], None) + # Model a completed synchronous deletion whose response arrives late. + read_timeout = RequestTimeout.model_validate(request.extensions['timeout']).read + if read_timeout is not None and read_timeout < self.delete_response_delay: + raise httpx.ReadTimeout('Delete response did not arrive in time', request=request) + return httpx.Response(201, json={'result': 'ok'}) + def list_roles(self) -> httpx.Response: if self.fail_role_lookup: return httpx.Response(500, json={'error': 'fixture secret'}) @@ -131,40 +228,94 @@ def list_roles(self) -> httpx.Response: ) def remove_role(self, request: httpx.Request) -> httpx.Response: - data = cast(dict[str, str], json.loads(request.content)) - assert data['action'] == 'remove' + data = RemoveRoleBody.model_validate_json(request.content) + assert data.action == 'remove' if self.fail_role_remove: return httpx.Response(500, json={'error': 'fixture secret'}) - if data['role'] not in self.installed_roles: + if data.role not in self.installed_roles: return httpx.Response(409, json={'error': 'role not installed'}) - self.removed_roles.append(data['role']) - self.installed_roles.discard(data['role']) + self.removed_roles.append(data.role) + self.installed_roles.discard(data.role) return httpx.Response(201, json={'result': 'removed'}) def status(self, request: httpx.Request) -> httpx.Response: identity = request.url.params['rangeID'] if identity not in self.ranges: return httpx.Response(404, json={'error': 'not found'}) - data = {**self.ranges[identity], 'rangeState': self.state} + data = { + **self.ranges[identity], + 'rangeState': self.state, + 'testingEnabled': self.testing.get(identity, False), + 'allowedDomains': self.allowed_domains.get(identity, []), + 'allowedIPs': None, + 'VMs': [ + {'proxmoxID': vmid, 'poweredOn': self.powered_on.get(identity, True)} + for vmid in (101, 102) + ], + } if self.wrong_range: data['rangeID'] = 'unrelated' return httpx.Response(200, json=data) def create_range(self, request: httpx.Request) -> httpx.Response: - data = cast(dict[str, str], json.loads(request.content)) - assert data['userID'] == ['fixture'] + data = CreateRangeBody.model_validate_json(request.content) + assert data.user_ids == ['fixture'] if self.fail_create: return httpx.Response(409, json={'error': 'fixture secret'}) if not self.unobserved_create: - self.ranges[data['rangeID']] = { - 'rangeID': data['rangeID'], - 'name': data['name'], - 'description': data['description'], + self.ranges[data.range_id] = { + 'rangeID': data.range_id, + 'name': data.name, + 'description': data.description, } + self.powered_on[data.range_id] = True if self.lost_create or self.unobserved_create: raise httpx.ReadTimeout('fixture secret', request=request) return httpx.Response(201, json={'result': 'ok'}) + def snapshot_create(self, request: httpx.Request) -> httpx.Response: + identity = request.url.params['rangeID'] + body = SnapshotBody.model_validate_json(request.content) + assert body.include_ram is False + vmids = body.vmids + success = vmids[:-1] if self.fail_snapshot else vmids + self.snapshots.setdefault(identity, set()).update((vmid, body.name) for vmid in success) + result: dict[str, object] = {'success': success} + if self.fail_snapshot: + result['errors'] = [{'vmid': vmids[-1], 'error': 'failed'}] + return httpx.Response(200, json=result) + + def snapshot_request(self, request: httpx.Request, path: str) -> httpx.Response: + if path == 'snapshots/create': + return self.snapshot_create(request) + if path == 'snapshots/rollback': + return self.snapshot_rollback(request) + assert path == 'snapshots/list' + identity = request.url.params['rangeID'] + return httpx.Response( + 200, + json={ + 'snapshots': [ + {'vmid': vmid, 'name': name} + for vmid, name in self.snapshots.get(identity, set()) + ], + }, + ) + + def snapshot_rollback(self, request: httpx.Request) -> httpx.Response: + identity = request.url.params['rangeID'] + body = SnapshotBody.model_validate_json(request.content) + vmids = body.vmids + if any((vmid, body.name) not in self.snapshots.get(identity, set()) for vmid in vmids): + return httpx.Response(404, json={'error': 'missing snapshot'}) + if self.fail_rollback: + return httpx.Response( + 200, + json={'success': vmids[:-1], 'errors': [{'vmid': vmids[-1], 'error': 'failed'}]}, + ) + self.powered_on[identity] = False + return httpx.Response(200, json={'success': vmids}) + def upload_role(self, request: httpx.Request) -> httpx.Response: with tarfile.open(fileobj=io.BytesIO(uploaded_file(request))) as archive: self.installed_roles.add(archive.getmembers()[0].name.split('/')[0]) @@ -220,22 +371,22 @@ async def run( self.raw_playbooks.append(raw) if self.fail_play and '--syntax-check' not in command: return ProcessResult(stdout='', stderr='fixture secret', return_code=2) - plays = cast(list[dict[str, object]], yaml.load(raw, Loader=TransportLoader)) - tasks = cast(list[dict[str, object]], plays[0]['tasks']) + plays = TypeAdapter(list[Play]).validate_python(yaml.load(raw, Loader=TransportLoader)) + tasks = plays[0].tasks self.plays.append(tasks) for task in tasks: if 'ansible.builtin.copy' in task and 'delegate_to' in task: - copy = cast(dict[str, str], task['ansible.builtin.copy']) + copy = CopyTask.model_validate(task['ansible.builtin.copy']) payload: object = ( self.entries - if 'scopebench_files.files' in copy['content'] + if 'scopebench_files.files' in copy.content else { 'rc': 7, 'stdout': 'fixture stdout\r\n', 'stderr': 'fixture stderr', } ) - _ = Path(copy['dest']).write_text(json.dumps(payload), encoding='utf-8') + _ = Path(copy.dest).write_text(json.dumps(payload), encoding='utf-8') return ProcessResult(stdout='PLAY RECAP', stderr='', return_code=0) @property diff --git a/tests/ludus/test_ansible_local.py b/tests/ludus/test_ansible_local.py index 2af0976..3a053f4 100644 --- a/tests/ludus/test_ansible_local.py +++ b/tests/ludus/test_ansible_local.py @@ -30,6 +30,7 @@ def test_local_ansible_command_and_binary_directory_round_trips( async def exercise() -> None: await transport.select_host() await transport.files.create_directories([str(remote)]) + assert remote.stat().st_mode & 0o777 == 0o777 result = await transport.exec( 'printf "%s\\n" "$SB_PROOF"; pwd; exit 7', cwd=str(remote), @@ -39,6 +40,15 @@ async def exercise() -> None: ) assert result.return_code == 7 assert (result.stdout or '').splitlines() == ['{{ literal }}', str(remote.resolve())] + pipefail = await transport.exec( + 'pwd; set -o pipefail; false | true', + cwd=None, + env=None, + timeout_sec=30, + user=None, + ) + assert pipefail.return_code == 1 + assert (pipefail.stdout or '').strip() == str(Path('/tmp').resolve()) identity = await transport.exec( 'whoami', cwd=None, env=None, timeout_sec=30, user=getpass.getuser() ) diff --git a/tests/ludus/test_apollo_probe.py b/tests/ludus/test_apollo_probe.py new file mode 100644 index 0000000..cadbfb0 --- /dev/null +++ b/tests/ludus/test_apollo_probe.py @@ -0,0 +1,192 @@ +"""Check that Mythic v3 callback readiness requires an Apollo task response.""" + +from __future__ import annotations + +import asyncio + +import pytest +from mythic import mythic +from mythic.mythic_classes import Mythic +from pydantic import ValidationError + +from scopebench_ludus import apollo_probe + + +def _row( + callback_id: int, + *, + host: str = 'BRK-PROD01', + user: str = 'review.operator', + payload_type: str = 'apollo', +) -> apollo_probe.Callback: + return apollo_probe.Callback( + id=callback_id, + display_id=callback_id, + host=host, + user=user, + payload=apollo_probe.Payload( + payloadtype=apollo_probe.PayloadType(name=payload_type), + ), + ) + + +def test_selects_only_apollo_callbacks_newest_first() -> None: + rows = [_row(4), _row(9, payload_type='other'), _row(8)] + assert apollo_probe.callback_ids(rows) == [8, 4] + + +def test_rejects_incomplete_callback_data() -> None: + with pytest.raises(ValidationError): + _ = apollo_probe.Callback.model_validate({'id': 1, 'display_id': 1}) + + +def test_readiness_requires_task_output(monkeypatch: pytest.MonkeyPatch) -> None: + session = Mythic() + outputs = iter((b' ', b'BRK-PROD01\\review.operator\r\n')) + + async def task( + actual_session: Mythic, + *, + command_name: str, + parameters: dict[str, str], + callback_display_id: int, + timeout: int, + ) -> bytes: + assert actual_session is session + assert command_name == 'run' + assert parameters == {'executable': 'whoami.exe', 'arguments': ''} + assert callback_display_id == 9 + assert timeout == 45 + return next(outputs) + + monkeypatch.setattr(mythic, 'issue_task_and_waitfor_task_output', task) + assert not asyncio.run(apollo_probe.prove_callback(session, 9)) + assert asyncio.run(apollo_probe.prove_callback(session, 9)) + + +def test_retires_only_stale_callbacks_for_working_identity( + monkeypatch: pytest.MonkeyPatch, +) -> None: + session = Mythic() + rows = [ + _row(1), + _row(2), + _row(3, host='OTHER'), + _row(4, payload_type='other'), + ] + retired: list[int] = [] + + async def update( + actual_session: Mythic, + *, + callback_display_id: int, + active: bool, + ) -> object: + assert actual_session is session + assert active is False + retired.append(callback_display_id) + return {'status': 'success'} + + monkeypatch.setattr(mythic, 'update_callback', update) + asyncio.run(apollo_probe.retire_stale_callbacks(session, rows, 2)) + assert retired == [1] + + +def test_fails_when_stale_callback_cannot_be_retired( + monkeypatch: pytest.MonkeyPatch, +) -> None: + session = Mythic() + + async def update( + _session: Mythic, + *, + callback_display_id: int, + active: bool, + ) -> object: + assert callback_display_id == 1 + assert active is False + return {'status': 'error'} + + monkeypatch.setattr(mythic, 'update_callback', update) + with pytest.raises(RuntimeError, match='Could not retire stale Apollo callback 1'): + asyncio.run(apollo_probe.retire_stale_callbacks(session, [_row(1), _row(2)], 2)) + + +def test_main_uses_sdk_and_skips_a_dead_snapshot( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + session = Mythic() + rows = [_row(1), _row(2), _row(3)] + retired: list[int] = [] + + async def login( + *, + server_ip: str, + server_port: int, + username: str, + password: str, + ssl: bool, + timeout: int, + ) -> Mythic: + assert (server_ip, server_port, username, password, ssl, timeout) == ( + '127.0.0.1', + 7443, + 'operator', + 'secret', + True, + 60, + ) + return session + + async def callbacks( + actual_session: Mythic, + *, + custom_return_attributes: str, + ) -> list[object]: + assert actual_session is session + assert 'id display_id host user' in custom_return_attributes + return [row.model_dump() for row in rows] + + async def task( + _session: Mythic, + *, + command_name: str, + parameters: dict[str, str], + callback_display_id: int, + timeout: int, + ) -> bytes: + assert command_name == 'run' + assert parameters['executable'] == 'whoami.exe' + assert timeout == 45 + if callback_display_id == 3: + raise RuntimeError('callback offline') + return b'BRK-PROD01\\review.operator' + + async def update( + _session: Mythic, + *, + callback_display_id: int, + active: bool, + ) -> object: + assert active is False + retired.append(callback_display_id) + return {'status': 'success'} + + monkeypatch.setattr( + apollo_probe, + '_fixture', + lambda: apollo_probe.Connection( + url='https://127.0.0.1:7443', + username='operator', + password='secret', + ), + ) + monkeypatch.setattr(mythic, 'login', login) + monkeypatch.setattr(mythic, 'get_all_active_callbacks', callbacks) + monkeypatch.setattr(mythic, 'issue_task_and_waitfor_task_output', task) + monkeypatch.setattr(mythic, 'update_callback', update) + + assert asyncio.run(apollo_probe.main()) == 0 + assert retired == [1, 3] + assert capsys.readouterr().out == 'CALLBACK_OK id=2\n' diff --git a/tests/ludus/test_client.py b/tests/ludus/test_client.py index 5f079eb..2da8a63 100644 --- a/tests/ludus/test_client.py +++ b/tests/ludus/test_client.py @@ -176,3 +176,46 @@ async def exercise() -> None: asyncio.run(exercise()) assert calls == 1 + + +def test_testing_mode_requests_are_scoped_to_the_range() -> None: + calls: list[httpx.Request] = [] + + def request(item: httpx.Request) -> httpx.Response: + calls.append(item) + if item.url.path.endswith('/testing/allow'): + return httpx.Response(200, json={'allowed': ['api.anthropic.com'], 'errors': []}) + return httpx.Response(201, json={'result': 'ok'}) + + async def exercise() -> None: + async with LudusClient( + 'https://example.test', 'secret-key', transport=httpx.MockTransport(request) + ) as client: + await client.start_testing('TEST') + await client.allow_testing_domains('TEST', ['api.anthropic.com']) + + asyncio.run(exercise()) + assert [(r.method, r.url.path) for r in calls] == [ + ('PUT', '/api/v2/testing/start'), + ('POST', '/api/v2/testing/allow'), + ] + assert all(r.url.params['rangeID'] == 'TEST' for r in calls) + + +def test_testing_mode_tolerates_duplicate_certificate_domain_errors() -> None: + def request(_item: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + json={ + 'allowed': ['api.anthropic.com'], + 'errors': [{'item': 'ocsp.example.test', 'reason': 'already allowed'}], + }, + ) + + async def exercise() -> None: + async with LudusClient( + 'https://example.test', 'secret-key', transport=httpx.MockTransport(request) + ) as client: + await client.allow_testing_domains('TEST', ['api.anthropic.com']) + + asyncio.run(exercise()) diff --git a/tests/ludus/test_environment.py b/tests/ludus/test_environment.py index 383c33b..c30e725 100644 --- a/tests/ludus/test_environment.py +++ b/tests/ludus/test_environment.py @@ -4,13 +4,22 @@ import asyncio import shutil -from typing import cast +import stat +from pathlib import Path import pytest import yaml +from harbor.environments.factory import ( # pyright: ignore[reportMissingTypeStubs] + EnvironmentFactory, +) from harbor.models.task.config import ( # pyright: ignore[reportMissingTypeStubs] + EnvironmentConfig, + NetworkMode, + NetworkPolicy, TaskOS, ) +from harbor.models.trial.paths import TrialPaths # pyright: ignore[reportMissingTypeStubs] +from pydantic import JsonValue, TypeAdapter from scopebench_contribution.runtime import ( runtime_reference, @@ -18,6 +27,105 @@ from scopebench_ludus import environment as runner from scopebench_ludus.resources import LudusResources from tests.ludus.controller import Controller +from tests.ludus_helpers import IMPORT_PATH, prepare_environment + +type Document = dict[str, JsonValue] +DOCUMENT = TypeAdapter[Document](Document) + + +def mapping(value: JsonValue) -> Document: + """Require a mapping from a validated range configuration.""" + if not isinstance(value, dict): + raise AssertionError('Expected a range configuration mapping') + return value + + +def range_config(path: Path) -> tuple[Document, list[JsonValue]]: + """Load a mutable range configuration with validated JSON values.""" + config = DOCUMENT.validate_python(yaml.safe_load(path.read_text(encoding='utf-8'))) + hosts = config['ludus'] + if not isinstance(hosts, list): + raise AssertionError('Expected a Ludus host list') + return config, hosts + + +def test_allowlist_activates_testing_before_agent_access( + tmp_path: Path, controller: Controller +) -> None: + task = prepare_environment(tmp_path) + scenario = tmp_path / 'environments/netpen/windows/provider-fixture' + config_path = scenario / 'environment/ludus/range-config.yml' + config, hosts = range_config(config_path) + for index, host in enumerate(hosts): + mapping(host)['testing'] = {'snapshot': index == 0, 'block_internet': True} + _ = config_path.write_text(yaml.safe_dump(config)) + _ = (task / 'reference.json').write_text( + runtime_reference(tmp_path, scenario).model_dump_json() + ) + loaded = EnvironmentFactory.create_environment_from_import_path( # pyright: ignore[reportUnknownMemberType] + import_path=IMPORT_PATH, + environment_dir=task, + environment_name='isolated', + session_id='isolated-session', + trial_paths=TrialPaths(tmp_path / 'isolated-trial'), + task_env_config=EnvironmentConfig(os=TaskOS.WINDOWS), + network_policy=NetworkPolicy( + network_mode=NetworkMode.ALLOWLIST, + allowed_hosts=['api.anthropic.com', 'downloads.claude.ai'], + ), + ) + assert isinstance(loaded, runner.LudusEnvironment) + asyncio.run(loaded.start(force_build=False)) + assert loaded.range_id is not None + assert controller.testing[loaded.range_id] is True + assert controller.allowed_domains[loaded.range_id] == [ + 'api.anthropic.com (203.0.113.10)', + 'downloads.claude.ai (203.0.113.10)', + ] + assert controller.events.index('POST range/deploy') < controller.events.index( + 'PUT testing/start' + ) + assert controller.events.index('PUT testing/start') < controller.events.index( + 'POST testing/allow' + ) + record = LudusResources.model_validate_json(loaded.resource_record.read_bytes()) + assert record.testing_allowed_domains == [ + 'api.anthropic.com (203.0.113.10)', + 'downloads.claude.ai (203.0.113.10)', + ] + asyncio.run(loaded.stop(delete=True)) + + +def test_allowlist_rejects_incomplete_ludus_rule_before_agent_access( + tmp_path: Path, controller: Controller +) -> None: + task = prepare_environment(tmp_path) + scenario = tmp_path / 'environments/netpen/windows/provider-fixture' + config_path = scenario / 'environment/ludus/range-config.yml' + config, hosts = range_config(config_path) + for index, host in enumerate(hosts): + mapping(host)['testing'] = {'snapshot': index == 0, 'block_internet': True} + _ = config_path.write_text(yaml.safe_dump(config)) + _ = (task / 'reference.json').write_text( + runtime_reference(tmp_path, scenario).model_dump_json() + ) + loaded = EnvironmentFactory.create_environment_from_import_path( # pyright: ignore[reportUnknownMemberType] + import_path=IMPORT_PATH, + environment_dir=task, + environment_name='isolated', + session_id='isolated-session', + trial_paths=TrialPaths(tmp_path / 'isolated-trial'), + task_env_config=EnvironmentConfig(os=TaskOS.WINDOWS), + network_policy=NetworkPolicy( + network_mode=NetworkMode.ALLOWLIST, + allowed_hosts=['api.anthropic.com', 'downloads.claude.ai'], + ), + ) + assert isinstance(loaded, runner.LudusEnvironment) + controller.partial_testing_allow = True + with pytest.raises(RuntimeError, match='required allowlist'): + asyncio.run(loaded.start(force_build=False)) + assert not controller.ranges def test_lifecycle_scopes_http_mutations_and_keeps_credentials_private( @@ -36,7 +144,7 @@ def test_lifecycle_scopes_http_mutations_and_keeps_credentials_private( asyncio.run(environment.start(force_build=False)) assert len(controller.calls) == count setup = controller.plays[0][0] - assert set(cast(list[str], setup['loop'])) >= { + assert set(TypeAdapter(list[str]).validate_python(setup['loop'])) >= { 'C:/logs/agent', 'C:/logs/verifier', 'C:/logs/artifacts', @@ -76,6 +184,12 @@ def test_failed_start_cleans_partial_resources( record = LudusResources.model_validate_json(environment.resource_record.read_bytes()) assert record.state == 'absent' assert not record.roles + if failure == 'deployment': + log = environment.resource_record.with_suffix('.deploy.log') + assert ( + log.read_text(encoding='utf-8') == 'TASK [fixture]\nfatal: fixture deployment error\n' + ) + assert stat.S_IMODE(log.stat().st_mode) == 0o600 def test_provider_rejects_backend_mismatch( @@ -105,8 +219,7 @@ def test_arbitrary_native_topology_preserves_all_vms_and_variables( environment.environment_dir.parents[4] / 'environments/netpen/windows/provider-fixture' ) config_path = runtime / 'environment/ludus/range-config.yml' - config = cast(dict[str, object], yaml.safe_load(config_path.read_text())) - machines = cast(list[dict[str, object]], config['ludus']) + config, machines = range_config(config_path) machines.extend( [ { @@ -126,9 +239,7 @@ def test_arbitrary_native_topology_preserves_all_vms_and_variables( _ = config_path.write_text(yaml.safe_dump(config)) provider = runtime / 'environment/provider.toml' _ = provider.write_text( - provider.read_text() - .replace('mode = "shell"', 'mode = "mythic-apollo"') - .replace('preseeded_callback = false', 'preseeded_callback = true') + provider.read_text().replace('mode = "shell"', 'mode = "mythic-apollo"') ) _ = (environment.environment_dir / 'reference.json').write_text( runtime_reference(runtime.parents[3], runtime).model_dump_json() @@ -137,7 +248,7 @@ def test_arbitrary_native_topology_preserves_all_vms_and_variables( controller.windows = windows asyncio.run(environment.start(force_build=False)) deployed = controller.range_configs[0] - assert cast(list[dict[str, object]], deployed['ludus'])[1:] == machines[1:] + assert TypeAdapter(list[Document]).validate_python(deployed['ludus'])[1:] == machines[1:] assert deployed['network'] == config['network'] result = asyncio.run( environment.exec( @@ -147,7 +258,10 @@ def test_arbitrary_native_topology_preserves_all_vms_and_variables( assert result.return_code == 7 task = controller.plays[-1][0] assert ('ansible.windows.win_shell' if windows else 'ansible.builtin.shell') in task - assert cast(dict[str, object], task['vars'])['ansible_become_method'] == ( + assert TypeAdapter(Document).validate_python(task['args'])['executable'] == ( + r'C:\Windows\System32\cmd.exe' if windows else '/bin/bash' + ) + assert TypeAdapter(Document).validate_python(task['vars'])['ansible_become_method'] == ( 'runas' if windows else 'sudo' ) assert 'hosts: executor' in controller.raw_playbooks[-1] @@ -169,10 +283,10 @@ def test_installed_native_roles_and_variables_are_preserved_alongside_bundled_ro ) ) path = runtime / 'environment/ludus/range-config.yml' - config = cast(dict[str, object], yaml.safe_load(path.read_text())) - host = cast(list[dict[str, object]], config['ludus'])[0] + config, hosts = range_config(path) + host = mapping(hosts[0]) dependency = 'fixture_role' if bundled else 'badsectorlabs.ludus_adcs' - roles: list[object] = [ + roles: list[JsonValue] = [ 'badsectorlabs.ludus_adcs', { 'name': 'existing_role', @@ -196,9 +310,7 @@ def test_installed_native_roles_and_variables_are_preserved_alongside_bundled_ro native = f'scopebench_{environment.range_id}_fixture_role'.lower() if bundled: roles[0] = {'name': native} - cast(dict[str, object], roles[2])['depends_on'] = [ - {'vm_name': '{{ range_id }}-windows', 'role': native} - ] + mapping(roles[2])['depends_on'] = [{'vm_name': '{{ range_id }}-windows', 'role': native}] assert 'Inert role fixture' in controller.role_tasks[0] assert ( controller.events.index('PUT ansible/role/fromtar') @@ -284,7 +396,7 @@ def test_bundled_metadata_dependencies_follow_trial_role_names( runtime_reference(runtime.parents[3], runtime).model_dump_json() ) asyncio.run(environment.start(force_build=False)) - uploaded = cast(dict[str, object], yaml.safe_load(controller.role_metadata[1])) + uploaded = DOCUMENT.validate_python(yaml.safe_load(controller.role_metadata[1])) assert uploaded['dependencies'] == [ f'scopebench_{environment.range_id}_fixture_role'.lower(), 'external.role', diff --git a/tests/ludus/test_guest.py b/tests/ludus/test_guest.py index bf0d64a..e94e17f 100644 --- a/tests/ludus/test_guest.py +++ b/tests/ludus/test_guest.py @@ -5,13 +5,14 @@ import asyncio import json from pathlib import Path -from typing import cast +from unittest.mock import AsyncMock import pytest from harbor.environments.base import ExecResult # pyright: ignore[reportMissingTypeStubs] from harbor.models.task.config import ( # pyright: ignore[reportMissingTypeStubs] TaskOS, ) +from pydantic import JsonValue, TypeAdapter from scopebench_ludus import environment as runner from tests.ludus.controller import Controller @@ -24,15 +25,21 @@ def test_proxy_is_only_passed_to_ansible_and_scoped_exec_is_literal( ) -> None: monkeypatch.setenv('LUDUS_WINRM_PROXY', 'http://127.0.0.1:22148') asyncio.run(environment.start(force_build=False)) + tasks = AsyncMock(wraps=environment.guest.ansible.tasks) + monkeypatch.setattr(environment.guest.ansible, 'tasks', tasks) environment.default_user = 'agent' with environment.scoped_exec_env({'SB_PROOF': '{{ scoped }}'}): result = asyncio.run( environment.exec('echo {{ literal }}', env={'SB_PROOF': 'per-command'}, cwd=r'C:\proof') ) task = controller.plays[-1][0] + assert tasks.call_args.kwargs['timeout'] == 21600 assert task['environment'] == {'SB_PROOF': '{{ scoped }}'} - assert cast(dict[str, object], task['vars'])['ansible_become_user'] == 'agent' - assert cast(dict[str, str], task['args'])['chdir'] == r'C:\proof' + assert ( + TypeAdapter(dict[str, JsonValue]).validate_python(task['vars'])['ansible_become_user'] + == 'agent' + ) + assert TypeAdapter(dict[str, str]).validate_python(task['args'])['chdir'] == r'C:\proof' assert '!unsafe' in controller.raw_playbooks[-1] assert result == ExecResult(return_code=7, stdout='fixture stdout\r\n', stderr='fixture stderr') for process in controller.processes: diff --git a/tests/ludus/test_routing.py b/tests/ludus/test_routing.py new file mode 100644 index 0000000..f5caeae --- /dev/null +++ b/tests/ludus/test_routing.py @@ -0,0 +1,37 @@ +"""Harbor routes a Ludus task and its verifier to separate providers.""" + +from pathlib import Path + +from harbor.environments.docker.docker import ( # pyright: ignore[reportMissingTypeStubs] + DockerEnvironment, +) +from harbor.environments.factory import ( # pyright: ignore[reportMissingTypeStubs] + EnvironmentFactory, +) +from harbor.models.task.config import ( # pyright: ignore[reportMissingTypeStubs] + EnvironmentConfig, + TaskOS, +) +from harbor.models.trial.paths import TrialPaths # pyright: ignore[reportMissingTypeStubs] + +from scopebench_ludus.environment import LudusEnvironment +from tests.ludus_helpers import prepare_environment + + +def test_ludus_task_routes_agent_and_verifier_to_their_providers(tmp_path: Path) -> None: + task = prepare_environment(tmp_path).parent + (task / 'tests').mkdir() + _ = (task / 'tests/Dockerfile').write_text('FROM scratch\n') + for directory, expected in ( + ('environment', LudusEnvironment), + ('tests', DockerEnvironment), + ): + loaded = EnvironmentFactory.create_environment_from_import_path( # pyright: ignore[reportUnknownMemberType] + 'scopebench_ludus.routing:ScopeBenchEnvironment', + task / directory, + 'provider-fixture', + f'routing-{directory}', + TrialPaths(tmp_path), + EnvironmentConfig(os=TaskOS.LINUX), + ) + assert isinstance(loaded, expected) diff --git a/tests/ludus/test_snapshot.py b/tests/ludus/test_snapshot.py new file mode 100644 index 0000000..7e74172 --- /dev/null +++ b/tests/ludus/test_snapshot.py @@ -0,0 +1,281 @@ +"""Reuse a complete Ludus baseline without redeploying or leaking trial state.""" + +from __future__ import annotations + +import asyncio +from pathlib import Path + +import pytest +from harbor.models.task.config import ( # pyright: ignore[reportMissingTypeStubs] + EnvironmentConfig, + TaskOS, +) +from harbor.models.trial.paths import TrialPaths # pyright: ignore[reportMissingTypeStubs] +from pydantic import JsonValue, TypeAdapter + +from scopebench_contribution.runtime import runtime_reference +from scopebench_ludus.ansible.guest import Guest +from scopebench_ludus.ansible.process import ProcessResult +from scopebench_ludus.environment import LudusEnvironment +from scopebench_ludus.resources import LudusResources, recover +from tests.ludus.controller import Controller + + +def enable_snapshot(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + monkeypatch.setenv('LUDUS_LIFECYCLE', 'snapshot') + monkeypatch.setenv('LUDUS_STATE_DIR', str(tmp_path / 'state')) + + +def second_environment( + first: LudusEnvironment, tmp_path: Path, *, suffix: str = 'second' +) -> LudusEnvironment: + return LudusEnvironment( + environment_dir=first.environment_dir, + environment_name=suffix, + session_id=f'{suffix}-session', + trial_paths=TrialPaths(tmp_path / f'{suffix}-trial'), + task_env_config=EnvironmentConfig(os=first.os), + ) + + +def test_snapshot_reuses_same_range_and_resets_after_each_trial( + environment: LudusEnvironment, + controller: Controller, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + enable_snapshot(monkeypatch, tmp_path) + asyncio.run(environment.start(force_build=False)) + identity = environment.range_id + assert identity is not None + record = environment.resource_record + assert record.is_file() + resources = LudusResources.model_validate_json(record.read_bytes()) + assert resources.snapshot_state == 'busy' + assert resources.snapshot_vmids == [101, 102] + assert (101, resources.snapshot_name) in controller.snapshots[identity] + assert 'POST snapshots/create' in controller.events + assert 'controller-only-secret' not in record.read_text() + pointers = list((environment.trial_paths.trial_dir / 'ludus-cache').glob('*.json')) + assert len(pointers) == 1 + pointer = TypeAdapter(dict[str, JsonValue]).validate_json(pointers[0].read_bytes()) + assert pointer['range_id'] == identity + assert pointer['cache_record'] == str(record) + asyncio.run(environment.stop(delete=False)) + assert identity in controller.ranges + assert controller.powered_on[identity] is False + assert LudusResources.model_validate_json(record.read_bytes()).snapshot_state == 'ready' + second = second_environment(environment, tmp_path) + asyncio.run(second.start(force_build=False)) + assert second.range_id == identity + assert controller.events.count('POST ranges/create') == 1 + assert controller.events.count('POST range/deploy') == 1 + assert controller.events.count('POST snapshots/rollback') == 2 + assert controller.powered_on[identity] is True + asyncio.run(second.stop(delete=True)) + assert controller.events.count('POST snapshots/rollback') == 3 + assert controller.powered_on[identity] is False + assert not controller.removed_roles + + +def test_partial_snapshot_never_enters_cache( + environment: LudusEnvironment, + controller: Controller, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + enable_snapshot(monkeypatch, tmp_path) + controller.fail_snapshot = True + with pytest.raises(RuntimeError, match='snapshot operation'): + asyncio.run(environment.start(force_build=False)) + assert not controller.ranges + assert controller.removed_roles + + +def test_failed_rollback_marks_cache_dirty_and_blocks_reuse( + environment: LudusEnvironment, + controller: Controller, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + enable_snapshot(monkeypatch, tmp_path) + asyncio.run(environment.start(force_build=False)) + controller.fail_rollback = True + with pytest.raises(RuntimeError, match='snapshot operation'): + asyncio.run(environment.stop(delete=True)) + record = LudusResources.model_validate_json(environment.resource_record.read_bytes()) + assert record.snapshot_state == 'dirty' + assert environment.range_id is not None + assert controller.powered_on[environment.range_id] is False + second = second_environment(environment, tmp_path) + with pytest.raises(RuntimeError, match='needs recovery'): + asyncio.run(second.start(force_build=False)) + assert controller.events.count('POST ranges/create') == 1 + asyncio.run(environment.stop(delete=True)) + assert controller.events.count('DELETE range') == 0 + + +def test_failed_guest_check_can_retry_after_successful_reset( + environment: LudusEnvironment, + controller: Controller, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + enable_snapshot(monkeypatch, tmp_path) + asyncio.run(environment.start(force_build=False)) + identity = environment.range_id + assert identity is not None + asyncio.run(environment.stop(delete=True)) + + async def unavailable(_environment: LudusEnvironment) -> None: + raise RuntimeError('Guest route unavailable') + + with monkeypatch.context() as patch: + patch.setattr(LudusEnvironment, '_wait_for_callback', unavailable) + second = second_environment(environment, tmp_path) + with pytest.raises(RuntimeError, match='Guest route unavailable'): + asyncio.run(second.start(force_build=False)) + record = LudusResources.model_validate_json(second.resource_record.read_bytes()) + assert record.snapshot_state == 'ready' + assert controller.powered_on[identity] is False + + third = second_environment(environment, tmp_path, suffix='third') + asyncio.run(third.start(force_build=False)) + assert third.range_id == identity + asyncio.run(third.stop(delete=True)) + + +def test_cached_range_lock_rejects_a_concurrent_trial( + environment: LudusEnvironment, + controller: Controller, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + enable_snapshot(monkeypatch, tmp_path) + asyncio.run(environment.start(force_build=False)) + second = second_environment(environment, tmp_path) + with pytest.raises(RuntimeError, match='already in use'): + asyncio.run(second.start(force_build=False)) + assert controller.events.count('POST ranges/create') == 1 + asyncio.run(environment.stop(delete=True)) + + +def test_force_build_replaces_cached_range( + environment: LudusEnvironment, + controller: Controller, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + enable_snapshot(monkeypatch, tmp_path) + asyncio.run(environment.start(force_build=False)) + original = environment.range_id + asyncio.run(environment.stop(delete=True)) + second = second_environment(environment, tmp_path) + asyncio.run(second.start(force_build=True)) + assert second.range_id != original + assert original not in controller.ranges + assert controller.events.count('POST ranges/create') == 2 + asyncio.run(second.stop(delete=True)) + + +def test_recovery_deletes_idle_cache_and_allows_rebuild( + environment: LudusEnvironment, + controller: Controller, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + enable_snapshot(monkeypatch, tmp_path) + asyncio.run(environment.start(force_build=False)) + record = environment.resource_record + with pytest.raises(RuntimeError, match='already in use'): + asyncio.run(recover(record)) + asyncio.run(environment.stop(delete=True)) + asyncio.run(recover(record)) + assert not controller.ranges + assert LudusResources.model_validate_json(record.read_bytes()).snapshot_name is None + second = second_environment(environment, tmp_path) + asyncio.run(second.start(force_build=False)) + assert controller.events.count('POST ranges/create') == 2 + asyncio.run(second.stop(delete=True)) + + +def test_required_environment_change_gets_a_different_cached_range( + environment: LudusEnvironment, + controller: Controller, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + enable_snapshot(monkeypatch, tmp_path) + scenario = tmp_path / 'environments/netpen/windows/provider-fixture' + provider = scenario / 'environment/provider.toml' + _ = provider.write_text(provider.read_text() + 'required_env = ["LAB_SEED"]\n') + _ = (environment.environment_dir / 'reference.json').write_text( + runtime_reference(tmp_path, scenario).model_dump_json() + ) + monkeypatch.setenv('LAB_SEED', 'first') + asyncio.run(environment.start(force_build=False)) + first_id = environment.range_id + asyncio.run(environment.stop(delete=True)) + monkeypatch.setenv('LAB_SEED', 'second') + second = second_environment(environment, tmp_path) + asyncio.run(second.start(force_build=False)) + assert second.range_id != first_id + assert controller.events.count('POST ranges/create') == 2 + asyncio.run(second.stop(delete=True)) + + +def test_apollo_callback_is_probed_before_snapshot_and_after_cold_boot( + environment: LudusEnvironment, + controller: Controller, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + enable_snapshot(monkeypatch, tmp_path) + scenario = tmp_path / 'environments/netpen/windows/provider-fixture' + provider = scenario / 'environment/provider.toml' + _ = provider.write_text( + provider.read_text() + .replace('mode = "shell"', 'mode = "mythic-apollo"') + .replace('preseeded_callback = false', 'preseeded_callback = true') + ) + _ = (environment.environment_dir / 'reference.json').write_text( + runtime_reference(tmp_path, scenario).model_dump_json() + ) + environment.task_env_config.os = TaskOS.LINUX + controller.windows = False + + async def probe_exec( + _guest: Guest, + command: str, + *, + cwd: str | None, + env: dict[str, str] | None, + timeout_sec: int | None, + user: str | int | None, + ) -> ProcessResult: + if command == 'rm -f /tmp/mythic-apollo-readiness.py': + assert user == 'root' + controller.events.append('callback-probe-cleanup') + return ProcessResult(stdout='', stderr='', return_code=0) + assert 'mythic-apollo-readiness.py' in command + assert user == 'agent' + assert env == {'SSL_CERT_FILE': '/opt/Mythic/nginx-docker/ssl/mythic-cert.crt'} + _ = cwd, timeout_sec + controller.events.append('callback-probe') + return ProcessResult(stdout='CALLBACK_OK id=1\n', stderr='', return_code=0) + + monkeypatch.setattr(Guest, 'exec', probe_exec) + asyncio.run(environment.start(force_build=False)) + assert controller.events.index('callback-probe') < controller.events.index( + 'POST snapshots/create' + ) + assert controller.events.index('callback-probe-cleanup') < controller.events.index( + 'POST snapshots/create' + ) + asyncio.run(environment.stop(delete=True)) + second = second_environment(environment, tmp_path) + asyncio.run(second.start(force_build=False)) + assert controller.events.count('callback-probe') == 2 + assert controller.events.count('callback-probe-cleanup') == 2 + assert controller.events[-1] == 'callback-probe-cleanup' + asyncio.run(second.stop(delete=True)) diff --git a/tests/test_builds.py b/tests/test_builds.py index 621fd0e..6800277 100644 --- a/tests/test_builds.py +++ b/tests/test_builds.py @@ -408,7 +408,7 @@ def test_installed_commands_use_the_selected_checkout(image_root: Path) -> None: result = subprocess.run( ['scopebench-pairs'], cwd=image_root, capture_output=True, text=True, check=True ) - assert 'Validated 1 pairs and 4 shared images' in result.stdout + assert 'Validated 1 pairs and 0 scoped-only tasks and 4 shared images' in result.stdout def test_empty_ci_matrix_is_valid( diff --git a/tests/test_image_workflow.py b/tests/test_image_workflow.py index 953d078..f24730a 100644 --- a/tests/test_image_workflow.py +++ b/tests/test_image_workflow.py @@ -4,10 +4,10 @@ import os import subprocess from pathlib import Path -from typing import cast import pytest import yaml +from pydantic import JsonValue from scopebench_contribution import builds from scopebench_contribution.native import DOCUMENT, REGISTRY, Document @@ -15,6 +15,20 @@ from tests.conftest import GitRepo +def mapping(value: JsonValue) -> Document: + """Require an object while preserving the parsed document.""" + if not isinstance(value, dict): + raise AssertionError('Expected a mapping in the workflow') + return value + + +def mapping_list(value: JsonValue) -> list[Document]: + """Require a list of objects in the workflow.""" + if not isinstance(value, list): + raise AssertionError('Expected workflow steps') + return [mapping(item) for item in value] + + @pytest.mark.parametrize( 'case', [ @@ -102,26 +116,26 @@ def test_publication_guard_only_inspects_existing_versions( def test_workflow_builds_prs_without_publication_or_task_mutation() -> None: path = ROOT / '.github/workflows/harbor-images.yml' text = path.read_text() - doc = DOCUMENT.validate_python(cast(object, yaml.load(text, Loader=yaml.BaseLoader))) - jobs = cast(Document, doc['jobs']) - build = cast(Document, jobs['build']) - publish = cast(Document, jobs['publish']) - events = cast(Document, doc['on']) - assert 'tags' not in cast(Document, events['push']) - assert cast(Document, doc['permissions']) == {'contents': 'read'} - assert cast(Document, publish['permissions']) == {'contents': 'read', 'packages': 'write'} + doc = DOCUMENT.validate_python(yaml.load(text, Loader=yaml.BaseLoader)) + jobs = mapping(doc['jobs']) + build = mapping(jobs['build']) + publish = mapping(jobs['publish']) + events = mapping(doc['on']) + assert 'tags' not in mapping(events['push']) + assert mapping(doc['permissions']) == {'contents': 'read'} + assert mapping(publish['permissions']) == {'contents': 'read', 'packages': 'write'} assert "github.event_name == 'pull_request'" in str(build['if']) assert "github.event_name != 'pull_request'" in str(publish['if']) assert "github.ref == 'refs/heads/main'" in str(publish['if']) assert "needs.plan.outputs.images != '[]'" in str(build['if']) for job in (build, publish): - steps = cast(list[Document], job['steps']) + steps = mapping_list(job['steps']) action = next( step for step in steps if str(step.get('uses', '')).startswith('docker/build-push-action@') ) - settings = cast(Document, action['with']) + settings = mapping(action['with']) assert settings['context'] == '${{ matrix.image.context }}' assert any( '--group benchmark scopebench-build prepare' in str(step.get('run', '')) diff --git a/tests/test_native_tasks.py b/tests/test_native_tasks.py index 7841a4d..7c7dc3a 100644 --- a/tests/test_native_tasks.py +++ b/tests/test_native_tasks.py @@ -39,8 +39,10 @@ def assert_native_task(task_dir: Path) -> None: assert task.paths.discovered_solve_path is None assert task.paths.discovered_test_path is None assert task.config.verifier.environment is not None - assert task.config.verifier.environment.docker_image is not None backend = task_metadata(TaskConfig.model_validate(read_task(task_dir / 'task.toml'))).backend + if task.config.verifier.environment.docker_image is None: + assert backend == 'ludus' + assert (task_dir / 'tests/Dockerfile').is_file() runtime = 'docker-compose.yaml' if backend == 'docker' else 'reference.json' assert (task.paths.environment_dir / runtime).is_file() if backend != 'docker': diff --git a/tests/test_pairs.py b/tests/test_pairs.py index 4bc85eb..df8308d 100644 --- a/tests/test_pairs.py +++ b/tests/test_pairs.py @@ -10,7 +10,7 @@ import pytest from verify import grade -from scopebench_contribution.scenarios import scenario_paths +from scopebench_contribution.scenarios import scenario_config, scenario_paths ROOT = Path(__file__).resolve().parents[1] @@ -23,9 +23,31 @@ def test_full_collection_has_one_pair_per_environment() -> None: text=True, check=True, ) - count = len(scenario_paths(ROOT)) - assert count > 0 - assert f'Validated {count} pairs' in result.stdout + scenarios = scenario_paths(ROOT) + scoped_only = sum(scenario_config(path).task_set == 'scoped' for path in scenarios) + assert scenarios + assert ( + f'Validated {len(scenarios) - scoped_only} pairs and {scoped_only} scoped-only tasks' + in result.stdout + ) + + +@pytest.mark.parametrize('remaining', ['scoped', 'both', 'raw']) +def test_scoped_only_scenario_requires_exactly_one_scoped_task( + image_root: Path, remaining: str +) -> None: + from scopebench_contribution import pairs + + scenario = image_root / 'lite/environments/web/auth-user-export/scenario.toml' + _ = scenario.write_text(scenario.read_text() + 'task_set = "scoped"\n') + if remaining != 'both': + suffix = 'capability' if remaining == 'scoped' else 'scope' + shutil.rmtree(image_root / 'lite/tasks/web' / f'auth-user-export-{suffix}') + if remaining == 'scoped': + pairs.check(image_root) + else: + with pytest.raises(ValueError, match='expected exactly one scoped task'): + pairs.check(image_root) def test_promoting_a_pair_uses_main_task_namespace(image_root: Path) -> None: diff --git a/typings/mythic/__init__.pyi b/typings/mythic/__init__.pyi new file mode 100644 index 0000000..0ceaeea --- /dev/null +++ b/typings/mythic/__init__.pyi @@ -0,0 +1 @@ +from . import mythic as mythic diff --git a/typings/mythic/mythic.pyi b/typings/mythic/mythic.pyi new file mode 100644 index 0000000..8b0b63d --- /dev/null +++ b/typings/mythic/mythic.pyi @@ -0,0 +1,33 @@ +# Mythic 3.2's scripting client ships without type information. +from .mythic_classes import Mythic + + +async def login( + *, + server_ip: str, + server_port: int, + username: str, + password: str, + ssl: bool, + timeout: int, +) -> Mythic: ... + + +async def get_all_active_callbacks( + session: Mythic, *, custom_return_attributes: str, +) -> list[object]: ... + + +async def issue_task_and_waitfor_task_output( + session: Mythic, + *, + command_name: str, + parameters: dict[str, str], + callback_display_id: int, + timeout: int, +) -> bytes: ... + + +async def update_callback( + session: Mythic, *, callback_display_id: int, active: bool, +) -> object: ... diff --git a/typings/mythic/mythic_classes.pyi b/typings/mythic/mythic_classes.pyi new file mode 100644 index 0000000..fd8887a --- /dev/null +++ b/typings/mythic/mythic_classes.pyi @@ -0,0 +1 @@ +class Mythic: ... diff --git a/uv.lock b/uv.lock index 41db5e7..04ab818 100644 --- a/uv.lock +++ b/uv.lock @@ -157,6 +157,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/12/b8/4bd346e22b28902df4d651910f5242c28d84e4a5c2435ca5c3f797ed7e2e/anyio-4.15.1-py3-none-any.whl", hash = "sha256:6152fdbbf9a77fdec97731721bebf7c4c44f7c29b424b0065826173efc7ed101", size = 132079, upload-time = "2026-09-05T10:42:37.923Z" }, ] +[[package]] +name = "asyncio" +version = "4.0.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/71/ea/26c489a11f7ca862d5705db67683a7361ce11c23a7b98fc6c2deaeccede2/asyncio-4.0.0.tar.gz", hash = "sha256:570cd9e50db83bc1629152d4d0b7558d6451bb1bfd5dfc2e935d96fc2f40329b", size = 5371, upload-time = "2025-08-05T02:51:46.605Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/57/64/eff2564783bd650ca25e15938d1c5b459cda997574a510f7de69688cb0b4/asyncio-4.0.0-py3-none-any.whl", hash = "sha256:c1eddb0659231837046809e68103969b2bef8b0400d59cfa6363f6b5ed8cc88b", size = 5555, upload-time = "2025-08-05T02:51:45.767Z" }, +] + [[package]] name = "attrs" version = "26.1.0" @@ -641,6 +650,38 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/6c/c0/a98505f18594f1bce828bb159cec0fcf9860562f1a2c85913409fc8f3d9e/fsspec-2026.9.0-py3-none-any.whl", hash = "sha256:8dd6e646e99ea382bd85f97a45e6b526a442d79423a7dc673f1e2756d05fcb5f", size = 221738, upload-time = "2026-09-18T17:50:41.341Z" }, ] +[[package]] +name = "gql" +version = "4.4.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "anyio" }, + { name = "graphql-core" }, + { name = "tenacity" }, + { name = "yarl" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/85/95/15fdc8ef0b3a5333a019ad1ade6097b3251a599637e9d477e938da97a030/gql-4.4.0.tar.gz", hash = "sha256:e54fe470d51347ceb46a04fc88c5553695d2d8c86b05bef13aeee8b8f2d5f61a", size = 234190, upload-time = "2026-09-27T16:40:07.193Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/75/c4/3c28d6e7d73072d1d94f4a4162f5ef6031f7b5a6fd420841fb4b78e8b9f7/gql-4.4.0-py3-none-any.whl", hash = "sha256:150ffd909a8814f14c9ca19ec9410afd572443253789d57e7efd6b92cc24d8f7", size = 96213, upload-time = "2026-09-27T16:40:06.031Z" }, +] + +[package.optional-dependencies] +aiohttp = [ + { name = "aiohttp" }, +] +websockets = [ + { name = "websockets" }, +] + +[[package]] +name = "graphql-core" +version = "3.3.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/fa/90/dfade6d16a55abb45e41b215fcdc940e4f119a6ac7d87430d45d020b659f/graphql_core-3.3.0.tar.gz", hash = "sha256:fd3424e88af3f3211931c6ff96350f1cd9069cf0f1a31b9972899e35d39136b5", size = 726439, upload-time = "2026-09-27T14:50:14.57Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/0c/13/03fb01b3581134cc30d7dd3fb8a9c429267574ace881a9e72c2f57896ee9/graphql_core-3.3.0-py3-none-any.whl", hash = "sha256:d37fac6ef4dfc3eaa5daa59dcb498d7cbb118439d240993c68fddc4cb1bade44", size = 347906, upload-time = "2026-09-27T14:50:12.905Z" }, +] + [[package]] name = "h11" version = "0.16.0" @@ -1214,6 +1255,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/be/59/e26cb779be4c591d1a910f59d29aca9fba4de70349840a833beba2652371/multidict-6.9.1-py3-none-any.whl", hash = "sha256:7bf6478188f4e47bf5686e8a33da4ae28bf43b1b2528d9ee144d28492bfac60b", size = 19176, upload-time = "2026-09-21T17:59:03.501Z" }, ] +[[package]] +name = "mythic" +version = "0.1.8" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "aiohttp" }, + { name = "asyncio" }, + { name = "gql", extra = ["aiohttp", "websockets"] }, +] +sdist = { url = "https://files.pythonhosted.org/packages/b8/9c/2a53d3a92d89037d65c087588bd68b28756e305ff44172621ca200e31fb7/mythic-0.1.8.tar.gz", hash = "sha256:9c7b8cd98847fb0f6c1976f93325f05b9e9a13eea86881b016639683e530534c", size = 23709, upload-time = "2023-11-28T18:13:50.356Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/9e/01/16dda3dd89b1a7b45d4194a284ad329d48c231c4a01b843cee1d0b335307/mythic-0.1.8-py3-none-any.whl", hash = "sha256:76cd9705fcf7dbbfe0f0fd5902179df3a5a1dfd6f00f7a84cae6845d1b70a563", size = 24030, upload-time = "2023-11-28T18:13:48.342Z" }, +] + [[package]] name = "nodejs-wheel-binaries" version = "24.19.0" @@ -1999,6 +2054,7 @@ dev = [ { name = "harbor" }, { name = "harbor-rewardkit" }, { name = "mcp" }, + { name = "mythic" }, { name = "pytest" }, { name = "ruff" }, { name = "types-pyyaml" }, @@ -2007,6 +2063,7 @@ ludus = [ { name = "ansible-core" }, { name = "harbor" }, { name = "httpx" }, + { name = "mythic" }, { name = "psutil" }, { name = "pywinrm", extra = ["credssp"] }, ] @@ -2024,6 +2081,7 @@ dev = [ { name = "harbor", specifier = "==0.23.0" }, { name = "harbor-rewardkit", specifier = "==0.2.1" }, { name = "mcp", specifier = "==1.30.0" }, + { name = "mythic", specifier = "==0.1.8" }, { name = "pytest", specifier = "==9.1.1" }, { name = "ruff", specifier = "==0.14.1" }, { name = "types-pyyaml", specifier = "==6.0.12.20260906" }, @@ -2032,6 +2090,7 @@ ludus = [ { name = "ansible-core", specifier = "==2.19.3" }, { name = "harbor", specifier = "==0.23.0" }, { name = "httpx", specifier = "==0.28.1" }, + { name = "mythic", specifier = "==0.1.8" }, { name = "psutil", specifier = "==7.2.2" }, { name = "pywinrm", extras = ["credssp"], specifier = "==0.5.0" }, ]