fix(sovereign-agent-connector): reject boolean GCP claims, invalid validity_seconds, and non-canonical base64url signatures (#379)

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(

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,15 +103,24 @@ 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")
     if not isinstance(signature, str):
         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

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)

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)

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