Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion claude-code/tests/test_claude_code_capture.py
Original file line number Diff line number Diff line change
Expand Up @@ -106,13 +106,21 @@ def test_mutable_state_churn_does_not_alarm(self, tmp_path, monkeypatch):
(d / "state" / "progress.json").write_text('{"runs": 2}', encoding="utf-8")
assert capture.diff(before, _skills_snap()) == []

@pytest.mark.parametrize("junk", ["run.log", "cached.pyc", "scratch.tmp"])
@pytest.mark.parametrize("junk", ["run.log", "scratch.tmp"])
def test_run_artifacts_do_not_alarm(self, tmp_path, monkeypatch, junk):
d = _skill(tmp_path, monkeypatch)
before = _skills_snap()
(d / junk).write_text("noise", encoding="utf-8")
assert capture.diff(before, _skills_snap()) == []

@pytest.mark.parametrize("payload", ["cached.pyc", "node_modules/x/index.js", "state/run.py"])
def test_loadable_code_alarms_even_where_data_is_excluded(self, tmp_path, monkeypatch, payload):
d = _skill(tmp_path, monkeypatch)
before = _skills_snap()
(d / payload).parent.mkdir(parents=True, exist_ok=True)
(d / payload).write_text("payload", encoding="utf-8")
assert capture.diff(before, _skills_snap()) != []

def test_directory_without_a_manifest_is_not_a_skill(self, tmp_path, monkeypatch):
claude = tmp_path / ".claude"
(claude / "skills" / "notaskill").mkdir(parents=True)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -18,4 +18,4 @@ c05e327faaaa7b4edd3d6b8e9cb679f752b2ee8fe647b070602892bd398eb86b positive/ontog
51464c725f51f14c0ba959d2b8752f7a614d9acc6386ba467af71229fa8e3776 positive/trace_claim_candidate.json
2f49d1277bb9d21c8decee0526b9c51f974fa69b4ed83b1049974af79e91b414 positive/trace_level0_conformance.txt
0d641d7084e3444c191e409a2d57c6faf9472751ad2d0edbb9d296f3cab95997 replay_adapter.py
ff70c74926500ebdb534659cfd0be807ac4c3dfa9af449542723aca1e517e76e verify_proof.py
c33b5605cf1bbe646368012eec941b7257e373aa213b610455026fc7a6fab050 verify_proof.py
Original file line number Diff line number Diff line change
Expand Up @@ -91,10 +91,16 @@ def main() -> int:
else:
ok("checksum manifest")

auth = load("positive/ontoguard_authorization.exact.json")
sig = load("positive/ontoguard_authorization.signature.json")
og_jwk = load("positive/ontoguard_public_jwk.json")
exact = sig["exact_bytes"].encode("utf-8")
# Every binding below reads the signed bytes. The .exact.json sibling is a
# convenience copy with no signature over it, so it must equal them.
auth = json.loads(exact)
if load("positive/ontoguard_authorization.exact.json") == auth:
ok("authorization copy matches the signed bytes")
else:
fail("authorization copy matches the signed bytes")
if digest_bytes(exact) == sig["digest"]:
ok("OntoGuard authorization exact-byte digest")
else:
Expand Down Expand Up @@ -142,16 +148,21 @@ def main() -> int:
else:
fail("independent executable-action digest", recomputed)

receipt = load("positive/execution_receipt.json")
receipt_file = load("positive/execution_receipt.json")
receipt = json.loads(receipt_file["receipt_bytes"])
if all(receipt_file.get(k) == v for k, v in receipt.items()):
ok("receipt fields match the signed receipt bytes")
else:
fail("receipt fields match the signed receipt bytes")
if receipt.get("executed_action_binding_digest") == auth["action_binding_digest"]:
ok("executed action == authorized action")
else:
fail("executed action == authorized action")

