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,12 @@ 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:
+        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 +95,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,62 @@ def test_gcp_placement_fails_closed(mutation, message: str) -> None:
             validity_seconds=300,
             signing_key=signing_key,
         )
+
+
+def test_auc_rejects_boolean_gcp_claims_invalid_validity_and_non_canonical_b64() -> None:
+    from agentrust_auc.core import _verify_signed
+
+    signing_key, public_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,
+            )
+
+    good = make_placement_assertion(
+        workload_key="ab" * 32,
+        jurisdiction="FR",
+        operator="op",
+        issued_at=NOW,
+        validity_seconds=300,
+        signing_key=signing_key,
+    )
+    padded = {**good, "signature": good["signature"] + "="}
+    with pytest.raises(BundleError):
+        _verify_signed(padded, public_key)
+    with pytest.raises(BundleError):
+        _verify_signed(good, public_key.upper())