From 87915ae0efb39f6b207698a449b541c52b8a5b82 Mon Sep 17 00:00:00 2001 From: Junyuan Zeng Date: Thu, 16 Jul 2026 14:00:58 -0700 Subject: [PATCH] feat(verify): pin signer identity via --san-uri on verify certificate Adds --san-uri (repeatable) to `model_signing verify certificate` and an equivalent `expected_san_uris` kwarg on `use_certificate_verifier`. When set, verification additionally requires that every listed URI appear in the leaf certificate's SubjectAlternativeName. The check runs on the leaf embedded in the Sigstore bundle, after chain-of-trust verification succeeds. Default behavior is unchanged when the flag is omitted. Motivation: SPIFFE SVIDs carry the workload identity in the URI SAN (X509-SVID spec requires exactly one URI SAN, and it MUST be the SPIFFE ID). Chain-of-trust alone proves the CA vouched for *some* leaf, not *which* leaf. Without a URI-SAN check, any certificate issued by the SPIRE trust bundle (or by a shared internal PKI) can produce accepted signatures. This flag turns the SPIFFE-required verification step into a single CLI option, so callers no longer have to write ad-hoc post-verify scripts that re-parse the bundle. Signed-off-by: Junyuan Zeng --- src/model_signing/_cli.py | 24 +++ .../_signing/sign_certificate.py | 45 ++++ src/model_signing/verifying.py | 6 + tests/_signing/certificate_test.py | 201 ++++++++++++++++++ 4 files changed, 276 insertions(+) create mode 100644 tests/_signing/certificate_test.py diff --git a/src/model_signing/_cli.py b/src/model_signing/_cli.py index 299d7342..5d529058 100644 --- a/src/model_signing/_cli.py +++ b/src/model_signing/_cli.py @@ -848,6 +848,21 @@ def _verify_private_key( show_default=True, help="Log SHA256 fingerprints of all certificates.", ) +@click.option( + "--san-uri", + "san_uris", + type=str, + multiple=True, + default=(), + help=( + "Require this URI to appear in the leaf certificate's " + "SubjectAlternativeName. Repeat to require multiple. Pins signer " + "identity so a different certificate issued by the same CA cannot " + "produce accepted signatures. This is the mechanism SPIFFE SVIDs use " + "to carry a workload identity (spiffe://...) and is where SPIFFE-aware " + "verifiers are required to check." + ), +) @_ignore_unsigned_files_option def _verify_certificate( model_path: pathlib.Path, @@ -857,6 +872,7 @@ def _verify_certificate( allow_symlinks: bool, certificate_chain: Iterable[pathlib.Path], log_fingerprints: bool, + san_uris: tuple[str, ...], ignore_unsigned_files: bool, ) -> None: """Verify using a certificate. @@ -870,6 +886,13 @@ def _verify_certificate( certificate chain, using `--certificate_chain` (this option can be repeated as needed, or all certificates could be placed in a single file). + To bind the signature to a specific signer identity (and not merely to + "some certificate issued by this CA"), pass `--san-uri`. The check runs + against the leaf certificate embedded in the bundle, after chain + verification succeeds. SPIFFE SVIDs carry the SPIFFE ID in the URI SAN; + passing the expected `spiffe://` URI is the SPIFFE-mandated verification + step. + Note that we don't offer certificate and key management protocols. """ if log_fingerprints: @@ -882,6 +905,7 @@ def _verify_certificate( model_signing.verifying.Config().use_certificate_verifier( certificate_chain=certificate_chain, log_fingerprints=log_fingerprints, + expected_san_uris=san_uris, ).set_hashing_config( model_signing.hashing.Config() .set_ignored_paths(paths=ignored, ignore_git_paths=ignore_git_paths) diff --git a/src/model_signing/_signing/sign_certificate.py b/src/model_signing/_signing/sign_certificate.py index e6fa6721..74b017f5 100644 --- a/src/model_signing/_signing/sign_certificate.py +++ b/src/model_signing/_signing/sign_certificate.py @@ -130,6 +130,7 @@ def __init__( self, certificate_chain_paths: Iterable[pathlib.Path] = frozenset(), log_fingerprints: bool = False, + expected_san_uris: Iterable[str] = frozenset(), ): """Initializes the verifier with the list of certificates to use. @@ -139,8 +140,15 @@ def __init__( in which case we would use the root certificates from the operating system, as per `certifi.where()`. log_fingerprints: Log the fingerprints of certificates + expected_san_uris: If non-empty, verification additionally requires + that every listed URI appear in the leaf certificate's + SubjectAltName URI entries. This binds the signature to a signer + identity (e.g. a SPIFFE ID, which per RFC-compliant SVIDs is + always carried in the URI SAN) rather than trusting any + certificate issued under the CA. """ self._log_fingerprints = log_fingerprints + self._expected_san_uris = frozenset(expected_san_uris) if not certificate_chain_paths: certificate_chain_paths = [pathlib.Path(certifi.where())] @@ -251,4 +259,41 @@ def _to_openssl_certificate(certificate_bytes, log_fingerprints): if not can_use_for_signing: raise ValueError("Signing certificate cannot be used for signing") + self._verify_san_identity(signing_certificate) + return signing_certificate.public_key() + + def _verify_san_identity( + self, signing_certificate: x509.Certificate + ) -> None: + """Assert the leaf's SubjectAltName carries every expected URI. + + Chain-of-trust proves the CA vouched for *some* leaf; it does not tell + us *which* leaf. If the caller declared expected SAN URIs (e.g. a + SPIFFE ID), the signing certificate embedded in the bundle must carry + them, otherwise a different (but still CA-issued) key could produce + accepted signatures. + """ + if not self._expected_san_uris: + return + + try: + san = signing_certificate.extensions.get_extension_for_class( + x509.SubjectAlternativeName + ).value + except x509.ExtensionNotFound as err: + raise ValueError( + "Signing certificate has no SubjectAlternativeName; cannot " + "verify expected signer identity." + ) from err + + actual_uris = frozenset( + san.get_values_for_type(x509.UniformResourceIdentifier) + ) + missing_uris = self._expected_san_uris - actual_uris + if missing_uris: + raise ValueError( + "Signing certificate SubjectAltName is missing expected " + f"URI(s): {sorted(missing_uris)} " + f"(present: {sorted(actual_uris)})" + ) diff --git a/src/model_signing/verifying.py b/src/model_signing/verifying.py index 87303490..6aca204e 100644 --- a/src/model_signing/verifying.py +++ b/src/model_signing/verifying.py @@ -273,6 +273,7 @@ def use_certificate_verifier( *, certificate_chain: Iterable[hashing.PathLike] = frozenset(), log_fingerprints: bool = False, + expected_san_uris: Iterable[str] = frozenset(), ) -> Self: """Configures the verification of signatures generated by a certificate. @@ -283,6 +284,10 @@ def use_certificate_verifier( certificate_chain: Certificate chain to establish root of trust. If empty, the operating system's one is used. log_fingerprints: Log certificates' SHA256 fingerprints + expected_san_uris: Optional URIs that must appear in the leaf + certificate's SubjectAltName. Binds the signature to a specific + signer identity (e.g. a SPIFFE ID) in addition to + chain-of-trust. Return: The new verification configuration. @@ -291,5 +296,6 @@ def use_certificate_verifier( self._verifier = certificate.Verifier( [pathlib.Path(c) for c in certificate_chain], log_fingerprints=log_fingerprints, + expected_san_uris=expected_san_uris, ) return self diff --git a/tests/_signing/certificate_test.py b/tests/_signing/certificate_test.py new file mode 100644 index 00000000..3d1c1c2d --- /dev/null +++ b/tests/_signing/certificate_test.py @@ -0,0 +1,201 @@ +# Copyright 2026 The Sigstore Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests for SubjectAltName identity pinning in the certificate verifier.""" + +import datetime +import pathlib + +from cryptography import x509 +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.x509 import oid +import pytest + +from model_signing import hashing +from model_signing import signing +from model_signing import verifying + + +_SAN_URI = "spiffe://demo.example.com/signer/demo" +_OTHER_URI = "spiffe://demo.example.com/signer/other" + + +def _issue_chain( + tmp_path: pathlib.Path, + san_uris: list[str], +) -> tuple[pathlib.Path, pathlib.Path, pathlib.Path]: + """Issue a self-signed CA + a leaf cert with the requested URI SANs. + + Returns (private_key_pem, leaf_cert_pem, ca_cert_pem). + """ + now = datetime.datetime.now(datetime.timezone.utc) + ca_key = ec.generate_private_key(ec.SECP256R1()) + ca_name = x509.Name([ + x509.NameAttribute(oid.NameOID.COMMON_NAME, "Demo Root CA"), + ]) + ca_cert = ( + x509.CertificateBuilder() + .subject_name(ca_name) + .issuer_name(ca_name) + .public_key(ca_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - datetime.timedelta(minutes=1)) + .not_valid_after(now + datetime.timedelta(days=365)) + .add_extension( + x509.BasicConstraints(ca=True, path_length=None), critical=True + ) + .sign(private_key=ca_key, algorithm=hashes.SHA256()) + ) + + leaf_key = ec.generate_private_key(ec.SECP256R1()) + leaf_name = x509.Name([ + x509.NameAttribute(oid.NameOID.COMMON_NAME, "demo-signer"), + ]) + san = x509.SubjectAlternativeName( + [x509.UniformResourceIdentifier(u) for u in san_uris] + ) + builder = ( + x509.CertificateBuilder() + .subject_name(leaf_name) + .issuer_name(ca_cert.subject) + .public_key(leaf_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - datetime.timedelta(minutes=1)) + .not_valid_after(now + datetime.timedelta(days=30)) + .add_extension( + x509.KeyUsage( + digital_signature=True, + content_commitment=False, + key_encipherment=False, + data_encipherment=False, + key_agreement=False, + key_cert_sign=False, + crl_sign=False, + encipher_only=False, + decipher_only=False, + ), + critical=True, + ) + .add_extension( + x509.ExtendedKeyUsage([oid.ExtendedKeyUsageOID.CODE_SIGNING]), + critical=False, + ) + ) + if san_uris: + builder = builder.add_extension(san, critical=False) + leaf_cert = builder.sign(private_key=ca_key, algorithm=hashes.SHA256()) + + key_path = tmp_path / "leaf.key" + key_path.write_bytes( + leaf_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + ) + leaf_path = tmp_path / "leaf.cert" + leaf_path.write_bytes( + leaf_cert.public_bytes(encoding=serialization.Encoding.PEM) + ) + ca_path = tmp_path / "ca.cert" + ca_path.write_bytes( + ca_cert.public_bytes(encoding=serialization.Encoding.PEM) + ) + return key_path, leaf_path, ca_path + + +def _sign_blob( + tmp_path: pathlib.Path, + key_path: pathlib.Path, + leaf_path: pathlib.Path, + ca_path: pathlib.Path, +) -> tuple[pathlib.Path, pathlib.Path]: + """Sign a small model and return (model_path, signature_path).""" + model_path = tmp_path / "model.bin" + model_path.write_bytes(b"hello, model_signing") + signature = tmp_path / "model.sig" + signing.Config().use_certificate_signer( + private_key=key_path, + signing_certificate=leaf_path, + certificate_chain=[ca_path], + ).set_hashing_config( + hashing.Config().set_ignored_paths(paths=[signature]) + ).sign(model_path, signature) + return model_path, signature + + +class TestVerifyCertificateSanIdentity: + """Identity pinning via `expected_san_uris` (e.g. SPIFFE SVID URI SAN).""" + + def test_matching_uri_verifies(self, tmp_path): + key, leaf, ca = _issue_chain(tmp_path, [_SAN_URI]) + model, sig = _sign_blob(tmp_path, key, leaf, ca) + verifying.Config().use_certificate_verifier( + certificate_chain=[ca], expected_san_uris=[_SAN_URI] + ).set_hashing_config( + hashing.Config().set_ignored_paths(paths=[sig]) + ).verify(model, sig) + + def test_no_expected_identity_still_verifies(self, tmp_path): + # Back-compat: without pinning, the check is skipped entirely. + key, leaf, ca = _issue_chain(tmp_path, [_SAN_URI]) + model, sig = _sign_blob(tmp_path, key, leaf, ca) + verifying.Config().use_certificate_verifier( + certificate_chain=[ca] + ).set_hashing_config( + hashing.Config().set_ignored_paths(paths=[sig]) + ).verify(model, sig) + + def test_wrong_uri_rejected(self, tmp_path): + key, leaf, ca = _issue_chain(tmp_path, [_SAN_URI]) + model, sig = _sign_blob(tmp_path, key, leaf, ca) + with pytest.raises(ValueError, match="missing expected URI"): + verifying.Config().use_certificate_verifier( + certificate_chain=[ca], expected_san_uris=[_OTHER_URI] + ).set_hashing_config( + hashing.Config().set_ignored_paths(paths=[sig]) + ).verify(model, sig) + + def test_no_san_in_leaf_rejected_when_pinning_requested(self, tmp_path): + key, leaf, ca = _issue_chain(tmp_path, []) + model, sig = _sign_blob(tmp_path, key, leaf, ca) + with pytest.raises(ValueError, match="no SubjectAlternativeName"): + verifying.Config().use_certificate_verifier( + certificate_chain=[ca], expected_san_uris=[_SAN_URI] + ).set_hashing_config( + hashing.Config().set_ignored_paths(paths=[sig]) + ).verify(model, sig) + + def test_cross_signer_attack_rejected(self, tmp_path): + """A different leaf under the same CA must not satisfy identity pin.""" + good = tmp_path / "good" + good.mkdir() + bad = tmp_path / "bad" + bad.mkdir() + # We can't share the CA private key across helper calls, so we + # simulate the shared-CA scenario by issuing two independent CAs and + # verifying the bad-signer bundle against the bad-signer CA (chain + # verification succeeds), while pinning the good-signer URI + # (identity check must fail). + _issue_chain(good, [_SAN_URI]) + bad_key, bad_leaf, bad_ca = _issue_chain(bad, [_OTHER_URI]) + model, sig = _sign_blob(bad, bad_key, bad_leaf, bad_ca) + with pytest.raises(ValueError, match="missing expected URI"): + verifying.Config().use_certificate_verifier( + certificate_chain=[bad_ca], expected_san_uris=[_SAN_URI] + ).set_hashing_config( + hashing.Config().set_ignored_paths(paths=[sig]) + ).verify(model, sig)