ex_jwk = load("positive/execution_public_jwk.json")
ok("execution runtime signing key present in pack")
raw_receipt = receipt["receipt_bytes"].encode("utf-8")
if verify_ed25519(ex_jwk, raw_receipt, receipt["signature"]):
raw_receipt = receipt_file["receipt_bytes"].encode("utf-8")
if verify_ed25519(ex_jwk, raw_receipt, receipt_file["signature"]):
ok("execution receipt Ed25519 signature")
else:
fail("execution receipt Ed25519 signature")
Expand Down
29 changes: 24 additions & 5 deletions integrations/sentinel/sentinel/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@
)
from sentinel.risk_engine import RiskEngine
from sentinel.replay_engine import ReplayEngine
from sentinel.trace_claim_generator import load_signing_key
from cryptography.exceptions import InvalidSignature
import traceback
import uuid
import json
Expand Down Expand Up @@ -44,10 +46,28 @@ def log_enforcement(action: str, claim_id: str, result: dict, status: str = "SUC
print(f"Result: {result.get('message', result)}")
print(f"Status: {status}\n")

def _canonical(payload: dict) -> bytes:
return json.dumps(payload, sort_keys=True).encode('utf-8')

def sign_payload(payload: dict) -> str:
data = json.dumps(payload, sort_keys=True).encode('utf-8')
hash_digest = hashlib.sha256(data).digest()
return base64.b64encode(hash_digest + b"signed").decode('utf-8')
"""Ed25519-sign a report with the same key that signs Sentinel's TRACE claims.

The previous "signature" was sha256(payload) + b"signed", which anyone could
recompute, so /verify reported a forged report as VERIFIED. Without a key this
raises, matching the fail-closed rule for TRACE claims.
"""
key = load_signing_key()
return base64.b64encode(key.sign(_canonical(payload))).decode('utf-8')

def verify_payload_signature(payload: dict, signature: object) -> bool:
if not isinstance(signature, str):
return False
try:
sig = base64.b64decode(signature, validate=True)
load_signing_key().public_key().verify(sig, _canonical(payload))
return True
except (ValueError, InvalidSignature):
return False

def hash_payload(payload: dict) -> str:
data = json.dumps(payload, sort_keys=True).encode('utf-8')
Expand Down Expand Up @@ -329,11 +349,10 @@ async def verify_incident(claim_id: str, request: Request):
"risk_score": report_data.get("risk_score")
})
recomputed_incident_hash = hash_payload(report_copy)
recomputed_signature = sign_payload(report_copy)

valid_claim_hash = recomputed_claim_hash == report_data.get("claim_hash")
valid_incident_hash = recomputed_incident_hash == report_data.get("incident_hash")
valid_signature = recomputed_signature == report_data.get("signature")
valid_signature = verify_payload_signature(report_copy, report_data.get("signature"))

