diff --git a/examples/sovereign-agent-connector/agentrust_auc/appraisal.py b/examples/sovereign-agent-connector/agentrust_auc/appraisal.py index 16d816d9..f85bf514 100644 --- a/examples/sovereign-agent-connector/agentrust_auc/appraisal.py +++ b/examples/sovereign-agent-connector/agentrust_auc/appraisal.py @@ -46,12 +46,13 @@ class AppraisalPolicy: def _workload_key_thumbprint(workload_key: str) -> bytes: - try: - key_bytes = bytes.fromhex(workload_key) - except ValueError as exc: - raise BundleError("workload key is not valid hex") from exc - if len(key_bytes) != 32: - raise BundleError("workload key must be a 32-byte Ed25519 public key") + if ( + not isinstance(workload_key, str) + or len(workload_key) != 64 + or any(c not in "0123456789abcdef" for c in workload_key) + ): + raise BundleError("workload key must be a 32-byte Ed25519 public key in lowercase hex") + key_bytes = bytes.fromhex(workload_key) x = base64.urlsafe_b64encode(key_bytes).rstrip(b"=").decode() canonical = json.dumps( {"crv": "Ed25519", "kty": "OKP", "x": x}, @@ -70,6 +71,11 @@ def _azure_imds_time(value: Any) -> datetime: raise BundleError("Azure IMDS document timestamp is malformed") from exc +def _validate_validity_seconds(validity_seconds: int) -> None: + if type(validity_seconds) is not int or validity_seconds <= 0: + raise BundleError("validity_seconds must be a positive integer") + + def make_placement_assertion( *, workload_key: str, @@ -80,6 +86,10 @@ def make_placement_assertion( signing_key: Ed25519PrivateKey, ) -> dict[str, Any]: """Create accountable placement evidence, explicitly separate from hardware.""" + _workload_key_thumbprint(workload_key) + _validate_validity_seconds(validity_seconds) + if not isinstance(issued_at, datetime) or issued_at.tzinfo is None: + raise BundleError("issued_at must be a timezone-aware datetime") return _sign( { "type": "auc:OperatorPlacementAssertion", @@ -106,6 +116,7 @@ def make_azure_placement_assertion( ) -> dict[str, Any]: """Join verified Azure IMDS identity to ARM placement and derive jurisdiction.""" _workload_key_thumbprint(workload_key) + _validate_validity_seconds(validity_seconds) nonce = imds_document.get("nonce") if not isinstance(nonce, str) or len(nonce) != 10 or not nonce.isdigit(): raise BundleError("Azure IMDS document has no valid 10-digit challenge") @@ -178,6 +189,7 @@ def make_gcp_placement_assertion( relying-party interpretation of Google's signed GCE zone claim. """ _workload_key_thumbprint(workload_key) + _validate_validity_seconds(validity_seconds) if observed_at.tzinfo is None: raise BundleError("GCP observation timestamp must be timezone-aware") gce = verified_claims.get("submods", {}).get("gce", {}) @@ -195,7 +207,7 @@ def make_gcp_placement_assertion( raise BundleError("GCP Confidential Space debug mode is not allowed") issued_at = verified_claims.get("iat") expires_at = verified_claims.get("exp") - if not isinstance(issued_at, int) or not isinstance(expires_at, int): + if type(issued_at) is not int or type(expires_at) is not int: raise BundleError("GCP token validity window is malformed") observed_epoch = int(observed_at.timestamp()) if not issued_at <= observed_epoch <= expires_at: @@ -205,7 +217,7 @@ def make_gcp_placement_assertion( instance_id = gce.get("instance_id") if not all(isinstance(value, str) and value for value in (image_digest, project_id)): raise BundleError("GCP workload identity claims are incomplete") - if not isinstance(instance_id, str | int): + if isinstance(instance_id, bool) or not isinstance(instance_id, str | int) or str(instance_id) == "": raise BundleError("GCP instance identity claim is missing") return _sign( diff --git a/examples/sovereign-agent-connector/agentrust_auc/core.py b/examples/sovereign-agent-connector/agentrust_auc/core.py index 8553d40c..77e86748 100644 --- a/examples/sovereign-agent-connector/agentrust_auc/core.py +++ b/examples/sovereign-agent-connector/agentrust_auc/core.py @@ -75,7 +75,20 @@ def _b64(value: bytes) -> str: def _unb64(value: str) -> bytes: - return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)) + if ( + not isinstance(value, str) + or not value + or len(value) % 4 == 1 + or any( + c not in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_" + for c in value + ) + ): + raise ValueError("invalid base64url string") + raw = base64.urlsafe_b64decode(value.encode("ascii") + b"=" * (-len(value) % 4)) + if _b64(raw) != value: + raise ValueError("non-canonical base64url string") + return raw def _public_hex(key: Ed25519PrivateKey) -> str: @@ -90,6 +103,12 @@ def _sign(body: dict[str, Any], key: Ed25519PrivateKey) -> dict[str, Any]: def _verify_signed(value: dict[str, Any], trusted_key_hex: str) -> None: + if ( + not isinstance(trusted_key_hex, str) + or len(trusted_key_hex) != 64 + or any(c not in "0123456789abcdef" for c in trusted_key_hex) + ): + raise BundleError("trusted_key_hex must be 64 lowercase hex characters") if value.get("issuer_key") != trusted_key_hex: raise BundleError("signed artifact issuer does not match its pinned trust key") signature = value.get("signature") @@ -97,8 +116,11 @@ def _verify_signed(value: dict[str, Any], trusted_key_hex: str) -> None: raise BundleError("signed artifact has no signature") body = {key: item for key, item in value.items() if key != "signature"} try: + sig_bytes = _unb64(signature) + if len(sig_bytes) != 64: + raise ValueError("Ed25519 signature must be 64 bytes") Ed25519PublicKey.from_public_bytes(bytes.fromhex(trusted_key_hex)).verify( - _unb64(signature), _canonical(body) + sig_bytes, _canonical(body) ) except (InvalidSignature, ValueError) as exc: raise BundleError("signed artifact signature is invalid") from exc diff --git a/examples/sovereign-agent-connector/tests/test_appraisal.py b/examples/sovereign-agent-connector/tests/test_appraisal.py index 9345bc55..10c4f028 100644 --- a/examples/sovereign-agent-connector/tests/test_appraisal.py +++ b/examples/sovereign-agent-connector/tests/test_appraisal.py @@ -286,3 +286,157 @@ def test_gcp_placement_fails_closed(mutation, message: str) -> None: validity_seconds=300, signing_key=signing_key, ) + + +def test_auc_rejects_boolean_gcp_claims_and_invalid_validity() -> None: + signing_key, _ = new_keypair() + for mut in ( + lambda c: c.update({"iat": True}), + lambda c: c.update({"exp": True}), + lambda c: c["submods"]["gce"].update({"instance_id": True}), + lambda c: c["submods"]["gce"].update({"instance_id": ""}), + ): + claims = _gcp_claims() + mut(claims) + with pytest.raises(BundleError): + make_gcp_placement_assertion( + workload_key="ab" * 32, + token="signed.jwt.token", + verified_claims=claims, + zone_jurisdictions={"europe-west9-a": "FR"}, + observed_at=NOW, + validity_seconds=300, + signing_key=signing_key, + ) + + for bad_val in (True, False, 0, -10, 1.5): + with pytest.raises(BundleError): + make_placement_assertion( + workload_key="ab" * 32, + jurisdiction="FR", + operator="op", + issued_at=NOW, + validity_seconds=bad_val, # type: ignore[arg-type] + signing_key=signing_key, + ) + for bad_wk in ("AB" * 32, "ab" * 31, 123): + with pytest.raises(BundleError): + make_placement_assertion( + workload_key=bad_wk, # type: ignore[arg-type] + jurisdiction="FR", + operator="op", + issued_at=NOW, + validity_seconds=300, + signing_key=signing_key, + ) + + +def _load_independent_verify(): + import importlib.util + import sys + from pathlib import Path + + root = Path(__file__).parents[1] + spec = importlib.util.spec_from_file_location( + "independent_verify_for_appraisal", root / "tools" / "independent_verify.py" + ) + assert spec is not None and spec.loader is not None + mod = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = mod + spec.loader.exec_module(mod) + return mod + + +@pytest.mark.parametrize("suffix", ["=", "=="]) +def test_padded_signature_rejected_by_both_core_and_independent_verifier( + suffix: str, +) -> None: + import base64 + from agentrust_auc.core import _verify_signed + + ind = _load_independent_verify() + signing_key, public_key = new_keypair() + good = make_placement_assertion( + workload_key="ab" * 32, + jurisdiction="FR", + operator="op", + issued_at=NOW, + validity_seconds=300, + signing_key=signing_key, + ) + # Canonical control passes both verifiers. + _verify_signed(good, public_key) + ind.verify_ed25519(good, public_key) + + padded_sig = good["signature"] + suffix + assert base64.urlsafe_b64decode(padded_sig + "=" * (-len(padded_sig) % 4)) == ( + base64.urlsafe_b64decode(good["signature"] + "==") + ) + padded = {**good, "signature": padded_sig} + with pytest.raises(BundleError): + _verify_signed(padded, public_key) + with pytest.raises(ind.VerificationError): + ind.verify_ed25519(padded, public_key) + + +@pytest.mark.parametrize("raw_len", [0, 32, 63, 65]) +def test_wrong_length_signature_rejected_by_both_core_and_independent_verifier( + raw_len: int, +) -> None: + import base64 + from agentrust_auc.core import _verify_signed + + ind = _load_independent_verify() + signing_key, public_key = new_keypair() + good = make_placement_assertion( + workload_key="ab" * 32, + jurisdiction="FR", + operator="op", + issued_at=NOW, + validity_seconds=300, + signing_key=signing_key, + ) + _verify_signed(good, public_key) + ind.verify_ed25519(good, public_key) + + wrong_sig = base64.urlsafe_b64encode(b"\x01" * raw_len).rstrip(b"=").decode("ascii") + mutated = {**good, "signature": wrong_sig} + with pytest.raises(BundleError): + _verify_signed(mutated, public_key) + with pytest.raises(ind.VerificationError): + ind.verify_ed25519(mutated, public_key) + + +@pytest.mark.parametrize( + "bad_key_fn", + [ + lambda k: k.upper(), + lambda k: " " + k, + lambda k: k + "\n", + lambda k: k[:62], + lambda k: "zz" * 32, + ], +) +def test_malformed_trusted_key_rejected_by_both_core_and_independent_verifier( + bad_key_fn, +) -> None: + from agentrust_auc.core import _verify_signed + + ind = _load_independent_verify() + signing_key, public_key = new_keypair() + good = make_placement_assertion( + workload_key="ab" * 32, + jurisdiction="FR", + operator="op", + issued_at=NOW, + validity_seconds=300, + signing_key=signing_key, + ) + _verify_signed(good, public_key) + ind.verify_ed25519(good, public_key) + + bad_key = bad_key_fn(public_key) + with pytest.raises(BundleError): + _verify_signed(good, bad_key) + with pytest.raises(ind.VerificationError): + ind.verify_ed25519(good, bad_key) diff --git a/examples/sovereign-agent-connector/tests/test_independent_verifier.py b/examples/sovereign-agent-connector/tests/test_independent_verifier.py index b5b7c9ed..768852a9 100644 --- a/examples/sovereign-agent-connector/tests/test_independent_verifier.py +++ b/examples/sovereign-agent-connector/tests/test_independent_verifier.py @@ -86,3 +86,56 @@ def test_independent_verifier_cli_reports_malformed_bundle_as_fail( assert code == 1 assert capsys.readouterr().err.startswith("FAIL:") + + +def test_independent_verifier_rejects_padded_mve1_signatures() -> None: + bundle = copy.deepcopy(load_bundle()) + anchors = load_anchors() + placement = bundle["base_bundle"]["placement_assertion"] + trusted_key = anchors["base"]["placement_operator"] + MODULE.verify_ed25519(placement, trusted_key) + + placement["signature"] = placement["signature"] + "=" + with pytest.raises(MODULE.VerificationError): + MODULE.verify_ed25519(placement, trusted_key) + with pytest.raises(MODULE.VerificationError): + MODULE.verify_live(bundle, anchors) + + +@pytest.mark.parametrize("raw_len", [32, 63, 65]) +def test_independent_verifier_rejects_wrong_length_signature(raw_len: int) -> None: + import base64 + + bundle = copy.deepcopy(load_bundle()) + anchors = load_anchors() + placement = bundle["base_bundle"]["placement_assertion"] + trusted_key = anchors["base"]["placement_operator"] + MODULE.verify_ed25519(placement, trusted_key) + + placement["signature"] = ( + base64.urlsafe_b64encode(b"\x02" * raw_len).rstrip(b"=").decode("ascii") + ) + with pytest.raises(MODULE.VerificationError): + MODULE.verify_ed25519(placement, trusted_key) + + +@pytest.mark.parametrize( + "bad_key_fn", + [ + lambda k: k.upper(), + lambda k: " " + k, + lambda k: k[:62], + lambda k: "zz" * 32, + ], +) +def test_independent_verifier_rejects_malformed_trusted_key(bad_key_fn) -> None: + bundle = copy.deepcopy(load_bundle()) + anchors = load_anchors() + placement = bundle["base_bundle"]["placement_assertion"] + trusted_key = anchors["base"]["placement_operator"] + MODULE.verify_ed25519(placement, trusted_key) + + bad_key = bad_key_fn(trusted_key) + mutated = {**placement, "issuer_key": bad_key} + with pytest.raises(MODULE.VerificationError): + MODULE.verify_ed25519(mutated, bad_key) diff --git a/examples/sovereign-agent-connector/tools/independent_verify.py b/examples/sovereign-agent-connector/tools/independent_verify.py index 841ca850..90320bd4 100644 --- a/examples/sovereign-agent-connector/tools/independent_verify.py +++ b/examples/sovereign-agent-connector/tools/independent_verify.py @@ -60,7 +60,20 @@ def digest(value: Any) -> str: def decode_b64url(value: str) -> bytes: - return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)) + if ( + not isinstance(value, str) + or not value + or len(value) % 4 == 1 + or any( + c not in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_" + for c in value + ) + ): + raise ValueError("invalid base64url string") + raw = base64.urlsafe_b64decode(value.encode("ascii") + b"=" * (-len(value) % 4)) + if base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") != value: + raise ValueError("non-canonical base64url string") + return raw def key_thumbprint(public_key_hex: str) -> bytes: @@ -74,13 +87,22 @@ def require(condition: bool, message: str) -> None: def verify_ed25519(body: dict[str, Any], trusted_hex: str, *, cmcp: bool = False) -> None: + require( + isinstance(trusted_hex, str) + and len(trusted_hex) == 64 + and all(c in "0123456789abcdef" for c in trusted_hex), + "invalid trusted public key hex", + ) require(body.get("issuer_key", trusted_hex) == trusted_hex, "unexpected signing key") signature = body.get("signature") require(isinstance(signature, str), "missing signature") unsigned = {key: value for key, value in body.items() if key != "signature"} try: + sig_bytes = decode_b64url(signature) + if len(sig_bytes) != 64: + raise ValueError("Ed25519 signature must be 64 bytes") Ed25519PublicKey.from_public_bytes(bytes.fromhex(trusted_hex)).verify( - decode_b64url(signature), canonical(unsigned, ascii_only=cmcp) + sig_bytes, canonical(unsigned, ascii_only=cmcp) ) except (InvalidSignature, ValueError) as exc: raise VerificationError("invalid Ed25519 signature") from exc