fix: bind recipe identity to certified artifact bytes (DGR-025)
Append +artifact.<sha256> to the llama.cpp runtime axis, computed from the exact bytes read by attest_loaded_runtime, so a differently-built shared object with copied lock values can no longer forge a certified runtime identity. Node/tracker parsers require the suffix; new test proves a byte-identical-lock but different-binary artifact produces a different recipe fingerprint. Regenerates conformance vectors accordingly. 105 passed in tests/test_native_identity_emission.py, tests/test_runtime_pin_identity.py, tests/test_runtime_recipe_identity.py.
This commit is contained in:
Dobromir Popov
@@ -1,30 +1,137 @@
|
||||
"""DGR-003 production-native identity emission boundary tests."""
|
||||
"""DGR-003 production-native identity emission boundary tests.
|
||||
|
||||
The executing-runtime attestation must be *extracted from the native artifact
|
||||
itself* — a Python object holding values copied out of the world-readable
|
||||
repository lock proves nothing and must not pass. These tests compile real
|
||||
(tiny) shared objects that embed the attestation marker and export the
|
||||
attestation symbol, then prove the positive path works and that every
|
||||
lock-copying forgery path fails closed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import hashlib
|
||||
import shutil
|
||||
import subprocess
|
||||
|
||||
import pytest
|
||||
|
||||
from meshnet_node.doctor import DoctorSelection, validate_loaded_backend
|
||||
from meshnet_node.native_backend import (
|
||||
ATTESTATION_MARKER_PREFIX,
|
||||
ATTESTATION_SYMBOL,
|
||||
ImmutableArtifactPin,
|
||||
NativeArtifactEvidence,
|
||||
NativeIdentityInputs,
|
||||
NativeLoadedArtifactReport,
|
||||
NativeNumericalRecipe,
|
||||
NativeRuntimeAttestation,
|
||||
NativeSessionRejected,
|
||||
NativeWorkerBackendAdapter,
|
||||
attest_loaded_runtime,
|
||||
attestation_payload,
|
||||
expected_attestation_payload,
|
||||
shard_identity_from_native_report,
|
||||
)
|
||||
from meshnet_node.native_protocol import SCHEMA_VERSION, pb
|
||||
from meshnet_node.runtime_pin import load_runtime_pin
|
||||
from meshnet_node.recipe_manifest import parse_recipe_manifest
|
||||
from meshnet_node.runtime_pin import load_runtime_pin
|
||||
from meshnet_node.runtime_recipe import RecipeIdentityError, tokenizer_identity
|
||||
from meshnet_tracker.capability import STATE_UNCERTIFIED, evaluate_report
|
||||
|
||||
CC = shutil.which("cc")
|
||||
requires_cc = pytest.mark.skipif(
|
||||
CC is None, reason="no C compiler to build a native attestation fixture"
|
||||
)
|
||||
|
||||
|
||||
def _digest(letter: str) -> str:
|
||||
return letter * 64
|
||||
|
||||
|
||||
def _inputs(**changes: object) -> NativeIdentityInputs:
|
||||
def _c_literal(data: bytes) -> str:
|
||||
# Every byte as \xNN; the next escape's backslash terminates each escape.
|
||||
return '"' + "".join(f"\\x{b:02x}" for b in data) + '"'
|
||||
|
||||
|
||||
def _build_native_artifact(
|
||||
directory,
|
||||
payload: bytes | None,
|
||||
*,
|
||||
export_symbol: bool = True,
|
||||
symbol_returns: bytes | None = None,
|
||||
extra_payloads: tuple[bytes, ...] = (),
|
||||
):
|
||||
"""Compile a real shared object carrying the requested attestation shape."""
|
||||
lines = []
|
||||
if payload is not None:
|
||||
marker = ATTESTATION_MARKER_PREFIX + payload
|
||||
lines.append(
|
||||
"__attribute__((used)) const char marker[] = " f"{_c_literal(marker)};"
|
||||
)
|
||||
for index, extra in enumerate(extra_payloads):
|
||||
lines.append(
|
||||
f"__attribute__((used)) const char extra{index}[] = "
|
||||
f"{_c_literal(ATTESTATION_MARKER_PREFIX + extra)};"
|
||||
)
|
||||
if export_symbol:
|
||||
if symbol_returns is None:
|
||||
body = "return marker;"
|
||||
else:
|
||||
lines.append(
|
||||
"__attribute__((used)) const char other[] = "
|
||||
f"{_c_literal(symbol_returns)};"
|
||||
)
|
||||
body = "return other;"
|
||||
lines.append(f"const char *{ATTESTATION_SYMBOL}(void) {{ {body} }}")
|
||||
source = directory / "attesting_runtime.c"
|
||||
source.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||||
artifact = directory / "libattesting_runtime.so"
|
||||
subprocess.run(
|
||||
[CC, "-shared", "-fPIC", "-O0", "-o", str(artifact), str(source)],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
)
|
||||
return artifact
|
||||
|
||||
|
||||
def _payload(**overrides: object) -> bytes:
|
||||
pin = load_runtime_pin()
|
||||
values: dict[str, object] = {
|
||||
"runtime_name": pin.runtime_name,
|
||||
"upstream_commit": pin.upstream_commit,
|
||||
"patched_tree": pin.patched_tree,
|
||||
"patch_stack_digest": pin.patch_stack_digest,
|
||||
"build_recipe_digest": pin.build_recipe_digest,
|
||||
"boundary_schema_version": 1,
|
||||
"protocol_schema_version": int(SCHEMA_VERSION),
|
||||
}
|
||||
values.update(overrides)
|
||||
return attestation_payload(**values) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def _forged_attestation(directory, **overrides: object) -> NativeRuntimeAttestation:
|
||||
"""A self-consistent native artifact whose embedded values are wrong."""
|
||||
return attest_loaded_runtime(_build_native_artifact(directory, _payload(**overrides)))
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def genuine_artifact(tmp_path_factory):
|
||||
if CC is None:
|
||||
pytest.skip("no C compiler to build a native attestation fixture")
|
||||
directory = tmp_path_factory.mktemp("genuine-runtime")
|
||||
return _build_native_artifact(
|
||||
directory, expected_attestation_payload(load_runtime_pin())
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def genuine_attestation(genuine_artifact):
|
||||
return attest_loaded_runtime(genuine_artifact)
|
||||
|
||||
|
||||
def _inputs(attestation: NativeRuntimeAttestation, **changes: object) -> NativeIdentityInputs:
|
||||
report = NativeLoadedArtifactReport(
|
||||
owned_start_layer=2,
|
||||
owned_end_layer=6,
|
||||
@@ -34,6 +141,7 @@ def _inputs(**changes: object) -> NativeIdentityInputs:
|
||||
architecture="llama",
|
||||
architecture_digest=_digest("a"),
|
||||
layer_count=8,
|
||||
runtime_attestation=attestation,
|
||||
)
|
||||
recipe = NativeNumericalRecipe(
|
||||
weight_quantization="Q4_K_M",
|
||||
@@ -54,13 +162,21 @@ def _inputs(**changes: object) -> NativeIdentityInputs:
|
||||
revision="0123456789abcdef",
|
||||
content_digest=_digest("b"),
|
||||
),
|
||||
"tokenizer_revision": "abcdef0123456789",
|
||||
"tokenizer_revision": tokenizer_identity(
|
||||
{"tokenizer.json": b'{"vocab":{"a":0}}\n'}
|
||||
),
|
||||
"numerical_recipe": recipe,
|
||||
}
|
||||
values.update(changes)
|
||||
return NativeIdentityInputs(**values) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def _report_with(attestation: NativeRuntimeAttestation) -> NativeLoadedArtifactReport:
|
||||
return NativeLoadedArtifactReport(
|
||||
2, 6, 1024, 768, 640, "llama", _digest("a"), 8, attestation
|
||||
)
|
||||
|
||||
|
||||
def _open(adapter: NativeWorkerBackendAdapter, **changes: object) -> pb.SessionOpen:
|
||||
identity = adapter.identity
|
||||
fields: dict[str, object] = {
|
||||
@@ -78,26 +194,217 @@ def _open(adapter: NativeWorkerBackendAdapter, **changes: object) -> pb.SessionO
|
||||
return pb.SessionOpen(**fields) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_native_identity_uses_loaded_report_not_a_caller_range():
|
||||
identity = shard_identity_from_native_report(_inputs())
|
||||
# --- copied lock values alone must never pass ------------------------------
|
||||
|
||||
|
||||
def test_copied_lock_values_cannot_author_an_attestation():
|
||||
pin = load_runtime_pin()
|
||||
values: dict[str, object] = {
|
||||
"runtime_name": pin.runtime_name,
|
||||
"upstream_commit": pin.upstream_commit,
|
||||
"patched_tree": pin.patched_tree,
|
||||
"patch_stack_digest": pin.patch_stack_digest,
|
||||
"build_recipe_digest": pin.build_recipe_digest,
|
||||
"boundary_schema_version": 1,
|
||||
"protocol_schema_version": int(SCHEMA_VERSION),
|
||||
}
|
||||
# The pre-repair forgery: a bare self-report of lock values.
|
||||
with pytest.raises(TypeError):
|
||||
NativeRuntimeAttestation(**values) # type: ignore[arg-type]
|
||||
with pytest.raises(RecipeIdentityError, match="attest_loaded_runtime"):
|
||||
NativeRuntimeAttestation(evidence=None, **values) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_native_artifact_evidence_cannot_be_authored_in_python():
|
||||
with pytest.raises(RecipeIdentityError, match="attest_loaded_runtime"):
|
||||
NativeArtifactEvidence("lib.so", _digest("a"), _digest("b"))
|
||||
with pytest.raises(RecipeIdentityError, match="attest_loaded_runtime"):
|
||||
NativeArtifactEvidence("lib.so", _digest("a"), _digest("b"), object())
|
||||
|
||||
|
||||
def test_marker_bytes_in_a_plain_file_are_not_an_executing_runtime(tmp_path):
|
||||
fake = tmp_path / "fake.so"
|
||||
fake.write_bytes(ATTESTATION_MARKER_PREFIX + _payload() + b"\x00")
|
||||
with pytest.raises(RecipeIdentityError, match="not a loadable"):
|
||||
attest_loaded_runtime(fake)
|
||||
|
||||
|
||||
def test_missing_native_artifact_fails_closed(tmp_path):
|
||||
with pytest.raises(RecipeIdentityError, match="not found"):
|
||||
attest_loaded_runtime(tmp_path / "never-built.so")
|
||||
|
||||
|
||||
@requires_cc
|
||||
def test_artifact_without_a_marker_fails_closed(tmp_path):
|
||||
artifact = _build_native_artifact(tmp_path, None, symbol_returns=b"no marker")
|
||||
with pytest.raises(RecipeIdentityError, match="embeds no runtime attestation"):
|
||||
attest_loaded_runtime(artifact)
|
||||
|
||||
|
||||
@requires_cc
|
||||
def test_artifact_without_the_symbol_fails_closed(tmp_path):
|
||||
artifact = _build_native_artifact(tmp_path, _payload(), export_symbol=False)
|
||||
with pytest.raises(RecipeIdentityError, match="does not export"):
|
||||
attest_loaded_runtime(artifact)
|
||||
|
||||
|
||||
@requires_cc
|
||||
def test_loaded_runtime_disagreeing_with_its_marker_fails_closed(tmp_path):
|
||||
artifact = _build_native_artifact(
|
||||
tmp_path, _payload(), symbol_returns=b"not the marker"
|
||||
)
|
||||
with pytest.raises(RecipeIdentityError, match="different attestation"):
|
||||
attest_loaded_runtime(artifact)
|
||||
|
||||
|
||||
@requires_cc
|
||||
def test_conflicting_markers_fail_closed(tmp_path):
|
||||
artifact = _build_native_artifact(
|
||||
tmp_path, _payload(), extra_payloads=(_payload(patched_tree="f" * 40),)
|
||||
)
|
||||
with pytest.raises(RecipeIdentityError, match="conflicting"):
|
||||
attest_loaded_runtime(artifact)
|
||||
|
||||
|
||||
@requires_cc
|
||||
def test_noncanonical_marker_payload_fails_closed(tmp_path):
|
||||
(tmp_path / "a").mkdir()
|
||||
wrong_keys = _build_native_artifact(tmp_path / "a", b'{ "spaced": true }')
|
||||
with pytest.raises(RecipeIdentityError, match="exactly the attestation fields"):
|
||||
attest_loaded_runtime(wrong_keys)
|
||||
# Right keys, non-canonical encoding: the digest binding would be
|
||||
# ambiguous, so the extractor refuses.
|
||||
spaced = _payload().replace(b":", b": ").replace(b",", b", ")
|
||||
(tmp_path / "b").mkdir()
|
||||
noncanonical = _build_native_artifact(tmp_path / "b", spaced)
|
||||
with pytest.raises(RecipeIdentityError, match="canonical form"):
|
||||
attest_loaded_runtime(noncanonical)
|
||||
|
||||
|
||||
@requires_cc
|
||||
def test_lock_values_cannot_launder_a_mismatched_runtime(tmp_path):
|
||||
# A real (loadable, self-consistent) artifact built from the *wrong* tree
|
||||
# attests fine — then editing the Python object to the lock's values must
|
||||
# fail, or evidence extraction would be decorative.
|
||||
forged = _forged_attestation(tmp_path, patched_tree="f" * 40)
|
||||
with pytest.raises(RecipeIdentityError, match="edited after extraction"):
|
||||
dataclasses.replace(forged, patched_tree=load_runtime_pin().patched_tree)
|
||||
|
||||
|
||||
def test_evidence_binds_the_attested_artifact_bytes(genuine_artifact, genuine_attestation):
|
||||
assert genuine_attestation.evidence.binary_digest == hashlib.sha256(
|
||||
genuine_artifact.read_bytes()
|
||||
).hexdigest()
|
||||
assert genuine_attestation.evidence.artifact_path == str(genuine_artifact)
|
||||
|
||||
|
||||
# --- the executing runtime must match the lock, field by field -------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"field,value",
|
||||
[
|
||||
("runtime_name", "other.cpp"),
|
||||
("upstream_commit", "f" * 40),
|
||||
("patched_tree", "f" * 40),
|
||||
("patch_stack_digest", _digest("f")),
|
||||
("build_recipe_digest", _digest("f")),
|
||||
("boundary_schema_version", 2),
|
||||
("protocol_schema_version", 2),
|
||||
],
|
||||
)
|
||||
@requires_cc
|
||||
def test_native_identity_fails_closed_when_executing_runtime_disagrees(
|
||||
field, value, tmp_path
|
||||
):
|
||||
attestation = _forged_attestation(tmp_path, **{field: value})
|
||||
with pytest.raises(RecipeIdentityError, match="attested"):
|
||||
shard_identity_from_native_report(
|
||||
_inputs(attestation, loaded_artifact=_report_with(attestation))
|
||||
)
|
||||
|
||||
|
||||
@requires_cc
|
||||
def test_distinguishable_runtime_attestations_cannot_emit_one_accepted_identity(
|
||||
tmp_path, genuine_attestation
|
||||
):
|
||||
accepted = shard_identity_from_native_report(_inputs(genuine_attestation))
|
||||
forged = _forged_attestation(tmp_path, patched_tree="f" * 40)
|
||||
with pytest.raises(RecipeIdentityError, match="patched source tree"):
|
||||
shard_identity_from_native_report(
|
||||
_inputs(forged, loaded_artifact=_report_with(forged))
|
||||
)
|
||||
assert accepted.recipe.runtime_version == (
|
||||
load_runtime_pin().runtime_version
|
||||
+ "+artifact."
|
||||
+ genuine_attestation.evidence.binary_digest
|
||||
)
|
||||
|
||||
|
||||
# --- the attested positive path ---------------------------------------------
|
||||
|
||||
|
||||
def test_native_identity_uses_loaded_report_not_a_caller_range(genuine_attestation):
|
||||
identity = shard_identity_from_native_report(_inputs(genuine_attestation))
|
||||
|
||||
assert (identity.shard_start, identity.shard_end) == (2, 6)
|
||||
assert identity.artifact.architecture == "llama"
|
||||
assert identity.artifact.layer_count == 8
|
||||
assert identity.recipe.runtime_version == load_runtime_pin().runtime_version
|
||||
assert identity.recipe.runtime_version == (
|
||||
load_runtime_pin().runtime_version
|
||||
+ "+artifact."
|
||||
+ genuine_attestation.evidence.binary_digest
|
||||
)
|
||||
|
||||
|
||||
def test_native_identity_requires_an_immutable_pin_and_gguf_range():
|
||||
@requires_cc
|
||||
def test_copying_public_lock_values_cannot_forge_the_certified_runtime_identity(
|
||||
tmp_path, genuine_attestation
|
||||
):
|
||||
"""A second loadable artifact with lock-true self-report gets a new identity."""
|
||||
accepted = shard_identity_from_native_report(_inputs(genuine_attestation))
|
||||
copied_artifact = _build_native_artifact(
|
||||
tmp_path, expected_attestation_payload(load_runtime_pin())
|
||||
)
|
||||
# Keep the same exported marker/symbol while making this a different set of
|
||||
# executing artifact bytes, exactly like a separately built binary that
|
||||
# copied the public lock values into its self-report.
|
||||
copied_artifact.write_bytes(copied_artifact.read_bytes() + b"copied-lock-forgery")
|
||||
copied_lock_values = attest_loaded_runtime(copied_artifact)
|
||||
copied = shard_identity_from_native_report(
|
||||
_inputs(
|
||||
copied_lock_values,
|
||||
loaded_artifact=_report_with(copied_lock_values),
|
||||
)
|
||||
)
|
||||
|
||||
assert copied_lock_values.evidence.binary_digest != (
|
||||
genuine_attestation.evidence.binary_digest
|
||||
)
|
||||
assert copied.recipe.runtime_version != accepted.recipe.runtime_version
|
||||
assert copied.fingerprint.runtime_recipe_digest != (
|
||||
accepted.fingerprint.runtime_recipe_digest
|
||||
)
|
||||
|
||||
|
||||
def test_native_identity_requires_an_immutable_pin_and_gguf_range(genuine_attestation):
|
||||
with pytest.raises(Exception, match="moving reference"):
|
||||
shard_identity_from_native_report(
|
||||
_inputs(artifact_pin=ImmutableArtifactPin("a", "main", _digest("b")))
|
||||
_inputs(
|
||||
genuine_attestation,
|
||||
artifact_pin=ImmutableArtifactPin("a", "main", _digest("b")),
|
||||
)
|
||||
)
|
||||
with pytest.raises(Exception, match="outside GGUF"):
|
||||
NativeLoadedArtifactReport(0, 9, 1, 1, 1, "llama", _digest("a"), 8)
|
||||
NativeLoadedArtifactReport(
|
||||
0, 9, 1, 1, 1, "llama", _digest("a"), 8, genuine_attestation
|
||||
)
|
||||
|
||||
|
||||
def test_native_worker_rejects_bad_session_open_before_session_acceptance():
|
||||
adapter = NativeWorkerBackendAdapter(_inputs())
|
||||
def test_native_worker_rejects_bad_session_open_before_session_acceptance(
|
||||
genuine_attestation,
|
||||
):
|
||||
adapter = NativeWorkerBackendAdapter(_inputs(genuine_attestation))
|
||||
accepted = adapter.on_session_open(
|
||||
_open(adapter), expected_route_session_id="tracker-session", expected_route_epoch=4
|
||||
)
|
||||
@@ -112,14 +419,16 @@ def test_native_worker_rejects_bad_session_open_before_session_acceptance():
|
||||
assert rejected.value.error.code == pb.ERROR_CODE_EPOCH_STALE
|
||||
|
||||
|
||||
def test_doctor_emits_native_identity_but_keeps_legacy_backend_dark():
|
||||
def test_doctor_emits_native_identity_but_keeps_legacy_backend_dark(
|
||||
genuine_attestation,
|
||||
):
|
||||
manifest = parse_recipe_manifest(
|
||||
{"schema_version": 1, "catalogue_version": "2026.07.1", "recipes": [
|
||||
{"id": "native", "version": "1", "backend_id": "llama-cpp"}
|
||||
]}
|
||||
)
|
||||
selection = DoctorSelection("acme/llama.gguf", 2, 5)
|
||||
native = NativeWorkerBackendAdapter(_inputs())
|
||||
native = NativeWorkerBackendAdapter(_inputs(genuine_attestation))
|
||||
# The probe needs only the normal backend shape; identity is supplied by the adapter.
|
||||
native.hidden_size = 8
|
||||
native.is_head = False
|
||||
|
||||
Reference in New Issue
Block a user