diff --git a/docs/automation_readiness/stage2_vrrp_readonly_s2_ro_10_trusted_runtime_composition.md b/docs/automation_readiness/stage2_vrrp_readonly_s2_ro_10_trusted_runtime_composition.md index be1d648..724702e 100644 --- a/docs/automation_readiness/stage2_vrrp_readonly_s2_ro_10_trusted_runtime_composition.md +++ b/docs/automation_readiness/stage2_vrrp_readonly_s2_ro_10_trusted_runtime_composition.md @@ -2,12 +2,18 @@ ## Decision summary -S2-RO-10 composes one bounded VRRP observation from the accepted S2-RO-01 -through S2-RO-09 components. Status: validated implementation candidate; -independent security review PASS, zero unresolved material findings. Ready for -separate local-commit authorization. It returns only normalized S2-RO-01 evidence. -No startup executable, CLI, live task registration, or S2-RO-11 behavior is added. -Implementation and offline validation do not authorize a live invocation. +Historically, S2-RO-10 composed and proved one bounded Lab1 VRRP observation +from the accepted S2-RO-01 through S2-RO-09 components. The current candidate +is a bounded target-aware composition extension for the complete accepted +Lab1 + Lab2 fixed registry. It returns only normalized S2-RO-01 evidence and +preserves the historical Lab1 path through the same target-aware operations. + +The Lab2 candidate is validated only with synthetic offline data. It does not +claim a Lab2 live PASS, a Lab2 S2-RO-11 PASS, full Stage-2 Lab2 runtime proof, +production readiness, or write automation. No startup executable, CLI, live +task registration, or S2-RO-11 behavior is added. + +`OFFLINE_IMPLEMENTATION_AUTHORITY != LIVE_EXECUTION_AUTHORITY` `OWNER_VERIFIED != AUTHORIZATION_CONSUMED != TRANSPORT_SUCCESS != PARSE_SUCCESS != FINAL_OBSERVATION_SUCCESS` @@ -52,8 +58,12 @@ backend, transport override, or dependency dictionary in configuration. Exact types and initialized fields are required. Inert nested records are captured and revalidated with their accepted constructors before external -acquisition. Target and known-host configuration must agree. Request fields -are captured and reparsed with S2-RO-01. Malformed objects and subclasses fail. +acquisition. The complete supplied S2-RO-02 fixed registry is retained, and +the request's exact `target_ref` selects its endpoint through the existing +registry policy. The single trusted known-host configuration must name an +endpoint in that registry and must equal the endpoint selected for the current +request before credential access or transport. Request fields are captured +and reparsed with S2-RO-01. Malformed objects and subclasses fail. The Owner-root API has no public inert pin configuration validator. Composition reuses its private pure path, FileId, and digest syntax checks. This is an @@ -71,9 +81,11 @@ there is no cache, hot reload, or repeated invocation loop here. ## Ordered composition 1. Validate and capture the exact request and trusted configuration. -2. Parse the canonical S2-RO-05 envelope, resolve and revalidate the S2-RO-02 - endpoint, resolve the S2-RO-03 credential binding, and validate all bindings - against the current UTC time. +2. Parse the canonical S2-RO-05 envelope, resolve and revalidate the exact + request-selected S2-RO-02 endpoint, require the trusted known-host endpoint + to equal it, resolve the S2-RO-03 credential binding with + `resolve_for_target(request.target_ref, request.credential_ref)`, and + validate all bindings against the current UTC time. 3. Acquire the pinned Owner root once. Construct `Stage2OwnerVerifier` and call `verify(envelope)` once. Validate the exact issued result and matching facts. S2-RO-06 owns the sole approval read; composition performs no separate @@ -83,7 +95,8 @@ there is no cache, hot reload, or repeated invocation loop here. 5. Acquire the S2-RO-07 known-host snapshot once and check the exact endpoint, source pins, Ed25519 blob, digest, fingerprint, and non-authority flag. 6. Revalidate authorization/time, construct the default S2-RO-04 backend, and - read the fixed credential binding once. + read the selected binding once with + `read_for_target(request.target_ref, binding)`. 7. Resolve the S2-RO-08 command once, revalidate authorization/time immediately before transport, then invoke S2-RO-09 once with exactly endpoint, credential, snapshot, and command specification. Release the credential reference in a @@ -100,7 +113,9 @@ there is no cache, hot reload, or repeated invocation loop here. S2-RO-05 performs additional internal pure binding checks. The S2-RO-08 parser internally revalidates command policy. These do not consume another replay -record, read another credential, or execute another command. +record, read another credential, or execute another command. Lab1 and Lab2 use +this one target-aware composition path; there is no Lab2-to-Lab1 fallback, +alternate credential read, or target-specific retry path. ## Replay and failure ownership @@ -188,7 +203,11 @@ Python process or caller-owned objects. No prior-slice or integration-plan file changes, dependencies, second trust root, second evidence schema, environment authority, CLI, interactive prompt, startup executable, live registration, scheduler, worker, AI loop, retry, -configuration backup/change, or S2-RO-11 implementation is included. +configuration backup/change, or S2-RO-11 implementation is included. S2-RO-09 +is unchanged. The existing S2-RO-10 public API, seven-field configuration, +sixteen-category failure enum, evidence schema, authority order, and replay +semantics are unchanged. S2-RO-11 is not modified; its existing one-call handoff +continues to depend only on request-bearing trusted configuration. The implementation task performs no real trust-root/approval/known-host read, Credential Manager read, persistent replay mutation, loopback, DNS, SSH, or @@ -196,61 +215,33 @@ RouterOS access. Remote base inspection and local branch creation are separate repository operations, not device validation. Staging, commit, push, PR, merge, and branch/worktree deletion require separate authorization. -## Validation and review - -Tests use synthetic exact-type records and private module monkeypatches. -Guards prohibit real acquisition and network. Integration tests use the real -S2-RO-05 ledger only against disposable synthetic SQLite state, force each -later failure, and prove the second invocation rejects before credentials or -transport. Tests also cover strict types, missing fields, ordered counts, -clock boundaries, unchanged stdout object identity, integrity cross-binding, -credential reference release, and sanitized failures for every child category. - -Required validation order: three-file scope, UTF-8 without BOM/LF-only, -focused tests with no skips, Stage-2 regression, full pytest, report-index, -whitespace, independent read-only security review, and documentation review. -The accepted guarded Windows launcher disables native/network operations, -plugin autoload, cache, and bytecode before pytest startup. Its inert non-TTY -Colorama/click stubs avoid native console initialization. Allowed Python/Node -regression children retain guards; Git fixtures remain temporary and local. - -The two real Win32 trust-root tests and the Flask process/socket lifecycle test -are the only accepted local safety skips. No S2-RO-10 test was skipped. - -### Recorded candidate evidence - -| Validation | Collected | Passed | Failed | Skipped | -| --- | ---: | ---: | ---: | ---: | -| Focused S2-RO-10 | 272 | 272 | 0 | 0 | -| Stage 2 | 1416 | 1414 | 0 | 2 | -| Full pytest | 3542 | 3539 | 0 | 3 | - -- Report-index: 14/14 PASS; zero failures, warnings, missing, or unknown entries. -- UTF-8 without BOM, LF-only, and whitespace validation: PASS. Because the - candidate is untracked, direct file checks supplement `git diff --check`. -- Independent read-only security review: PASS; zero unresolved material - findings. The reviewer inspected all three files and relevant accepted - contracts, using the recorded guarded test results without another test run. -- Documentation readability and local reference links: PASS. -- Prior S2-RO-01 through S2-RO-09 files: unchanged. Index: empty. - -Focused validation executed `tests/stage2/test_trusted_runtime_composition.py`; -regression executed `tests/stage2` and the full suite. Report-index executed -`network_lab.py --task report-index`. All used the accepted guarded Python 3.13 -validation setup, pytest 8.4.2, and existing dependencies. Early test-fixture -issues involving an oversized pytest identifier and a deliberately mutated -comparison object were corrected in the candidate tests; the final runs pass. - -The two Stage-2 skips are -`test_disposable_regular_file_is_accepted_by_exact_native_path` and -`test_confirmed_trailing_dot_alias_is_rejected_as_noncanonical`, both requiring -real Win32 APIs. The additional full-suite skip is -`test_canonical_flask_process_lifecycle_and_get_only_routes`, which requires -process/socket lifecycle operations excluded by this task. - -Candidate hashes are frozen after validation and review and returned to the -Owner. This delivery grants no staging, commit, push, PR, merge, or live-access -authority. S2-RO-11 remains unimplemented and separately gated. +## Offline validation and review boundary + +Tests use synthetic exact-type records and private module monkeypatches. The +Lab1 regression and Lab2 success cases verify exact target-aware resolver and +backend arguments, one Owner verification, one replay consumption, one +known-host acquisition, one credential acquisition, one command resolution, +one transport invocation, exact command text, unchanged stdout object identity, +one normalized evidence object, attempt one, retry zero, and non-authority. + +Negative cases reject both cross-lab credential pairs, wrong trusted credential +locators, known-host/endpoint mismatches, and authorization binding mismatches. +Disposable SQLite cases prove that Lab2 authorization remains spent after +known-host, credential, command, transport, parser, and evidence-construction +failure. They also prove no alternate credential read, second replay success, +transport fallback, or retry occurs. + +Guards prohibit real Credential Manager, Owner trust-root, approval artifact, +known-host source, socket, DNS, SSH, and device access. The real S2-RO-05 ledger +is used only against disposable synthetic test state. No Lab1 or Lab2 device is +contacted, and no RouterOS command is emitted. Public failures remain sanitized. + +The required local validation is exact three-file scope, UTF-8 without BOM and +LF-only text, `git diff --check`, focused S2-RO-10 tests without skips, Stage-2 +regression with only accepted safety skips, full pytest with only accepted +safety skips, report-index, documentation readability, and secret-diff review. +An independent read-only security review remains the next task. This candidate +does not inherit the historical Lab1 security review as a Lab2 review. References: [integration gates](actual_automation_integration_plan.md), [S2-RO-01 evidence](stage2_vrrp_readonly_s2_ro_01_contract.md), diff --git a/tests/stage2/test_trusted_runtime_composition.py b/tests/stage2/test_trusted_runtime_composition.py index febd83a..f02936d 100644 --- a/tests/stage2/test_trusted_runtime_composition.py +++ b/tests/stage2/test_trusted_runtime_composition.py @@ -24,6 +24,7 @@ REAL_PARSE = m._parser.parse_stage2_vrrp_readonly_output REAL_POLICY = m._policy.resolve_stage2_vrrp_readonly_command REAL_EVIDENCE = m._contract.Stage2VrrpObservationEvidence +REAL_RESOLVER = m._resolver.build_stage2_fixed_credential_resolver def issue(kind, **values): @@ -60,8 +61,11 @@ def setup(monkeypatch, tmp_path): "1.0", "mikrotik.vrrp_status", "run.synthetic", "target.mikrotik.lab01", "credential.mikrotik.lab01", "authorization.synthetic", True) endpoint = m._target.Stage2FixedTargetEndpoint(request.target_ref, "192.0.2.10", 22, "SSH", True) + lab2_endpoint = m._target.Stage2FixedTargetEndpoint( + m._target.STAGE2_SECOND_TARGET_REF, "192.0.2.20", 22, "SSH", True) config = m.Stage2TrustedRuntimeConfiguration( - m._target.Stage2FixedTargetRegistry(endpoint), r"c:\synthetic\root.json", IDENTITY, "a" * 64, + m._target.Stage2FixedTargetRegistry(endpoint, lab2_endpoint), + r"c:\synthetic\root.json", IDENTITY, "a" * 64, m._authorization.Stage2ReplayLedgerConfiguration( tmp_path / "synthetic.db", INSTANCE, (Path(__file__).resolve().parents[2],)), m._host.Stage2KnownHostSourceConfiguration(r"c:\synthetic\host.json", IDENTITY, "b" * 64, endpoint), @@ -90,11 +94,13 @@ def setup(monkeypatch, tmp_path): source_file_identity=IDENTITY, source_artifact_sha256="b" * 64, execution_authorized=False) consumed = m._authorization.Stage2ConsumptionRecord( AUTH_ID, hashlib.sha256(envelope.to_canonical_bytes()).hexdigest(), 1100) - s = SimpleNamespace(request=request, config=config, envelope=envelope, root=root, + s = SimpleNamespace(request=request, config=config, envelope=envelope, + endpoint=endpoint, lab2_endpoint=lab2_endpoint, root=root, verification=verification, snapshot=snapshot, consumed=consumed, credential=m._windows.Stage2ResolvedCredential("synthetic-user", SENTINEL.encode()), command=REAL_POLICY(request), raw=RAW, trace=[], fail={}, returns={}, spent=False, - parser_input=None, transport_inputs=None, credential_released=[], evidence_fields=None) + parser_input=None, transport_inputs=None, resolver_inputs=[], credential_inputs=[], + credential_released=[], evidence_fields=None) def step(name, value): s.trace.append(name) @@ -102,8 +108,13 @@ def step(name, value): raise s.fail[name] return s.returns.get(name, value) + class Resolver: + def resolve_for_target(self, target_ref, credential_ref): + s.resolver_inputs.append((target_ref, credential_ref)) + return REAL_RESOLVER().resolve_for_target(target_ref, credential_ref) + def acquire_root(**pins): - assert pins == dict(expected_path=config.owner_trust_root_expected_path, + assert pins == dict(expected_path=s.config.owner_trust_root_expected_path, expected_file_identity=IDENTITY, expected_file_sha256="a" * 64) return step("root", s.root) @@ -118,12 +129,13 @@ def verify(self, value): class Ledger: def __init__(self, value): - assert value == config.replay_ledger_configuration + assert value == s.config.replay_ledger_configuration step("ledger", None) def consume(self, e, r, registry, binding, *, utc_now): assert e == s.envelope and r == s.request and registry == s.config.target_registry - assert binding == m._resolver.build_stage2_fixed_credential_resolver().resolve(r.credential_ref) + assert binding == REAL_RESOLVER().resolve_for_target( + r.target_ref, r.credential_ref) assert utc_now is m._utc_now value = step("consume", s.consumed) if s.spent: @@ -132,20 +144,31 @@ def consume(self, e, r, registry, binding, *, utc_now): return value def host(value): - assert value.expected_path == r"c:\synthetic\host.json" - assert value.expected_file_identity == IDENTITY and value.expected_file_sha256 == "b" * 64 - assert value.expected_endpoint.address == "192.0.2.10" + expected = s.config.known_host_configuration + assert value.expected_path == expected.expected_path + assert value.expected_file_identity == expected.expected_file_identity + assert value.expected_file_sha256 == expected.expected_file_sha256 + assert value.expected_endpoint == expected.expected_endpoint return step("host", s.snapshot) class Backend: - def read(self, binding): - assert binding.credential_ref == request.credential_ref + def __init__(self, value): + self.configuration = value + + def read_for_target(self, target_ref, binding): + s.credential_inputs.append((target_ref, binding)) + expected = REAL_RESOLVER().resolve_for_target( + target_ref, binding.credential_ref) + if self.configuration.locator_ref != expected.locator_ref: + raise m._windows.Stage2WindowsCredentialError( + m._windows.Stage2WindowsCredentialFailure.UNSUPPORTED_LOCATOR) + assert binding == expected return step("credential", s.credential) def backend(value): - assert value == config.credential_configuration + assert value == s.config.credential_configuration step("backend", None) - return Backend() + return Backend(value) def command(value): assert value == s.request @@ -153,7 +176,7 @@ def command(value): def result(): return issue(m._transport.Stage2PinnedSshCommandResult, - schema_version="s2-ro-09.pinned-ssh-command-result.v1", target_ref=request.target_ref, + schema_version="s2-ro-09.pinned-ssh-command-result.v1", target_ref=s.request.target_ref, command_policy_version=s.command.command_policy_version, stdout_bytes=s.raw, stdout_sha256=hashlib.sha256(s.raw).hexdigest(), host_key_fingerprint=s.snapshot.host_key_fingerprint, exit_status=0, execution_attempts=1, execution_authorized=False) @@ -161,7 +184,7 @@ def result(): def transport(*args): assert len(args) == 4 ep, cred, snap, spec = args - assert type(ep) is m._target.Stage2FixedTargetEndpoint and ep.address == "192.0.2.10" + assert type(ep) is m._target.Stage2FixedTargetEndpoint and ep == s.endpoint assert cred is s.credential and snap is s.snapshot and spec is s.command s.transport_inputs = (type(ep), type(cred), type(snap), type(spec)) return step("transport", result()) @@ -186,6 +209,8 @@ def binding(*args, **kwargs): step("binding", None) return real_binding(*args, **kwargs) + monkeypatch.setattr(m._resolver, "build_stage2_fixed_credential_resolver", + lambda: Resolver()) monkeypatch.setattr(m._root, "acquire_owner_trust_root_configuration", acquire_root) monkeypatch.setattr(m._owner, "Stage2OwnerVerifier", Verifier) monkeypatch.setattr(m._owner.Stage2ExactOwnerApprovalSource, "read_exact", deny) @@ -205,6 +230,53 @@ def binding(*args, **kwargs): return s +@pytest.fixture +def lab2_setup(setup): + s = setup + s.request = m._contract.Stage2VrrpObservationRequest( + "1.0", "mikrotik.vrrp_status", "run.synthetic", + m._target.STAGE2_SECOND_TARGET_REF, + m._resolver.STAGE2_SECOND_CREDENTIAL_REF, + "authorization.synthetic", True) + s.endpoint = s.lab2_endpoint + s.config = replace( + s.config, + known_host_configuration=m._host.Stage2KnownHostSourceConfiguration( + r"c:\synthetic\host.json", IDENTITY, "b" * 64, s.endpoint), + credential_configuration=m._windows.Stage2TrustedWindowsCredentialConfiguration( + m._resolver.STAGE2_SECOND_CREDENTIAL_LOCATOR_REF, + "synthetic-only-target")) + s.envelope = m._authorization.Stage2AuthorizationEnvelope( + "1.0", AUTH_ID, s.request.operation_id, + hashlib.sha256(s.request.to_canonical_bytes()).hexdigest(), + s.request.authorization_ref, s.request.target_ref, + s.request.credential_ref, 1000, 1300, 1) + s.verification = issue( + m._owner.Stage2VerifiedOwnerApproval, + schema_version=m._owner.APPROVAL_SCHEMA_VERSION, + artifact_sha256="c" * 64, approval_ref=s.request.authorization_ref, + verified_issuer_ref=s.root.issuer_ref, + approval_source_id=s.root.approval_source_id, + public_key_fingerprint=s.root.public_key_sha256_fingerprint, + payload_sha256=hashlib.sha256( + m._authorization.owner_verification_payload(s.envelope)).hexdigest(), + verified=True, execution_authorized=False) + s.snapshot = issue( + m._host.Stage2KnownHostSnapshot, + schema_version="s2-ro-07.known-host.v1", + target_ref=s.request.target_ref, address=s.endpoint.address, port=22, + host_key_algorithm="ssh-ed25519", + host_key_blob=s.snapshot.host_key_blob, + host_key_sha256=s.snapshot.host_key_sha256, + host_key_fingerprint=s.snapshot.host_key_fingerprint, + source_file_identity=IDENTITY, source_artifact_sha256="b" * 64, + execution_authorized=False) + s.consumed = m._authorization.Stage2ConsumptionRecord( + AUTH_ID, hashlib.sha256(s.envelope.to_canonical_bytes()).hexdigest(), 1100) + s.command = REAL_POLICY(s.request) + return s + + def rejected(call, category): with pytest.raises(m.Stage2TrustedRuntimeError) as caught: call() @@ -239,12 +311,47 @@ def test_success_exact_order_and_evidence(setup): assert evidence.raw_output_sha256 == hashlib.sha256(RAW).hexdigest() assert evidence.records[0].instance_name == "synthetic-vrrp" assert evidence.attempt_count == 1 and evidence.retry_count == 0 + assert s.resolver_inputs == [(s.request.target_ref, s.request.credential_ref)] + assert s.credential_inputs == [( + s.request.target_ref, + REAL_RESOLVER().resolve_for_target( + s.request.target_ref, s.request.credential_ref))] assert set(s.evidence_fields) == {f.name for f in fields(REAL_EVIDENCE)} assert SENTINEL.encode() not in evidence.to_canonical_bytes() assert RAW not in evidence.to_canonical_bytes() assert m._contract.parse_stage2_vrrp_evidence_canonical_json(evidence.to_canonical_bytes()) == evidence +def test_lab2_success_uses_one_exact_target_aware_path(lab2_setup): + s = lab2_setup + evidence = s.call() + binding = REAL_RESOLVER().resolve_for_target( + s.request.target_ref, s.request.credential_ref) + assert s.config.target_registry.lookup(s.request.target_ref) == s.endpoint + assert s.endpoint.target_ref == m._target.STAGE2_SECOND_TARGET_REF + assert binding.credential_ref == m._resolver.STAGE2_SECOND_CREDENTIAL_REF + assert s.resolver_inputs == [(s.request.target_ref, s.request.credential_ref)] + assert s.credential_inputs == [(s.request.target_ref, binding)] + assert s.trace == [ + "binding", "root", "verifier", "verify", "ledger", "consume", "host", + "binding", "backend", "credential", "command", "binding", "transport", + "parser", "evidence"] + assert s.trace.count("verify") == 1 + assert s.trace.count("consume") == 1 + assert s.trace.count("host") == 1 + assert s.trace.count("credential") == 1 + assert s.trace.count("command") == 1 + assert s.trace.count("transport") == 1 + assert s.command.command_text == "/interface vrrp print detail" + assert s.parser_input is s.raw + assert type(evidence) is REAL_EVIDENCE + assert evidence.target_ref == m._target.STAGE2_SECOND_TARGET_REF + assert evidence.attempt_count == 1 and evidence.retry_count == 0 + assert evidence.execution_authorized is False + assert evidence.raw_output_byte_count == len(s.raw) + assert evidence.raw_output_sha256 == hashlib.sha256(s.raw).hexdigest() + + @pytest.mark.parametrize("value", [None, {}, object(), True]) def test_request_wrong_types(setup, value): setup.request = value @@ -530,6 +637,43 @@ def test_real_replay_remains_spent_after_each_downstream_failure(setup, monkeypa assert "host" not in setup.trace and "credential" not in setup.trace and "transport" not in setup.trace +@pytest.mark.parametrize("stage,category", [ + ("host", F.KNOWN_HOST_ACQUISITION_FAILED), + ("credential", F.CREDENTIAL_ACQUISITION_FAILED), + ("command", F.COMMAND_POLICY_FAILED), + ("transport", F.TRANSPORT_FAILED), + ("parser", F.OUTPUT_PARSE_FAILED), + ("evidence", F.EVIDENCE_CONSTRUCTION_FAILED), +]) +def test_lab2_real_replay_remains_spent_after_each_downstream_failure( + lab2_setup, monkeypatch, stage, category): + s = lab2_setup + path = s.config.replay_ledger_configuration.database_path + with sqlite3.connect(path) as connection: + for ddl in m._authorization._SCHEMA: + connection.execute(ddl) + connection.execute( + "INSERT INTO ledger_metadata VALUES (?, ?)", ("1.0", INSTANCE)) + monkeypatch.setattr(m._authorization, "Stage2ReplayLedger", REAL_LEDGER) + monkeypatch.setattr(m, "_monotonic_now", lambda: 1) + s.fail[stage] = RuntimeError(SENTINEL) + rejected(s.call, category) + with sqlite3.connect(path) as connection: + assert connection.execute( + "SELECT authorization_id FROM consumed_authorizations").fetchall() == [ + (AUTH_ID,)] + assert s.trace.count("transport") <= 1 + assert s.trace.count("credential") <= 1 + assert len(s.credential_inputs) <= 1 + s.trace.clear() + s.credential_inputs.clear() + rejected(s.call, F.REPLAY_REJECTED) + assert "host" not in s.trace + assert "credential" not in s.trace + assert "transport" not in s.trace + assert s.credential_inputs == [] + + def test_request_and_nested_configuration_captured_before_external_stage(setup, monkeypatch): original = m._root.acquire_owner_trust_root_configuration # Mutate only caller-owned inputs after private capture. @@ -543,6 +687,7 @@ def mutate(**kwargs): object.__setattr__(supplied_config.target_registry._endpoint, "address", "192.0.2.99") setup.request = saved_request setup.config = saved_config + setup.endpoint = saved_config.target_registry.lookup(saved_request.target_ref) return value monkeypatch.setattr(m._root, "acquire_owner_trust_root_configuration", mutate) evidence = m.execute_stage2_vrrp_trusted_runtime( @@ -597,6 +742,91 @@ def test_unknown_credential_binding_prevents_root(setup): assert setup.trace == [] +def test_lab2_target_rejects_lab1_credential_without_fallback(lab2_setup): + s = lab2_setup + s.request = replace( + s.request, credential_ref=m._resolver.STAGE2_FIXED_CREDENTIAL_REF) + rejected(s.call, F.CREDENTIAL_BINDING_FAILED) + assert s.resolver_inputs == [( + m._target.STAGE2_SECOND_TARGET_REF, + m._resolver.STAGE2_FIXED_CREDENTIAL_REF)] + assert s.credential_inputs == [] and s.trace == [] + + +def test_lab1_target_rejects_lab2_credential_without_fallback(setup): + s = setup + s.request = replace( + s.request, credential_ref=m._resolver.STAGE2_SECOND_CREDENTIAL_REF) + rejected(s.call, F.CREDENTIAL_BINDING_FAILED) + assert s.resolver_inputs == [( + m._target.STAGE2_FIXED_TARGET_REF, + m._resolver.STAGE2_SECOND_CREDENTIAL_REF)] + assert s.credential_inputs == [] and s.trace == [] + + +@pytest.mark.parametrize("fixture_name,wrong_locator", [ + ("setup", m._resolver.STAGE2_SECOND_CREDENTIAL_LOCATOR_REF), + ("lab2_setup", m._resolver.STAGE2_CREDENTIAL_LOCATOR_REF), +]) +def test_wrong_trusted_credential_locator_rejects_without_acquisition_or_transport( + request, fixture_name, wrong_locator): + s = request.getfixturevalue(fixture_name) + s.config = replace( + s.config, + credential_configuration=m._windows.Stage2TrustedWindowsCredentialConfiguration( + wrong_locator, "synthetic-only-target")) + rejected(s.call, F.CREDENTIAL_ACQUISITION_FAILED) + assert len(s.credential_inputs) == 1 + assert s.trace.count("credential") == 0 + assert s.trace.count("consume") == 1 and s.spent + assert "transport" not in s.trace + + +def test_lab2_target_rejects_lab1_known_host_before_credentials(lab2_setup): + s = lab2_setup + lab1_endpoint = s.config.target_registry.lookup( + m._target.STAGE2_FIXED_TARGET_REF) + s.config = replace( + s.config, + known_host_configuration=m._host.Stage2KnownHostSourceConfiguration( + r"c:\synthetic\host.json", IDENTITY, "b" * 64, lab1_endpoint)) + rejected(s.call, F.TARGET_RESOLUTION_FAILED) + assert s.resolver_inputs == [] and s.credential_inputs == [] + assert s.trace == [] + + +def test_registry_known_host_endpoint_mismatch_rejects_configuration(setup): + s = setup + mismatched_endpoint = replace(s.endpoint, address="192.0.2.99") + mismatched_host = m._host.Stage2KnownHostSourceConfiguration( + r"c:\synthetic\host.json", IDENTITY, "b" * 64, mismatched_endpoint) + invalid = drift(s.config, "known_host_configuration", mismatched_host) + rejected( + lambda: m.execute_stage2_vrrp_trusted_runtime( + s.request, s.envelope.to_canonical_bytes(), invalid), + F.INVALID_TRUSTED_RUNTIME_CONFIGURATION) + assert s.resolver_inputs == [] and s.credential_inputs == [] + assert s.trace == [] + + +@pytest.mark.parametrize("field,value", [ + ("target_ref", m._target.STAGE2_FIXED_TARGET_REF), + ("credential_ref", m._resolver.STAGE2_FIXED_CREDENTIAL_REF), +]) +def test_lab2_authorization_target_or_credential_mismatch_rejects_before_root( + lab2_setup, monkeypatch, field, value): + s = lab2_setup + mismatched = drift(s.envelope, field, value) + monkeypatch.setattr( + m._authorization, "parse_stage2_authorization_envelope", + lambda raw: mismatched) + rejected( + lambda: m.execute_stage2_vrrp_trusted_runtime(s.request, b"synthetic", s.config), + F.AUTHORIZATION_ENVELOPE_FAILED) + assert s.trace == ["binding"] + assert s.credential_inputs == [] + + @pytest.mark.parametrize("point,category", [("utc", F.AUTHORIZATION_ENVELOPE_FAILED), ("monotonic", F.INTERNAL_FAILURE)]) def test_failed_system_clock_has_no_fallback(setup, monkeypatch, point, category): def failure(): diff --git a/validation_framework/stage2_trusted_runtime_composition.py b/validation_framework/stage2_trusted_runtime_composition.py index eb4ef95..3054f55 100644 --- a/validation_framework/stage2_trusted_runtime_composition.py +++ b/validation_framework/stage2_trusted_runtime_composition.py @@ -115,7 +115,11 @@ def _configuration_values(configuration): registry = configuration.target_registry _exact(registry, _target.Stage2FixedTargetRegistry) endpoint = _capture(registry._endpoint, _target.Stage2FixedTargetEndpoint) - registry = _target.Stage2FixedTargetRegistry(endpoint) + lab2_endpoint = None + if registry._lab2_endpoint is not None: + lab2_endpoint = _capture( + registry._lab2_endpoint, _target.Stage2FixedTargetEndpoint) + registry = _target.Stage2FixedTargetRegistry(endpoint, lab2_endpoint) path = configuration.owner_trust_root_expected_path identity = configuration.owner_trust_root_expected_file_identity digest = configuration.owner_trust_root_expected_file_sha256 @@ -131,7 +135,7 @@ def _configuration_values(configuration): host = _host.Stage2KnownHostSourceConfiguration( host.expected_path, host.expected_file_identity, host.expected_file_sha256, _capture(host.expected_endpoint, _target.Stage2FixedTargetEndpoint)) - _require(host.expected_endpoint == endpoint) + _require(registry.lookup(host.expected_endpoint.target_ref) == host.expected_endpoint) credential = _capture(configuration.credential_configuration, _windows.Stage2TrustedWindowsCredentialConfiguration) return registry, path, identity, digest, ledger, host, credential @@ -261,9 +265,12 @@ def _run(request, raw_envelope, configuration): stage = _F.TARGET_RESOLUTION_FAILED endpoint = _capture(configuration.target_registry.lookup(request.target_ref), _target.Stage2FixedTargetEndpoint) + _require(configuration.known_host_configuration.expected_endpoint == endpoint) stage = _F.CREDENTIAL_BINDING_FAILED - binding = _capture(_resolver.build_stage2_fixed_credential_resolver().resolve(request.credential_ref), - _resolver.Stage2CredentialBinding) + binding = _capture( + _resolver.build_stage2_fixed_credential_resolver().resolve_for_target( + request.target_ref, request.credential_ref), + _resolver.Stage2CredentialBinding) stage = _F.AUTHORIZATION_ENVELOPE_FAILED _authorization.validate_stage2_authorization_binding( envelope, request, configuration.target_registry, binding, now=_utc_now()) @@ -289,7 +296,8 @@ def _run(request, raw_envelope, configuration): stage = _F.CREDENTIAL_ACQUISITION_FAILED try: credential = _windows.build_stage2_windows_credential_backend( - configuration.credential_configuration).read(binding) + configuration.credential_configuration).read_for_target( + request.target_ref, binding) _validate_credential(credential) stage = _F.COMMAND_POLICY_FAILED command = _policy.resolve_stage2_vrrp_readonly_command(request)