From 44aead105fe3e36f438eb99fc05fb8b6b16cd2c6 Mon Sep 17 00:00:00 2001 From: RobinLee Date: Sat, 12 Sep 2026 14:47:50 +0800 Subject: [PATCH] feat(stage2): bind Lab2 authorization before replay --- ..._s2_ro_05_authorization_envelope_ledger.md | 77 ++++- .../test_authorization_envelope_ledger.py | 280 +++++++++++++++++- .../stage2_authorization_envelope_ledger.py | 20 +- 3 files changed, 357 insertions(+), 20 deletions(-) diff --git a/docs/automation_readiness/stage2_vrrp_readonly_s2_ro_05_authorization_envelope_ledger.md b/docs/automation_readiness/stage2_vrrp_readonly_s2_ro_05_authorization_envelope_ledger.md index ffce51e..3c16682 100644 --- a/docs/automation_readiness/stage2_vrrp_readonly_s2_ro_05_authorization_envelope_ledger.md +++ b/docs/automation_readiness/stage2_vrrp_readonly_s2_ro_05_authorization_envelope_ledger.md @@ -2,10 +2,13 @@ ## Decision summary -S2-RO-05 provides strict offline authorization data and permanent local replay -consumption. It does not authenticate the Owner or authorize execution. -Status: uncommitted implementation candidate, subject to validation and -independent review. No earlier slice is modified. +S2-RO-05 accepts exactly the Lab1/Lab1 and Lab2/Lab2 target/credential pairs +through the existing S2-RO-03 resolver. Mixed or noncanonical bindings reject +before replay filesystem access, path identity checks, or SQLite connection. +The durable replay engine and SQL schema are unchanged. This is an offline +binding-policy extension, not Owner authentication or execution authorization. +Status: bounded implementation candidate, subject to validation and independent +review. No earlier slice or S2-RO-06-or-later production module is modified. `VALID ENVELOPE != OWNER APPROVAL != EXECUTION AUTHORITY` @@ -30,8 +33,8 @@ No new dependency or CLI registration is added. | operation_id | Exact `mikrotik.vrrp_status` (20 characters) | | request_sha256 | 64 lowercase hexadecimal characters | | authorization_ref | Dotted lowercase ASCII reference beginning `authorization.`, maximum 160 characters | -| target_ref | Exact `target.mikrotik.lab01` (20 characters) | -| credential_ref | Exact `credential.mikrotik.lab01` (24 characters) | +| target_ref | Exact `target.mikrotik.lab01` or `target.mikrotik.lab02`, subject to the pair rule below | +| credential_ref | Exact matching `credential.mikrotik.lab01` or `credential.mikrotik.lab02` | | issued_at | Integer UTC Unix seconds, 0 through 253402300799 | | expires_at | Integer UTC Unix seconds, same bound | | max_attempts | Exact integer 1 | @@ -40,12 +43,37 @@ Boolean values are not integers for this contract. Reference segments match `[a-z][a-z0-9_-]*`, separated by dots. No normalization, whitespace trimming, coercion, optional fields, or defaults are applied. +Only these conceptual pairs are accepted: + +- `target.mikrotik.lab01` + `credential.mikrotik.lab01`. +- `target.mikrotik.lab02` + `credential.mikrotik.lab02`. + +S2-RO-03 `resolve_for_target(target_ref, credential_ref)` is the pair authority; +S2-RO-05 has no duplicate pair registry. Both cross-lab pairs, unknown identities, +aliases, prefixes, case-confused values, whitespace changes, and noncanonical +types reject. There is no wildcard, normalization, fallback, or default-to-Lab1. +Historical Lab1 canonical bytes remain valid under the same schema `1.0`. + The request digest is SHA-256 of S2-RO-01 `to_canonical_bytes()`. It binds all request fields including run identity and read-only intent. Explicit operation, authorization, target and credential references must also match. S2-RO-02 lookup and the S2-RO-03 binding remain immutable authorities; S2-RO-04 is neither imported nor called. Logical target binding is not physical-device attestation. +Binding validation checks exact envelope/request/registry/binding object types, +revalidates the envelope, reparses the request, looks up and validates its +registry endpoint, and resolves the validated target/credential pair through +S2-RO-03. The supplied binding must equal that canonical result. Operation, +authorization reference, endpoint target, credential, and exact canonical +request digest must all match before time validation. Caller-supplied bindings +cannot override the resolver. S2-RO-02 must be configured with the exact pair +for a Lab2 lookup; this extension does not discover or provision endpoints. + +`consume` still performs binding validation before `_path_identity` and +`sqlite3.connect`. Invalid bindings have zero ledger filesystem access, path +identity calls, connections, BEGINs, INSERTs, and durable mutations. Both labs +follow the same path and bounded failure categories; neither has a relaxed path. + The envelope is a frozen, slotted dataclass of immutable scalars. Exported dictionaries are fresh copies. Parsing accepts only built-in bytes, at most 2048 bytes, strict UTF-8 and exactly ten fields. Duplicate keys, unknown or @@ -128,6 +156,12 @@ The replay key is authorization UUID alone. The envelope SHA-256 is audit metadata, not a second replay namespace. Same-ID/different-envelope replay is rejected. No raw envelope, request, endpoint, credential, or command is stored. +Lab1 and Lab2 share this replay-key meaning: changing the lab or any envelope +field does not make an already consumed authorization UUID reusable. Only +temporary pre-provisioned synthetic ledgers are used to validate this extension. +No real replay database, rows, ledger identity, authorization package, credential +store, Owner trust-root artifact, or private key is accessed or provisioned. + Records are retained permanently. The limit is 100000 records and 64 MiB of database pages. Capacity exhaustion blocks; it never prunes or resets history. @@ -199,11 +233,26 @@ fresh-process replay and concurrent consumers, malformed schema/storage, missing storage, busy handling, injected write and ambiguous-commit failures, post-commit expiry, sanitized errors, and absent forbidden capabilities. -Run focused tests, S2-RO-01 through S2-RO-04 regressions, full `python -m pytest`, -and `python network_lab.py --task report-index`. Review all three candidate -files independently, including documentation readability and no-execution -boundaries. Validation must not modify existing tracked files. Optional missing -runtime reports may produce a documented policy-accepted report-index WARN. - -Passing validation and review supports only local commit authorization. This -slice does not itself authorize commit, push, PR, merge, or S2-RO-06 work. +The Lab2 extension adds exact-pair round trips, legacy Lab1 compatibility, +resolver-authority checks, cross-lab and malformed-input pre-I/O guards, and +synthetic consume-once/replay cases for both labs. Guards observe zero filesystem, +path identity, SQLite connect, BEGIN, INSERT, and mutation counts on rejected +bindings; test-ledger content, size, mtime, and directory entries remain intact. +Valid consumption uses one exact connection and one parameterized INSERT, with +no reconnect, retry, fallback ledger, or new replay namespace. Owner payload +domain bytes and both `execution_authorized=False` contracts are preserved. + +Validate in this order using a disposable external copy and writable external +pytest basetemps: focused S2-RO-05, all `tests/stage2`, full pytest, report-index, +then `git diff --check`. Python uses `-B`; pytest uses `-p no:cacheprovider` +and `--tb=short`. Never print environment mappings or retain unsanitized +tracebacks. All tests require zero failures; unchanged platform skips are +acceptable. Optional missing reports may remain accepted WARN without a +mandatory failure. No dependency installation or download is part of this task. +Source HEAD/tree/status and content/size/mtime must remain unchanged during +external validation; only the three authorized candidate files are applied. + +The separately bounded implementation authorization permits one local commit +only after every mandatory gate passes. Independent review remains a separate +gate. Push, PR, merge, cleanup, S2-RO-06-or-later Lab2 extension, real authorization +or replay operations, Lab1/Lab2 contact, and Stage 3 require separate approval. diff --git a/tests/stage2/test_authorization_envelope_ledger.py b/tests/stage2/test_authorization_envelope_ledger.py index 44be15c..e222776 100644 --- a/tests/stage2/test_authorization_envelope_ledger.py +++ b/tests/stage2/test_authorization_envelope_ledger.py @@ -1,7 +1,7 @@ """Synthetic offline S2-RO-05 evidence. No production ledger or credentials.""" import ast -from dataclasses import FrozenInstanceError, replace +from dataclasses import FrozenInstanceError, fields, replace import hashlib import json import multiprocessing @@ -401,3 +401,281 @@ def test_valid_snapshot_rollback_is_explicitly_not_protected(ledger): assert consume(ledger) doc = Path(__file__).resolve().parents[2] / "docs/automation_readiness/stage2_vrrp_readonly_s2_ro_05_authorization_envelope_ledger.md" assert "VALID_SNAPSHOT_ROLLBACK_PROTECTION = OUT_OF_SCOPE" in doc.read_text(encoding="utf-8") + + +# Synthetic identities only. The endpoint is derived from the existing TEST-NET +# fixture; these tests never connect to an endpoint or inspect a real ledger. +def pair_inputs(lab): + args = inputs() + if lab == 1: + return args + envelope, request, registry, _ = args + request = replace(request, target_ref="target.mikrotik.lab02", + credential_ref="credential.mikrotik.lab02") + envelope = replace(envelope, target_ref=request.target_ref, + credential_ref=request.credential_ref, + request_sha256=hashlib.sha256(request.to_canonical_bytes()).hexdigest()) + first = registry.lookup("target.mikrotik.lab01") + second = replace(first, target_ref=request.target_ref, address=first.address[:-1] + "1") + registry = replace(registry, _lab2_endpoint=second) + binding = build_stage2_fixed_credential_resolver().resolve_for_target( + request.target_ref, request.credential_ref) + return envelope, request, registry, binding + + +def tampered(instance, **changes): + """Bypass constructors only to prove consumption revalidates hostile data.""" + result = object.__new__(type(instance)) + for field in fields(instance): + object.__setattr__(result, field.name, changes.get(field.name, getattr(instance, field.name))) + return result + + +class ReferenceSubclass(str): + pass + + +def confused_refs(reference): + return (None, {}, "", reference.upper(), reference + ".alias", reference[:-1], + reference + "*", " " + reference, reference + " ", reference + "\n", + reference + "\x00", reference.replace("lab0", "lab"), + ReferenceSubclass(reference), reference.replace("mikrotik", "mikrotіk")) + + +@pytest.mark.parametrize("lab", [1, 2]) +def test_exact_pair_roundtrip_binding_domain_and_no_authority(lab): + envelope, request, registry, binding = args = pair_inputs(lab) + raw = envelope.to_canonical_bytes() + assert m.parse_stage2_authorization_envelope(raw) == envelope + assert m.parse_stage2_authorization_envelope(raw).to_canonical_bytes() == raw + assert raw == json.dumps(envelope.to_dict(), sort_keys=True, separators=(",", ":"), + ensure_ascii=False, allow_nan=False).encode("utf-8") + assert m.validate_stage2_authorization_binding(*args, now=1100) is None + assert m.SCHEMA_VERSION == "1.0" + domain = b"Network_Automation_Lab/S2-RO-05/authorization-envelope/v1\x00" + assert m.OWNER_PAYLOAD_DOMAIN == domain + assert m.owner_verification_payload(envelope) == domain + raw + assert envelope.execution_authorized is False + assert not hasattr(envelope, "__dict__") + with pytest.raises(FrozenInstanceError): + envelope.target_ref = "changed" + assert binding == build_stage2_fixed_credential_resolver().resolve_for_target( + request.target_ref, request.credential_ref) + assert registry.lookup(request.target_ref).target_ref == envelope.target_ref + + +@pytest.mark.parametrize("lab", [1, 2]) +@pytest.mark.parametrize("field", ["target_ref", "credential_ref"]) +def test_envelope_rejects_mixed_unknown_and_noncanonical_pairs(lab, field): + envelope = pair_inputs(lab)[0] + other = pair_inputs(3 - lab)[0] + original = getattr(envelope, field) + for value in (*confused_refs(original), getattr(other, field), original.replace(f"lab0{lab}", "lab03")): + with pytest.raises(m.Stage2AuthorizationError) as caught: + replace(envelope, **{field: value}) + assert str(caught.value) == "INVALID_ENVELOPE" + assert caught.value.__context__ is None + assert caught.value.__cause__ is None + # JSON cannot retain a str subclass, so test that case at object ingress. + if type(value) is not ReferenceSubclass: + with pytest.raises(m.Stage2AuthorizationError, match="INVALID_ENVELOPE"): + m.parse_stage2_authorization_envelope(encode(dict(envelope.to_dict(), **{field: value}))) + + +@pytest.mark.parametrize("lab", [1, 2]) +def test_pair_envelopes_keep_strict_parser_and_scalar_invariants(lab): + envelope = pair_inputs(lab)[0] + record, raw = envelope.to_dict(), envelope.to_canonical_bytes() + invalid_values = { + "schema_version": ("2.0", True), "authorization_id": (AUTH_ID.upper(), "not-a-uuid"), + "operation_id": ("other", None), "request_sha256": ("A" * 64, "0" * 63), + "authorization_ref": (" authorization.demo", "authorization." + "x" * 160), + "issued_at": (True, -1, 1000.0), "expires_at": (1000, 1301, m.MAX_UNIX_SECONDS + 1), + "max_attempts": (True, 0, 2), + } + for field, values in invalid_values.items(): + for value in values: + with pytest.raises(m.Stage2AuthorizationError, match="INVALID_ENVELOPE"): + m.parse_stage2_authorization_envelope(encode(dict(record, **{field: value}))) + for field in record: + missing = dict(record) + del missing[field] + with pytest.raises(m.Stage2AuthorizationError, match="INVALID_ENVELOPE"): + m.parse_stage2_authorization_envelope(encode(missing)) + for candidate in (raw + b"\n", b" " + raw, b"\xef\xbb\xbf" + raw, + json.dumps(record, indent=2).encode(), encode(dict(record, extra=1)), + raw.replace(b'"schema_version":"1.0"', b'"schema_version":"1.0","schema_version":"1.0"'), + raw.replace(b"mikrotik", b"mikrot\\u0069k"), raw + b"\xff"): + with pytest.raises(m.Stage2AuthorizationError, match="INVALID_ENVELOPE"): + m.parse_stage2_authorization_envelope(candidate) + + +_MISMATCHES = ( + "other_envelope", "other_binding", "request_cross_target", "request_cross_credential", + "envelope_cross_target", "envelope_cross_credential", "digest", "authorization", + "operation", "envelope_operation", "all_unknown", "binding_locator", "binding_backend", + "binding_type", "registry_type", "envelope_type", "request_type", +) + + +def invalid_args(lab, case): + args, other = list(pair_inputs(lab)), pair_inputs(3 - lab) + if case == "other_envelope": args[0] = other[0] + elif case == "other_binding": args[3] = other[3] + elif case.startswith("request_cross_"): + field = "target_ref" if case.endswith("target") else "credential_ref" + args[1] = tampered(args[1], **{field: getattr(other[1], field)}) + elif case.startswith("envelope_cross_"): + field = "target_ref" if case.endswith("target") else "credential_ref" + args[0] = tampered(args[0], **{field: getattr(other[0], field)}) + elif case == "digest": args[0] = replace(args[0], request_sha256="0" * 64) + elif case == "authorization": args[0] = replace(args[0], authorization_ref="authorization.other") + elif case == "operation": args[1] = tampered(args[1], operation_id="operation.other") + elif case == "envelope_operation": args[0] = tampered(args[0], operation_id="operation.other") + elif case == "all_unknown": + for index in (0, 1): + args[index] = tampered(args[index], target_ref="target.unknown.lab", + credential_ref="credential.unknown.lab") + args[3] = tampered(args[3], credential_ref="credential.unknown.lab", locator_ref="locator.unknown.lab") + elif case == "binding_locator": args[3] = tampered(args[3], locator_ref=other[3].locator_ref) + elif case == "binding_backend": args[3] = tampered(args[3], backend_kind="OTHER") + else: args[{"binding_type": 3, "registry_type": 2, "envelope_type": 0, "request_type": 1}[case]] = object() + return args + + +def assert_pre_io_rejection(ledger, args, monkeypatch): + path = ledger._configuration.database_path + before = path.read_bytes() + before_stat = path.stat() + before_entries = sorted(item.name for item in path.parent.iterdir()) + counts = dict(path_identity=0, filesystem=0, connect=0, begin=0, insert=0, mutation=0) + + def filesystem(*a, **k): + counts["filesystem"] += 1 + raise AssertionError("invalid binding reached filesystem") + + def path_identity(*a, **k): + counts["path_identity"] += 1 + raise AssertionError("invalid binding reached path identity") + + class DeniedConnection: + in_transaction = False + + def execute(self, sql, *a): + if sql.startswith("BEGIN"): counts["begin"] += 1 + if sql.startswith("INSERT"): counts["insert"] += 1 + raise AssertionError("invalid binding reached SQL") + + def commit(self): + counts["mutation"] += 1 + raise AssertionError("invalid binding reached commit") + + def close(self): pass + + def connect(*a, **k): + counts["connect"] += 1 + return DeniedConnection() + + with monkeypatch.context() as guard: + guard.setattr(m, "_path_identity", path_identity) + guard.setattr(m.sqlite3, "connect", connect) + for name in ("resolve", "stat", "lstat", "open", "read_bytes", "write_bytes", "mkdir"): + guard.setattr(Path, name, filesystem) + with pytest.raises(m.Stage2AuthorizationError) as caught: + consume(ledger, args) + assert caught.value.code is m.Stage2AuthorizationFailure.INVALID_BINDING + assert str(caught.value) == "INVALID_BINDING" + assert caught.value.__context__ is None + assert caught.value.__cause__ is None + assert counts == dict(path_identity=0, filesystem=0, connect=0, begin=0, insert=0, mutation=0) + assert path.read_bytes() == before + after_stat = path.stat() + assert (after_stat.st_size, after_stat.st_mtime_ns) == (before_stat.st_size, before_stat.st_mtime_ns) + assert sorted(item.name for item in path.parent.iterdir()) == before_entries + + +@pytest.mark.parametrize("lab", [1, 2]) +@pytest.mark.parametrize("case", _MISMATCHES) +def test_pair_and_binding_mismatches_have_zero_ledger_io(lab, case, ledger, monkeypatch): + assert_pre_io_rejection(ledger, invalid_args(lab, case), monkeypatch) + + +@pytest.mark.parametrize("lab", [1, 2]) +@pytest.mark.parametrize("index,field", [(0, "target_ref"), (0, "credential_ref"), + (1, "target_ref"), (1, "credential_ref"), (3, "credential_ref")]) +def test_unknown_alias_prefix_case_and_noncanonical_inputs_have_zero_io(lab, index, field, ledger, monkeypatch): + original = pair_inputs(lab) + reference = getattr(original[index], field) + for value in (*confused_refs(reference), reference.replace(f"lab0{lab}", "lab03")): + args = list(original) + args[index] = tampered(args[index], **{field: value}) + assert_pre_io_rejection(ledger, args, monkeypatch) + + +@pytest.mark.parametrize("lab", [1, 2]) +def test_valid_pairs_consume_once_and_share_uuid_replay_key(lab, ledger, monkeypatch): + args = pair_inputs(lab) + real_connect = sqlite3.connect + expected_uri = ledger._configuration.database_path.as_uri() + "?mode=rw" + calls, statements, commits = [], [], [] + + class ObservedConnection: + def __init__(self, db): self.db = db + def __getattr__(self, name): return getattr(self.db, name) + def execute(self, sql, *parameters): + statements.append((sql, parameters)) + return self.db.execute(sql, *parameters) + def commit(self): + commits.append(1) + return self.db.commit() + + def connect(database, **options): + calls.append((database, options)) + assert database == expected_uri + assert options == dict(uri=True, timeout=0, isolation_level=None) + return ObservedConnection(real_connect(database, **options)) + + with monkeypatch.context() as guard: + guard.setattr(m.sqlite3, "connect", connect) + result = consume(ledger, args) + assert len(calls) == 1 + assert [sql for sql, _ in statements].count("BEGIN IMMEDIATE") == 1 + inserts = [(sql, params) for sql, params in statements if sql.startswith("INSERT")] + assert len(inserts) == 1 + assert inserts[0][0] == "INSERT INTO consumed_authorizations (authorization_id, envelope_sha256, consumed_at) VALUES (?, ?, ?)" + assert inserts[0][1] == ((AUTH_ID, hashlib.sha256(args[0].to_canonical_bytes()).hexdigest(), 1100),) + assert commits == [1] + assert result.execution_authorized is False + for repeated in (args, (replace(args[0], issued_at=1001), *args[1:]), pair_inputs(3 - lab)): + before_calls = len(calls) + with pytest.raises(m.Stage2AuthorizationError, match="REPLAY"): + consume(ledger, repeated) + assert len(calls) == before_calls + 1 + assert commits == [1] + with real_connect(ledger._configuration.database_path) as db: + assert db.execute("SELECT * FROM consumed_authorizations").fetchall() == [ + (AUTH_ID, result.envelope_sha256, 1100)] + + +@pytest.mark.parametrize("lab", [1, 2]) +def test_envelope_and_binding_reuse_resolver_authority_before_time(monkeypatch, lab): + args = pair_inputs(lab) + resolver_type = type(build_stage2_fixed_credential_resolver()) + original = resolver_type.resolve_for_target + calls = [] + + def observed(self, target_ref, credential_ref): + calls.append((target_ref, credential_ref)) + return original(self, target_ref, credential_ref) + + with monkeypatch.context() as guard: + guard.setattr(resolver_type, "resolve_for_target", observed) + args[0].__post_init__() + assert calls == [(args[1].target_ref, args[1].credential_ref)] + calls.clear() + with monkeypatch.context() as guard: + guard.setattr(resolver_type, "resolve_for_target", observed) + m.validate_stage2_authorization_binding(*args, now=1100) + assert calls == [(args[1].target_ref, args[1].credential_ref)] * 2 + with pytest.raises(m.Stage2AuthorizationError, match="INVALID_BINDING"): + m.validate_stage2_authorization_binding(*invalid_args(lab, "other_binding"), now=-1) diff --git a/validation_framework/stage2_authorization_envelope_ledger.py b/validation_framework/stage2_authorization_envelope_ledger.py index 67c66d6..2a83642 100644 --- a/validation_framework/stage2_authorization_envelope_ledger.py +++ b/validation_framework/stage2_authorization_envelope_ledger.py @@ -26,7 +26,7 @@ ) from validation_framework.stage2_mikrotik_credential_resolver import ( Stage2CredentialBinding, STAGE2_FIXED_CREDENTIAL_REF, - build_stage2_fixed_credential_resolver, + Stage2CredentialResolverError, build_stage2_fixed_credential_resolver, ) @@ -106,9 +106,7 @@ class Stage2AuthorizationEnvelope: def __post_init__(self): exact = ((self.schema_version, SCHEMA_VERSION), - (self.operation_id, VRRP_OBSERVATION_OPERATION_ID), - (self.target_ref, STAGE2_FIXED_TARGET_REF), - (self.credential_ref, STAGE2_FIXED_CREDENTIAL_REF)) + (self.operation_id, VRRP_OBSERVATION_OPERATION_ID)) if (any(type(value) is not str or value != expected for value, expected in exact) or not _uuid(self.authorization_id) or not _digest(self.request_sha256) or type(self.authorization_ref) is not str @@ -119,6 +117,16 @@ def __post_init__(self): or type(self.max_attempts) is not int or self.max_attempts != 1): _fail(Stage2AuthorizationFailure.INVALID_ENVELOPE) + # S2-RO-03 owns the exact pair policy; this check performs no I/O. + invalid_pair = False + try: + build_stage2_fixed_credential_resolver().resolve_for_target( + self.target_ref, self.credential_ref) + except Stage2CredentialResolverError: + invalid_pair = True + if invalid_pair: + _fail(Stage2AuthorizationFailure.INVALID_ENVELOPE) + def __repr__(self): return "Stage2AuthorizationEnvelope()" @@ -186,7 +194,9 @@ def validate_stage2_authorization_binding(envelope, request, registry, credentia validated_request = parse_stage2_vrrp_observation_request(request.to_dict()) endpoint = registry.lookup(validated_request.target_ref) endpoint.__post_init__() - expected_binding = build_stage2_fixed_credential_resolver().resolve(request.credential_ref) + expected_binding = build_stage2_fixed_credential_resolver().resolve_for_target( + validated_request.target_ref, validated_request.credential_ref) + credential_binding.__post_init__() if (credential_binding != expected_binding or envelope.operation_id != request.operation_id or envelope.authorization_ref != request.authorization_ref