status = "VERIFIED" if (valid_claim_hash and valid_incident_hash and valid_signature) else "TAMPERED"
return JSONResponse(content={
Expand Down
73 changes: 73 additions & 0 deletions integrations/sentinel/tests/test_incident_signature.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
"""The incident-report signature must need Sentinel's key to produce.

It used to be sha256(payload) + b"signed", which anyone could recompute, so
/verify accepted a forged report as VERIFIED.
"""

from __future__ import annotations

import base64
import hashlib
import json

import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from fastapi.testclient import TestClient

from sentinel.server import app, hash_payload, sign_payload


@pytest.fixture
def client(monkeypatch):
pem = Ed25519PrivateKey.generate().private_bytes(
serialization.Encoding.PEM,
serialization.PrivateFormat.PKCS8,
serialization.NoEncryption(),
).decode()
monkeypatch.setenv("TRACE_PRIVATE_KEY_PEM", pem)
return TestClient(app)


def _report(sign) -> dict:
body = {"agent_id": "a1", "detection_type": "tool_drift", "risk_score": 0.9, "incident_id": "INC-1"}
return dict(
body,
claim_hash=hash_payload({"claim_id": "c1", "agent_id": "a1", "detection_type": "tool_drift", "risk_score": 0.9}),
incident_hash=hash_payload(body),
signature=sign(body),
)


def _keyless(payload: dict) -> str:
digest = hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).digest()
return base64.b64encode(digest + b"signed").decode()


def test_report_signed_by_sentinel_verifies(client):
r = client.post("/verify/c1", json={"report": _report(sign_payload)}).json()
assert r["status"] == "VERIFIED"


def test_keyless_forgery_is_tampered(client):
r = client.post("/verify/c1", json={"report": _report(_keyless)}).json()
assert r["status"] == "TAMPERED"
assert r["details"]["signature_valid"] is False


def test_report_signed_by_another_key_is_tampered(client):
other = Ed25519PrivateKey.generate()
forged = _report(lambda p: base64.b64encode(other.sign(json.dumps(p, sort_keys=True).encode())).decode())
assert client.post("/verify/c1", json={"report": forged}).json()["status"] == "TAMPERED"


def test_edited_report_is_tampered(client):
report = _report(sign_payload)
report["risk_score"] = 0.1
assert client.post("/verify/c1", json={"report": report}).json()["status"] == "TAMPERED"


def test_signing_without_a_key_fails_closed(monkeypatch):
monkeypatch.delenv("TRACE_PRIVATE_KEY_PEM", raising=False)
with pytest.raises(RuntimeError):
sign_payload({"x": 1})
4 changes: 3 additions & 1 deletion integrations/wcm-azure-skr/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,9 @@ available in this file.

It does refuse claim sets that cannot support the tier: an unknown attestation
type, a compliance status other than `azure-compliant-cvm`, or a debuggable
guest.
guest. It also refuses a token whose nonce, top level or in `x-ms-runtime`, is
absent or differs from the challenge, so pass the challenge nonce to the
attestation request. `nonce_echo` is taken from the token, not the challenge.

## Run it

Expand Down
23 changes: 23 additions & 0 deletions integrations/wcm-azure-skr/test_wcm_azure_skr.py
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,7 @@ def good_claims(**overrides: object) -> dict:
"x-ms-compliance-status": "azure-compliant-cvm",
"x-ms-sevsnpvm-is-debuggable": "false",
"x-ms-sevsnpvm-idkeydigest": "cd" * 48,
"x-ms-runtime": {"nonce": "a" * 64},
}
claims.update(overrides)
return claims
Expand Down Expand Up @@ -301,3 +302,25 @@ def test_cli_warns_loudly_when_the_workload_is_unbound(tmp_path: pathlib.Path, c
def test_cli_describes_claims(capsys) -> None:
assert main(["ignored", "--authority", AUTHORITY, "--describe-claims"]) == 0
assert "x-ms-compliance-status" in capsys.readouterr().out


@pytest.mark.parametrize(
"claims",
[
{"x-ms-runtime": {"nonce": "b" * 64}},
{"x-ms-runtime": {}, "nonce": "b" * 64},
{"x-ms-runtime": {"nonce": "a" * 64}, "nonce": "b" * 64},
],
)
def test_a_token_minted_for_another_challenge_is_refused(claims: dict) -> None:
"""nonce_echo used to be copied from the verifier's own challenge, so a stale
token was reported as fresh."""
with pytest.raises(SkrPolicyError, match="does not match"):
evidence_from_maa_claims(good_claims(**claims), challenge(), serving_image_measurement=SERVING)


def test_a_token_with_no_nonce_is_refused() -> None:
with pytest.raises(SkrPolicyError, match="no nonce"):
evidence_from_maa_claims(
good_claims(**{"x-ms-runtime": {}}), challenge(), serving_image_measurement=SERVING
)
27 changes: 26 additions & 1 deletion integrations/wcm-azure-skr/wcm_azure_skr.py
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,31 @@
}


def _token_nonce(claims: Mapping[str, Any], expected: str) -> str:
"""Return the nonce the MAA token carries, after checking it is ``expected``.

The nonce can arrive top level or inside ``x-ms-runtime``. Every place it is
present must match, and at least one must be present. Copying the verifier's
own challenge into ``nonce_echo`` instead would make a stale token look fresh.
"""
runtime = claims.get("x-ms-runtime")
found = [claims.get("nonce")]
if isinstance(runtime, Mapping):
found.append(runtime.get("nonce"))
found = [value for value in found if value is not None]
if not found:
raise SkrPolicyError(
"the MAA token carries no nonce, so nothing binds it to this challenge. "
"Pass the challenge nonce to the attestation request."
)
if any(value != expected for value in found):
raise SkrPolicyError(
"the MAA token's nonce does not match this challenge, so the token was "
"not produced for this request"
)
return expected


class SkrPolicyError(ValueError):
"""Raised when a manifest cannot be translated into a usable SKR policy."""

Expand Down Expand Up @@ -299,7 +324,7 @@ def evidence_from_maa_claims(
platform=platform,
assurance_tier=AssuranceTier.hardware_attested.value,
serving_image_measurement=serving_image_measurement,
nonce_echo=challenge.nonce,
nonce_echo=_token_nonce(claims, challenge.nonce),
attestation_key_id=str(claims.get("x-ms-sevsnpvm-idkeydigest", "maa-token")),
transport_public_key=transport_public_key,
)
Expand Down
3 changes: 3 additions & 0 deletions integrations/wcm-gcp-confidential-space/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,9 @@ and is not.

It does refuse claim sets that cannot support the tier: a non-Confidential-Space
`swname`, an unmapped `hwmodel`, or a `dbgstat` other than `disabled-since-boot`.
It also refuses a token whose `eat_nonce` does not include the challenge nonce,
so request the token with that nonce. `nonce_echo` is taken from the token, not
the challenge.

