diff --git a/CHANGELOG.md b/CHANGELOG.md index c81b18a3..f4cbc871 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,7 +10,8 @@ All versions prior to 1.0.0 are untracked. ## [Unreleased] ### Added --Added the `digest` subcommand to compute and print a model's digest. This enables other tools to easily pair the attestations with a model directory. +- Added the `digest` subcommand to compute and print a model's digest. This enables other tools to easily pair the attestations with a model directory. +- Added `--tsa-url` option for RFC 3161 timestamps in PKI signing methods. ### Changed - Standardized CLI flags to use hyphens (e.g., `--trust-config` instead of `--trust_config`). Underscore variants are still accepted for backwards compatibility via token normalization. diff --git a/README.md b/README.md index f1417a21..e8c25a21 100644 --- a/README.md +++ b/README.md @@ -165,6 +165,15 @@ And then we use the private key to sign. [...]$ model_signing sign key bert-base-uncased --private-key key.priv ``` +#### Timestamp Authority Support + +PKI signing methods support RFC 3161 timestamps via `--tsa-url`: + +```bash +[...]$ model_signing sign key bert-base-uncased --private-key key.priv \ + --tsa-url https://timestamp.sigstore.dev/api/v1/timestamp +``` + All signing methods support changing the signature name and location via the `--signature` flag: diff --git a/src/model_signing/_cli.py b/src/model_signing/_cli.py index 94da05be..bcf78027 100644 --- a/src/model_signing/_cli.py +++ b/src/model_signing/_cli.py @@ -155,6 +155,14 @@ def set_attribute(self, key, value): help="Whether to allow following symlinks when signing or verifying files.", ) +# Decorator for the commonly used option to set a TSA URL for PKI signing +_tsa_url_option = click.option( + "--tsa-url", + type=str, + metavar="TSA_URL", + help="URL of an RFC 3161 Timestamp Authority for trusted timestamps.", +) + def _resolve_ignore_paths( model_path: pathlib.Path, paths: Iterable[pathlib.Path] @@ -454,6 +462,7 @@ def _sign_sigstore( @_allow_symlinks_option @_write_signature_option @_private_key_option +@_tsa_url_option @click.option( "--password", type=str, @@ -468,6 +477,7 @@ def _sign_private_key( signature: pathlib.Path, private_key: pathlib.Path, password: str | None = None, + tsa_url: str | None = None, ) -> None: """Sign using a private key (paired with a public one). @@ -487,7 +497,7 @@ def _sign_private_key( model_path, list(ignore_paths) + [signature] ) model_signing.signing.Config().use_elliptic_key_signer( - private_key=private_key, password=password + private_key=private_key, password=password, tsa_url=tsa_url ).set_hashing_config( model_signing.hashing.Config() .set_ignored_paths(paths=ignored, ignore_git_paths=ignore_git_paths) @@ -507,6 +517,7 @@ def _sign_private_key( @_allow_symlinks_option @_write_signature_option @_pkcs11_uri_option +@_tsa_url_option def _sign_pkcs11_key( model_path: pathlib.Path, ignore_paths: Iterable[pathlib.Path], @@ -514,6 +525,7 @@ def _sign_pkcs11_key( allow_symlinks: bool, signature: pathlib.Path, pkcs11_uri: str, + tsa_url: str | None = None, ) -> None: """Sign using a private key using a PKCS #11 URI. @@ -533,7 +545,7 @@ def _sign_pkcs11_key( model_path, list(ignore_paths) + [signature] ) model_signing.signing.Config().use_pkcs11_signer( - pkcs11_uri=pkcs11_uri + pkcs11_uri=pkcs11_uri, tsa_url=tsa_url ).set_hashing_config( model_signing.hashing.Config() .set_ignored_paths(paths=ignored, ignore_git_paths=ignore_git_paths) @@ -555,6 +567,7 @@ def _sign_pkcs11_key( @_private_key_option @_signing_certificate_option @_certificate_root_of_trust_option +@_tsa_url_option def _sign_certificate( model_path: pathlib.Path, ignore_paths: Iterable[pathlib.Path], @@ -564,6 +577,7 @@ def _sign_certificate( private_key: pathlib.Path, signing_certificate: pathlib.Path, certificate_chain: Iterable[pathlib.Path], + tsa_url: str | None = None, ) -> None: """Sign using a certificate. @@ -589,6 +603,7 @@ def _sign_certificate( private_key=private_key, signing_certificate=signing_certificate, certificate_chain=certificate_chain, + tsa_url=tsa_url, ).set_hashing_config( model_signing.hashing.Config() .set_ignored_paths(paths=ignored, ignore_git_paths=ignore_git_paths) @@ -610,6 +625,7 @@ def _sign_certificate( @_pkcs11_uri_option @_signing_certificate_option @_certificate_root_of_trust_option +@_tsa_url_option def _sign_pkcs11_certificate( model_path: pathlib.Path, ignore_paths: Iterable[pathlib.Path], @@ -619,6 +635,7 @@ def _sign_pkcs11_certificate( pkcs11_uri: str, signing_certificate: pathlib.Path, certificate_chain: Iterable[pathlib.Path], + tsa_url: str | None = None, ) -> None: """Sign using a certificate. @@ -645,6 +662,7 @@ def _sign_pkcs11_certificate( pkcs11_uri=pkcs11_uri, signing_certificate=signing_certificate, certificate_chain=certificate_chain, + tsa_url=tsa_url, ).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..871eaf2c 100644 --- a/src/model_signing/_signing/sign_certificate.py +++ b/src/model_signing/_signing/sign_certificate.py @@ -46,6 +46,7 @@ def __init__( private_key_path: pathlib.Path, signing_certificate_path: pathlib.Path, certificate_chain_paths: Iterable[pathlib.Path], + tsa_url: str | None = None, ): """Initializes the signer with the key, certificate and trust chain. @@ -54,12 +55,13 @@ def __init__( signing_certificate_path: The path to the signing certificate. certificate_chain_paths: Paths to other certificates used to establish chain of trust. + tsa_url: Optional URL of an RFC 3161 Timestamp Authority. Raises: ValueError: Signing certificate's public key does not match the private key's public pair. """ - super().__init__(private_key_path) + super().__init__(private_key_path, tsa_url=tsa_url) self._signing_certificate = x509.load_pem_x509_certificate( signing_certificate_path.read_bytes() ) @@ -196,6 +198,10 @@ def _verify_certificates( The public key is extracted from the signing certificate from the chain of trust, after the chain is validated. It must match the public key from the key used during signing. + + If a TSA timestamp is present in the verification material, it will be + used as the verification time, allowing verification of signatures made + with certificates that have since expired. """ def _to_openssl_certificate(certificate_bytes, log_fingerprints): @@ -209,8 +215,15 @@ def _to_openssl_certificate(certificate_bytes, log_fingerprints): signing_chain.certificates[0].raw_bytes ) - max_signing_time = signing_certificate.not_valid_before_utc - self._store.set_time(max_signing_time) + tsa_timestamp = sigstore_pb.get_timestamp_from_bundle( + verification_material + ) + if tsa_timestamp is not None: + verification_time = tsa_timestamp + else: + verification_time = signing_certificate.not_valid_before_utc + + self._store.set_time(verification_time) trust_chain_ssl = [ _to_openssl_certificate( diff --git a/src/model_signing/_signing/sign_ec_key.py b/src/model_signing/_signing/sign_ec_key.py index 08cde115..e268377c 100644 --- a/src/model_signing/_signing/sign_ec_key.py +++ b/src/model_signing/_signing/sign_ec_key.py @@ -86,18 +86,23 @@ class Signer(sigstore_pb.Signer): """Signer using an elliptic curve private key.""" def __init__( - self, private_key_path: pathlib.Path, password: str | None = None + self, + private_key_path: pathlib.Path, + password: str | None = None, + tsa_url: str | None = None, ): """Initializes the signer with the private key and optional password. Args: private_key_path: The path to the PEM encoded private key. password: Optional password for the private key. + tsa_url: Optional URL of an RFC 3161 Timestamp Authority. """ self._private_key = serialization.load_pem_private_key( private_key_path.read_bytes(), password ) _check_supported_ec_key(self._private_key.public_key()) + self._tsa_url = tsa_url @override def sign(self, payload: signing.Payload) -> signing.Signature: @@ -105,14 +110,13 @@ def sign(self, payload: signing.Payload) -> signing.Signature: "utf-8" ) + signature_bytes = self._private_key.sign( + sigstore_pb.pae(raw_payload), + ec.ECDSA(get_ec_key_hash(self._private_key.public_key())), + ) + raw_signature = intoto_pb.Signature( - sig=base64.b64encode( - self._private_key.sign( - sigstore_pb.pae(raw_payload), - ec.ECDSA(get_ec_key_hash(self._private_key.public_key())), - ) - ), - keyid="", + sig=base64.b64encode(signature_bytes), keyid="" ) envelope = intoto_pb.Envelope( @@ -121,10 +125,17 @@ def sign(self, payload: signing.Payload) -> signing.Signature: signatures=[raw_signature], ) + verification_material = self._get_verification_material() + + if self._tsa_url: + verification_material.timestamp_verification_data = ( + sigstore_pb.request_timestamp(signature_bytes, self._tsa_url) + ) + return sigstore_pb.Signature( bundle_pb.Bundle( media_type=sigstore_pb._BUNDLE_MEDIA_TYPE, - verification_material=self._get_verification_material(), + verification_material=verification_material, dsse_envelope=envelope, ) ) diff --git a/src/model_signing/_signing/sign_pkcs11.py b/src/model_signing/_signing/sign_pkcs11.py index b4213251..c5028406 100644 --- a/src/model_signing/_signing/sign_pkcs11.py +++ b/src/model_signing/_signing/sign_pkcs11.py @@ -81,9 +81,13 @@ class Signer(sigstore_pb.Signer): """Signer using PKCS #11 URIs with elliptic curves keys.""" def __init__( - self, pkcs11_uri: str, module_paths: Iterable[str] = frozenset() + self, + pkcs11_uri: str, + module_paths: Iterable[str] = frozenset(), + tsa_url: str | None = None, ): self.session = None + self._tsa_url = tsa_url self.pkcs11_uri = Pkcs11URI() self.pkcs11_uri.parse(pkcs11_uri) @@ -183,10 +187,17 @@ def sign(self, payload: signing.Payload) -> signing.Signature: signatures=[raw_signature], ) + verification_material = self._get_verification_material() + + if self._tsa_url: + verification_material.timestamp_verification_data = ( + sigstore_pb.request_timestamp(sig, self._tsa_url) + ) + return sigstore_pb.Signature( bundle_pb.Bundle( media_type=sigstore_pb._BUNDLE_MEDIA_TYPE, - verification_material=self._get_verification_material(), + verification_material=verification_material, dsse_envelope=envelope, ) ) @@ -217,6 +228,7 @@ def __init__( signing_certificate_path: pathlib.Path, certificate_chain_paths: Iterable[pathlib.Path], module_paths: Iterable[str] = frozenset(), + tsa_url: str | None = None, ): """Initializes the signer with the key, certificate and trust chain. @@ -226,12 +238,13 @@ def __init__( certificate_chain_paths: Paths to other certificates used to establish chain of trust. module_paths: Paths to PKCS #11 modules to load. + tsa_url: Optional URL of an RFC 3161 Timestamp Authority. Raises: ValueError: Signing certificate's public key does not match the private key's public pair. """ - super().__init__(pkcs11_uri, module_paths=module_paths) + super().__init__(pkcs11_uri, module_paths=module_paths, tsa_url=tsa_url) self._signing_certificate = x509.load_pem_x509_certificate( signing_certificate_path.read_bytes() ) diff --git a/src/model_signing/_signing/sign_sigstore_pb.py b/src/model_signing/_signing/sign_sigstore_pb.py index 800e0036..9053fa0d 100644 --- a/src/model_signing/_signing/sign_sigstore_pb.py +++ b/src/model_signing/_signing/sign_sigstore_pb.py @@ -22,17 +22,66 @@ """ import abc +import base64 +from datetime import datetime import json +import logging import pathlib import sys from typing import cast +import urllib.error +import urllib.request +import rfc3161_client from sigstore_models.bundle import v1 as bundle_pb from typing_extensions import override from model_signing._signing import signing +logger = logging.getLogger(__name__) + + +_TSA_CLIENT_TIMEOUT: int = 5 +_USER_AGENT: str = "model-signing" + + +def _request_timestamp( + url: str, signature: bytes +) -> rfc3161_client.TimeStampResponse: + """Request a timestamp from a TSA for the given signature.""" + try: + timestamp_request = ( + rfc3161_client.TimestampRequestBuilder() + .hash_algorithm(rfc3161_client.base.HashAlgorithm.SHA256) + .data(signature) + .nonce(nonce=True) + .build() + ) + except ValueError as error: + raise ValueError(f"invalid TSA request: {error}") from error + + req = urllib.request.Request( + url, + data=timestamp_request.as_bytes(), + headers={ + "Content-Type": "application/timestamp-query", + "User-Agent": _USER_AGENT, + }, + ) + + try: + with urllib.request.urlopen(req, timeout=_TSA_CLIENT_TIMEOUT) as resp: + response_bytes = resp.read() + except urllib.error.URLError as error: + raise ValueError(f"TSA request failed: {error}") from error + + try: + return rfc3161_client.decode_timestamp_response(response_bytes) + except ValueError as error: + raise ValueError(f"invalid TSA response: {error}") from error + + if sys.version_info >= (3, 11): from typing import Self else: @@ -163,3 +212,52 @@ def _verify_bundle(self, bundle: bundle_pb.Bundle) -> tuple[str, bytes]: Since the bundle is generated via proto, we need to do more checks to replace what `verify_dsse` from `sigstore_python` does. """ + + +def request_timestamp( + signature_bytes: bytes, tsa_url: str +) -> bundle_pb.TimestampVerificationData: + """Requests a timestamp from a TSA and returns verification data. + + Args: + signature_bytes: The signature bytes to timestamp. + tsa_url: The URL of the RFC 3161 Timestamp Authority. + + Returns: + TimestampVerificationData to include in the bundle. + """ + response = _request_timestamp(tsa_url, signature_bytes) + + return bundle_pb.TimestampVerificationData( + rfc3161_timestamps=[ + bundle_pb.RFC3161SignedTimestamp( + signed_timestamp=base64.b64encode(response.as_bytes()) + ) + ] + ) + + +def get_timestamp_from_bundle( + verification_material: bundle_pb.VerificationMaterial, +) -> datetime | None: + """Extracts the timestamp from the bundle's verification material. + + Args: + verification_material: The bundle's verification material. + + Returns: + The timestamp datetime if present and valid, None otherwise. + """ + ts_data = verification_material.timestamp_verification_data + if not ts_data or not ts_data.rfc3161_timestamps: + return None + + ts = ts_data.rfc3161_timestamps[0] + try: + response = rfc3161_client.decode_timestamp_response( + base64.b64decode(ts.signed_timestamp) + ) + return response.tst_info.gen_time + except (ValueError, KeyError) as error: + logger.warning(f"Failed to decode timestamp from bundle: {error}") + return None diff --git a/src/model_signing/signing.py b/src/model_signing/signing.py index 4de79be2..b8e9735c 100644 --- a/src/model_signing/signing.py +++ b/src/model_signing/signing.py @@ -187,7 +187,11 @@ def use_sigstore_signer( return self def use_elliptic_key_signer( - self, *, private_key: hashing.PathLike, password: str | None = None + self, + *, + private_key: hashing.PathLike, + password: str | None = None, + tsa_url: str | None = None, ) -> Self: """Configures the signing to be performed using elliptic curve keys. @@ -197,11 +201,15 @@ def use_elliptic_key_signer( Args: private_key: The path to the private key to use for signing. password: An optional password for the key, if encrypted. + tsa_url: Optional URL of an RFC 3161 Timestamp Authority. When + provided, the signature will include a trusted timestamp. Return: The new signing configuration. """ - self._signer = ec_key.Signer(pathlib.Path(private_key), password) + self._signer = ec_key.Signer( + pathlib.Path(private_key), password, tsa_url=tsa_url + ) return self def use_certificate_signer( @@ -210,6 +218,7 @@ def use_certificate_signer( private_key: hashing.PathLike, signing_certificate: hashing.PathLike, certificate_chain: Iterable[hashing.PathLike], + tsa_url: str | None = None, ) -> Self: """Configures the signing to be performed using signing certificates. @@ -221,6 +230,8 @@ def use_certificate_signer( signing_certificate: The path to the signing certificate. certificate_chain: Optional paths to other certificates to establish a chain of trust. + tsa_url: Optional URL of an RFC 3161 Timestamp Authority. When + provided, the signature will include a trusted timestamp. Return: The new signing configuration. @@ -229,11 +240,16 @@ def use_certificate_signer( pathlib.Path(private_key), pathlib.Path(signing_certificate), [pathlib.Path(c) for c in certificate_chain], + tsa_url=tsa_url, ) return self def use_pkcs11_signer( - self, *, pkcs11_uri: str, module_paths: Iterable[str] = frozenset() + self, + *, + pkcs11_uri: str, + module_paths: Iterable[str] = frozenset(), + tsa_url: str | None = None, ) -> Self: """Configures the signing to be performed using PKCS #11. @@ -243,6 +259,8 @@ def use_pkcs11_signer( Args: pkcs11_uri: The PKCS11 URI. module_paths: Optional list of paths of PKCS #11 modules. + tsa_url: Optional URL of an RFC 3161 Timestamp Authority. When + provided, the signature will include a trusted timestamp. Return: The new signing configuration. @@ -254,7 +272,7 @@ def use_pkcs11_signer( "PKCS #11 functionality requires the 'pkcs11' extra. " "Install with 'pip install model-signing[pkcs11]'." ) from e - self._signer = pkcs11.Signer(pkcs11_uri, module_paths) + self._signer = pkcs11.Signer(pkcs11_uri, module_paths, tsa_url=tsa_url) return self def use_pkcs11_certificate_signer( @@ -264,6 +282,7 @@ def use_pkcs11_certificate_signer( signing_certificate: pathlib.Path, certificate_chain: Iterable[pathlib.Path], module_paths: Iterable[str] = frozenset(), + tsa_url: str | None = None, ) -> Self: """Configures the signing to be performed using signing certificates. @@ -276,6 +295,8 @@ def use_pkcs11_certificate_signer( certificate_chain: Optional paths to other certificates to establish a chain of trust. module_paths: Optional list of paths of PKCS #11 modules. + tsa_url: Optional URL of an RFC 3161 Timestamp Authority. When + provided, the signature will include a trusted timestamp. Return: The new signing configuration. @@ -293,5 +314,6 @@ def use_pkcs11_certificate_signer( signing_certificate, certificate_chain, module_paths=module_paths, + tsa_url=tsa_url, ) return self diff --git a/tests/api_test.py b/tests/api_test.py index 1195d9f1..1d607e6d 100644 --- a/tests/api_test.py +++ b/tests/api_test.py @@ -429,3 +429,31 @@ def test_sign_and_verify_sharded(self, base_path, populate_tmpdir): signature, ignore_git_paths, ["model.sig", "ignored"] ) assert get_model_name(signature) == os.path.basename(model_path) + + +class TestTSASigning: + @pytest.mark.integration + def test_sign_and_verify_with_tsa(self, base_path, populate_tmpdir): + os.chdir(base_path) + + model_path = populate_tmpdir + signature = Path(model_path / "model.sig") + private_key = Path(TESTDATA / "keys/certificate/signing-key.pem") + public_key = Path(TESTDATA / "keys/certificate/signing-key-pub.pem") + + signing.Config().use_elliptic_key_signer( + private_key=private_key, + tsa_url="https://timestamp.sigstore.dev/api/v1/timestamp", + ).set_hashing_config( + hashing.Config().set_ignored_paths( + paths=[signature], ignore_git_paths=False + ) + ).sign(model_path, signature) + + verifying.Config().use_elliptic_key_verifier( + public_key=public_key + ).set_hashing_config( + hashing.Config().set_ignored_paths( + paths=[signature], ignore_git_paths=False + ) + ).verify(model_path, signature)