chore: apply ruff format, and gate it so it cannot drift again

.github/workflows/ci.yml

@@ -36,6 +36,11 @@ jobs:
       - name: Lint
         run: ruff check src/ tests/
 
+      # Checked, not applied. Without this gate 34 of 77 files drifted out of
+      # format unnoticed, because `ruff check` does not cover formatting.
+      - name: Format
+        run: ruff format --check src/ tests/
+
       - name: Type check
         run: mypy src/ca2a_runtime/ src/ca2a_verify/
 

src/ca2a_runtime/attestation.py

@@ -70,9 +70,7 @@ def attest_channel(provider: BaseProvider, public_key: str, nonce: str) -> Chann
     )
 
 
-def offer_channel(
-    provider: BaseProvider, *, nonce: str
-) -> tuple[X25519PrivateKey, ChannelOffer]:
+def offer_channel(provider: BaseProvider, *, nonce: str) -> tuple[X25519PrivateKey, ChannelOffer]:
     """Generate a channel keypair and bind its public key into a report under ``nonce``.
 
     The private key is returned to the callee and, on hardware, never leaves the

src/ca2a_runtime/challenge.py

@@ -50,7 +50,9 @@ def _mac(secret: bytes, expiry: int, rand: str) -> str:
 def issue_challenge(secret: bytes, *, ttl_seconds: int = DEFAULT_TTL_SECONDS) -> str:
     """Issue a challenge that expires ``ttl_seconds`` from now."""
     if ttl_seconds <= 0:
-        raise ValueError("ttl_seconds must be positive; a challenge that never validates is not a challenge")
+        raise ValueError(
+            "ttl_seconds must be positive; a challenge that never validates is not a challenge"
+        )
     expiry = int(time.time()) + ttl_seconds
     rand = secrets.token_hex(_RANDOM_BYTES)
     return f"{_PREFIX}.{expiry}.{rand}.{_mac(secret, expiry, rand)}"

src/ca2a_runtime/cli.py

@@ -94,16 +94,12 @@ def _load_records(path: str) -> list[DelegationRecord]:
                         else frozenset(str(s) for s in item["effective_scope"])
                     ),
                     denial_reason=(
-                        None
-                        if item.get("denial_reason") is None
-                        else str(item["denial_reason"])
+                        None if item.get("denial_reason") is None else str(item["denial_reason"])
                     ),
                     # A record written before this field existed hashes as
                     # "not_offered", which is what its emitter could honestly have
                     # claimed: it never appraised a caller.
-                    caller_attestation=str(
-                        item.get("caller_attestation", CALLER_NOT_OFFERED)
-                    ),
+                    caller_attestation=str(item.get("caller_attestation", CALLER_NOT_OFFERED)),
                 )
             )
         except (KeyError, TypeError, ValueError) as exc:
@@ -169,7 +165,7 @@ def _cmd_start(args: argparse.Namespace) -> int:
         # The callee cannot claim an assurance level; the caller appraises the
         # offer. Say what that appraisal will be so it is not a surprise.
         print(
-            'note: software-only provider, callers appraise this channel key as '
+            "note: software-only provider, callers appraise this channel key as "
             'assurance="none" and the seal carries no hardware guarantee',
             file=sys.stderr,
         )

src/ca2a_runtime/config.py

@@ -14,9 +14,7 @@
 
 from ca2a_runtime.errors import ConfigError
 
-VALID_PROVIDERS = frozenset(
-    {"auto", "tpm", "sev-snp", "tdx", "opaque", "software-only"}
-)
+VALID_PROVIDERS = frozenset({"auto", "tpm", "sev-snp", "tdx", "opaque", "software-only"})
 VALID_ENFORCEMENT = frozenset({"enforcing", "advisory", "silent"})
 
 DEFAULT_LISTEN_ADDR = "127.0.0.1:8443"
@@ -93,9 +91,7 @@ def from_dict(cls, data: dict[str, Any]) -> Ca2aConfig:
             if not isinstance(raw_local, list) or not all(
                 isinstance(item, str) and item for item in raw_local
             ):
-                raise ConfigError(
-                    "local_policy must be a list of non-empty capability strings"
-                )
+                raise ConfigError("local_policy must be a list of non-empty capability strings")
             local_policy = frozenset(raw_local)
 
         bundle = data.get("policy_bundle_path")

src/ca2a_runtime/delegation/credential.py

@@ -124,9 +124,7 @@ def from_dict(cls, data: dict[str, Any]) -> DelegationCredential:
             raise InvalidCredential("malformed credential", detail=str(exc)) from exc
 
 
-def verify_chain(
-    chain: list[DelegationCredential], *, max_depth: int = 8
-) -> None:
+def verify_chain(chain: list[DelegationCredential], *, max_depth: int = 8) -> None:
     """Verify a root-to-leaf delegation chain, raising on the first violation.
 
     A well-formed chain of length N delegates from the root issuer down to the