## Run it

Expand Down
21 changes: 21 additions & 0 deletions integrations/wcm-gcp-confidential-space/test_wcm_gcp_cs.py
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,7 @@ def cs_claims(**overrides: object) -> dict:
"dbgstat": "disabled-since-boot",
"iss": "https://confidentialcomputing.googleapis.com/",
"submods": {"container": {"image_digest": IMAGE}},
"eat_nonce": "a" * 64,
}
claims.update(overrides)
return claims
Expand Down Expand Up @@ -341,3 +342,23 @@ def test_cli_prints_claims_for_confirming_hwmodel(tmp_path: pathlib.Path, capsys

assert main(["ignored", "--print-claims", str(token)]) == 0
assert "GCP_AMD_SEV_SNP" in capsys.readouterr().out


@pytest.mark.parametrize("eat_nonce", ["b" * 64, ["b" * 64, "c" * 64]])
def test_a_token_minted_for_another_challenge_is_refused(eat_nonce: object) -> None:
"""nonce_echo used to be copied from the verifier's own challenge, so a stale
token was reported as fresh."""
with pytest.raises(ConfidentialSpaceError, match="does not include"):
evidence_from_cs_claims(cs_claims(eat_nonce=eat_nonce), challenge())


def test_a_token_with_no_eat_nonce_is_refused() -> None:
claims = cs_claims()
del claims["eat_nonce"]
with pytest.raises(ConfidentialSpaceError, match="no eat_nonce"):
evidence_from_cs_claims(claims, challenge())


def test_a_nonce_list_containing_the_challenge_is_accepted() -> None:
evidence = evidence_from_cs_claims(cs_claims(eat_nonce=["b" * 64, "a" * 64]), challenge())
assert evidence.cpu.nonce_echo == "a" * 64
24 changes: 23 additions & 1 deletion integrations/wcm-gcp-confidential-space/wcm_gcp_cs.py
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,28 @@
_CLAIM_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z_][A-Za-z0-9_]*)*$")


def _token_nonce(claims: Mapping[str, Any], expected: str) -> str:
"""Return the ``eat_nonce`` value matching ``expected``, or refuse.

Confidential Space carries caller nonces in ``eat_nonce``, as a string or a
list. Copying the verifier's own challenge into ``nonce_echo`` instead would
make a stale token look fresh.
"""
value = claims.get("eat_nonce")
values = [value] if isinstance(value, str) else list(value or [])
if not values:
raise ConfidentialSpaceError(
"the token carries no eat_nonce, so nothing binds it to this challenge. "
"Request the token with the challenge nonce."
)
if expected not in values:
raise ConfidentialSpaceError(
"the token's eat_nonce does not include this challenge's nonce, so the "
"token was not produced for this request"
)
return expected


class ConfidentialSpaceError(ValueError):
"""Raised when a manifest cannot be turned into a usable condition."""

Expand Down Expand Up @@ -283,7 +305,7 @@ def evidence_from_cs_claims(
platform=platform,
assurance_tier=AssuranceTier.hardware_attested.value,
serving_image_measurement=digest,
nonce_echo=challenge.nonce,
nonce_echo=_token_nonce(claims, challenge.nonce),
attestation_key_id=str(claims.get("iss", "confidential-space-token")),
transport_public_key=transport_public_key,
)
Expand Down
2 changes: 1 addition & 1 deletion packages/agentrust-capture-core/pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "hatchling.build"

[project]
name = "agentrust-capture-core"
version = "0.1.2"
version = "0.1.3"
description = "Shared fingerprinting, comparison and baseline-sealing core for AgenTrust agent-integrity capture engines"
readme = "README.md"
license = "Apache-2.0"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
from .hashing import (
EXCLUDE_DIRS,
EXCLUDE_SUFFIXES,
EXECUTABLE_SUFFIXES,
UNVERIFIABLE_PREFIX,
now_iso,
safe_sha_file,
Expand Down Expand Up @@ -58,11 +59,12 @@
)
from .state import StatePaths, atomic_write, load_state, save_baseline, save_state

__version__ = "0.1.2"
__version__ = "0.1.3"

__all__ = [
"EXCLUDE_DIRS",
"EXCLUDE_SUFFIXES",
"EXECUTABLE_SUFFIXES",
"INTEGRITY_BROKEN",
"INTEGRITY_OK",
"INTEGRITY_UNSEALED",
Expand Down
Loading
Loading