Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions src/model_signing/_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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.
Expand All @@ -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:
Expand All @@ -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)
Expand Down
45 changes: 45 additions & 0 deletions src/model_signing/_signing/sign_certificate.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand All @@ -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())]
Expand Down Expand Up @@ -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)})"
)
6 changes: 6 additions & 0 deletions src/model_signing/verifying.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand All @@ -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.
Expand All @@ -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
201 changes: 201 additions & 0 deletions tests/_signing/certificate_test.py
Original file line number Diff line number Diff line change
@@ -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)
Loading