@@ -147,9 +145,7 @@ def verify_chain(
         seen_ids.add(cred.credential_id)
 
         if cred.depth > max_depth:
-            raise DelegationDepthExceeded(
-                f"hop {i} depth {cred.depth} exceeds max {max_depth}"
-            )
+            raise DelegationDepthExceeded(f"hop {i} depth {cred.depth} exceeds max {max_depth}")
 
         if prev is None:
             if cred.parent_id is not None:
@@ -162,9 +158,7 @@ def verify_chain(
                     f"hop {i} parent_id does not match previous credential_id"
                 )
             if cred.issuer != prev.subject:
-                raise BrokenDelegationLink(
-                    f"hop {i} issuer is not the previous hop's subject"
-                )
+                raise BrokenDelegationLink(f"hop {i} issuer is not the previous hop's subject")
             if cred.depth != prev.depth + 1:
                 raise BrokenDelegationLink(f"hop {i} depth is not previous + 1")
             if not cred.scope.issubset(prev.scope):

src/ca2a_runtime/provenance.py

@@ -205,9 +205,7 @@ def verify_dag(records: list[DelegationRecord]) -> list[DelegationRecord]:
     return records
 
 
-def cross_check_chain(
-    records: list[DelegationRecord], chain: list[DelegationCredential]
-) -> None:
+def cross_check_chain(records: list[DelegationRecord], chain: list[DelegationCredential]) -> None:
     """Confirm a verified provenance chain lines up with a delegation chain.
 
     Ties provenance to authority: hop record ``i`` must reference credential

src/ca2a_runtime/tee/sev_snp.py

@@ -78,9 +78,7 @@ def snp_report_data(public_key: str, nonce: str) -> bytes:
     The 32-byte binding digest, zero-padded to the field width. See
     :mod:`ca2a_runtime.tee.binding` and ``docs/spec/attestation.md``.
     """
-    return pad_report_data(
-        derive_binding(SNP_PREFIX, public_key, nonce), TSM_REPORT_DATA_LEN
-    )
+    return pad_report_data(derive_binding(SNP_PREFIX, public_key, nonce), TSM_REPORT_DATA_LEN)
 
 
 @dataclass(frozen=True)

src/ca2a_runtime/tee/tdx.py

@@ -73,9 +73,7 @@ def tdx_report_data(public_key: str, nonce: str) -> bytes:
     The 32-byte binding digest, zero-padded to the field width. See
     :mod:`ca2a_runtime.tee.binding` and ``docs/spec/attestation.md``.
     """
-    return pad_report_data(
-        derive_binding(TDX_PREFIX, public_key, nonce), TSM_REPORT_DATA_LEN
-    )
+    return pad_report_data(derive_binding(TDX_PREFIX, public_key, nonce), TSM_REPORT_DATA_LEN)
 
 
 @dataclass(frozen=True)
@@ -151,7 +149,9 @@ def parse(cls, blob: bytes) -> TdxQuote:
         try:
             chain = x509.load_pem_x509_certificates(cert_bytes)
         except ValueError as exc:
-            raise AttestationFailed("could not parse PCK certificate chain", detail=str(exc)) from exc
+            raise AttestationFailed(
+                "could not parse PCK certificate chain", detail=str(exc)
+            ) from exc
 
         return cls(
             version=version,

src/ca2a_runtime/tee/tpm.py

@@ -203,8 +203,7 @@ def _require_host(cls) -> None:
             raise AttestationUnsupported(
                 "TPM quote generation requires a TPM device",
                 detail=(
-                    f"none of {', '.join(TPM_DEVICES)} exist; run on a host with a "
-                    "TPM 2.0 or vTPM"
+                    f"none of {', '.join(TPM_DEVICES)} exist; run on a host with a TPM 2.0 or vTPM"
                 ),
             )
         if not _tpm2_pytss_available():

src/ca2a_runtime/transport/server.py

@@ -83,7 +83,9 @@ def do_POST(self) -> None:
             return
         length = int(self.headers.get("Content-Length", "0") or "0")
         if length <= 0 or length > _MAX_BODY:
-            self._send_json(400, {"error": {"code": "BAD_REQUEST", "message": "invalid body length"}})
+            self._send_json(
+                400, {"error": {"code": "BAD_REQUEST", "message": "invalid body length"}}
+            )
             return
         try:
             message = json.loads(self.rfile.read(length))

src/ca2a_runtime/transport/wire.py

@@ -54,9 +54,7 @@ def serialize_error(err: CA2AError) -> dict[str, Any]:
     }
 
 
-def serialize_channel_offer(
-    offer: ChannelOffer, *, challenge: str | None = None
-) -> dict[str, Any]:
+def serialize_channel_offer(offer: ChannelOffer, *, challenge: str | None = None) -> dict[str, Any]:
     """Serialize a channel offer (a peer's attested channel key).
 
     Used in both directions, which is why it lives here rather than beside either

src/ca2a_verify/dag.py

@@ -169,9 +169,7 @@ def verify_trace_dag(
         except InvalidSignature as exc:
             raise TraceRecordInvalid(f"record {i} signature does not verify") from exc
         except ValueError as exc:
-            raise TraceRecordInvalid(
-                f"record {i} could not be verified", detail=str(exc)
-            ) from exc
+            raise TraceRecordInvalid(f"record {i} could not be verified", detail=str(exc)) from exc
 
         record_hash = trace_record_hash(record)
         if record_hash in seen_hashes:
@@ -202,9 +200,7 @@ def verify_trace_dag(
     )
 
 
-def cross_check_trace_dag(
-    records: list[dict[str, Any]], chain: list[DelegationCredential]
-) -> None:
+def cross_check_trace_dag(records: list[dict[str, Any]], chain: list[DelegationCredential]) -> None:
     """Tie a verified TRACE DAG to the delegation chain it should reflect.
 
     Confirms the DAG has one record per credential and that every non-root hop

src/ca2a_verify/sev_snp.py

@@ -28,9 +28,7 @@
 __all__ = ["SEV_GUEST_DEVICE", "verify_cert_chain", "verify_sev_snp_report"]
 
 
-def verify_cert_chain(
-    chain: list[x509.Certificate], trusted_roots: list[x509.Certificate]
-) -> None:
+def verify_cert_chain(chain: list[x509.Certificate], trusted_roots: list[x509.Certificate]) -> None:
     """Verify a leaf-to-root certificate chain against a set of trusted roots.
 
     ``chain`` is ordered leaf first (VCEK), root last (ARK). Delegates to
@@ -46,9 +44,7 @@ def verify_cert_chain(
     try:
         _shared_verify_cert_chain(chain, trusted_roots)
     except CertChainError as exc:
-        raise AttestationFailed(
-            "certificate chain verification failed", detail=str(exc)
-        ) from exc
+        raise AttestationFailed("certificate chain verification failed", detail=str(exc)) from exc
 
 
 def verify_sev_snp_report(

src/ca2a_verify/tdx.py

@@ -57,9 +57,7 @@ def verify_tdx_quote(
     quote = TdxQuote.parse(quote_bytes)
 
     if quote.tee_type != TEE_TYPE_TDX:
-        raise AttestationFailed(
-            "quote is not a TDX quote", detail=f"tee_type={quote.tee_type:#x}"
-        )
+        raise AttestationFailed("quote is not a TDX quote", detail=f"tee_type={quote.tee_type:#x}")
 
     # 1. PCK chain to a trusted Intel root.
     verify_cert_chain(quote.pck_chain, trusted_roots)

src/ca2a_verify/verify.py

@@ -48,7 +48,7 @@ def _parse_chain(data: Any) -> list[DelegationCredential]:
     if isinstance(data, dict) and "chain" in data:
         data = data["chain"]
     if not isinstance(data, list):
-        raise InvalidCredential("chain document must be a list or {\"chain\": [...]}")
+        raise InvalidCredential('chain document must be a list or {"chain": [...]}')
     return [DelegationCredential.from_dict(item) for item in data]