From 1666c04203b2d398e32c2ad87a9ccaa41040c557 Mon Sep 17 00:00:00 2001 From: Renata Date: Wed, 8 Jul 2026 00:10:25 +0200 Subject: [PATCH 1/6] feat: detect ECDSA/RSA signing scheme from key material _from_crypto() now derives the scheme (RSA or ECDSA P-256) from the actual key bytes instead of assuming RSA, so keys are loaded correctly regardless of type without callers needing to know it in advance (e.g. reading a YubiKey's public key). Adds test_ecdsa_keys.py covering detection, curve/scheme validation, and a sign/verify round trip. --- taf/constants.py | 1 + taf/tests/tuf/test_keys/test_ecdsa_keys.py | 92 ++++++++++++++++++++++ taf/tuf/keys.py | 52 ++++++++---- 3 files changed, 128 insertions(+), 17 deletions(-) create mode 100644 taf/tests/tuf/test_keys/test_ecdsa_keys.py diff --git a/taf/constants.py b/taf/constants.py index 310c8262..06eea119 100644 --- a/taf/constants.py +++ b/taf/constants.py @@ -7,6 +7,7 @@ DEFAULT_RSA_SIGNATURE_SCHEME = "rsa-pkcs1v15-sha256" +DEFAULT_ECDSA_SIGNATURE_SCHEME = "ecdsa-sha2-nistp256" CAPSTONE = "capstone" PROTECTED_DIRECTORY_NAME = "protected" diff --git a/taf/tests/tuf/test_keys/test_ecdsa_keys.py b/taf/tests/tuf/test_keys/test_ecdsa_keys.py new file mode 100644 index 00000000..86a9772b --- /dev/null +++ b/taf/tests/tuf/test_keys/test_ecdsa_keys.py @@ -0,0 +1,92 @@ +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ec import SECP256R1, SECP384R1 +from cryptography.hazmat.primitives.asymmetric.ec import ( + generate_private_key as generate_ec_private_key, +) + +from taf.constants import DEFAULT_ECDSA_SIGNATURE_SCHEME, DEFAULT_RSA_SIGNATURE_SCHEME +from taf.tuf.keys import get_sslib_key_from_value, load_signer_from_pem + + +def _ec_keypair_pem(curve=SECP256R1()): + private_key = generate_ec_private_key(curve) + private_pem = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + public_pem = private_key.public_key().public_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PublicFormat.SubjectPublicKeyInfo, + ) + return private_pem, public_pem.decode() + + +def test_get_sslib_key_from_value_detects_ecdsa_with_default_rsa_scheme(): + """Callers that don't know the key type ahead of time (e.g. reading a + YubiKey's public key) pass no explicit scheme and rely on the RSA + default. An EC P-256 key must still be tagged correctly rather than + silently mislabeled as RSA or rejected. + """ + _, public_pem = _ec_keypair_pem() + + key = get_sslib_key_from_value(public_pem) + + assert key.keytype == "ecdsa" + assert key.scheme == DEFAULT_ECDSA_SIGNATURE_SCHEME + + +def test_get_sslib_key_from_value_accepts_explicit_ecdsa_scheme(): + _, public_pem = _ec_keypair_pem() + + key = get_sslib_key_from_value(public_pem, DEFAULT_ECDSA_SIGNATURE_SCHEME) + + assert key.keytype == "ecdsa" + assert key.scheme == DEFAULT_ECDSA_SIGNATURE_SCHEME + + +def test_get_sslib_key_from_value_rejects_unsupported_curve(): + _, public_pem = _ec_keypair_pem(SECP384R1()) + + with pytest.raises(ValueError): + get_sslib_key_from_value(public_pem) + + +def test_get_sslib_key_from_value_rejects_mismatched_scheme_for_ecdsa_key(): + _, public_pem = _ec_keypair_pem() + + with pytest.raises(ValueError): + get_sslib_key_from_value(public_pem, "rsassa-pss-sha256") + + +def test_ecdsa_signer_round_trip(): + """An EC private key loaded through the same RSA-default-scheme path + used everywhere else must produce a signer that verifies against the + corresponding public key loaded the same way. + """ + private_pem, public_pem = _ec_keypair_pem() + + signer = load_signer_from_pem(private_pem) + public_key = get_sslib_key_from_value(public_pem) + + assert signer.public_key.keyid == public_key.keyid + + signature = signer.sign(b"DATA") + public_key.verify_signature(signature, b"DATA") + + with pytest.raises(Exception): + public_key.verify_signature(signature, b"NOT DATA") + + +def test_rsa_scheme_default_still_used_for_rsa_keys(): + """Regression guard: the RSA path must be entirely unaffected by the + EC-detection branch.""" + from taf.tuf.keys import generate_rsa_keypair + + _, public_pem = generate_rsa_keypair(key_size=2048) + + key = get_sslib_key_from_value(public_pem.decode()) + + assert key.keytype == "rsa" + assert key.scheme == DEFAULT_RSA_SIGNATURE_SCHEME diff --git a/taf/tuf/keys.py b/taf/tuf/keys.py index 6bca7af2..a998d741 100644 --- a/taf/tuf/keys.py +++ b/taf/tuf/keys.py @@ -17,11 +17,15 @@ load_pem_public_key, ) from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey +from cryptography.hazmat.primitives.asymmetric.ec import ( + EllipticCurvePublicKey, + SECP256R1, +) from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import rsa -from taf.constants import DEFAULT_RSA_SIGNATURE_SCHEME +from taf.constants import DEFAULT_ECDSA_SIGNATURE_SCHEME, DEFAULT_RSA_SIGNATURE_SCHEME def generate_rsa_keypair(key_size=3072, password=None) -> Tuple[bytes, bytes]: @@ -101,12 +105,31 @@ def _get_legacy_keyid(key: SSlibKey) -> str: return hasher.hexdigest() -def _from_crypto(pub: RSAPublicKey, scheme=DEFAULT_RSA_SIGNATURE_SCHEME) -> SSlibKey: +def _from_crypto( + pub: Union[RSAPublicKey, EllipticCurvePublicKey], + scheme=DEFAULT_RSA_SIGNATURE_SCHEME, +) -> SSlibKey: """Converts pyca/cryptography public key to SSlibKey with default signing - scheme and legacy keyid.""" + scheme and legacy keyid. + + Detects the key's actual type (RSA or ECDSA P-256) from the key material + itself. This lets callers that only pass the historical RSA default + scheme (because they don't know ahead of time what type of key they're + about to load, e.g. reading a YubiKey's public key) still get back a + correctly scheme-tagged key when it turns out to be an EC key. + """ # securesystemslib does not (yet) check if keytype and scheme are compatible # https://github.com/secure-systems-lab/securesystemslib/issues/766 - if not isinstance(pub, RSAPublicKey): + if isinstance(pub, RSAPublicKey): + pass + elif isinstance(pub, EllipticCurvePublicKey): + if not isinstance(pub.curve, SECP256R1): + raise ValueError(f"unsupported EC curve '{pub.curve.name}'") + if scheme in (DEFAULT_RSA_SIGNATURE_SCHEME, DEFAULT_ECDSA_SIGNATURE_SCHEME): + scheme = DEFAULT_ECDSA_SIGNATURE_SCHEME + else: + raise ValueError(f"scheme '{scheme}' not valid for an ecdsa key") + else: raise ValueError(f"keytype '{type(pub)}' not supported") key = SSlibKey.from_crypto(pub, scheme=scheme) # FIXME: include the 'keyid_hash_algorithms' entry in the key portion @@ -117,17 +140,14 @@ def _from_crypto(pub: RSAPublicKey, scheme=DEFAULT_RSA_SIGNATURE_SCHEME) -> SSli return key -def load_public_key_from_file( - path: Union[str, Path], scheme=DEFAULT_RSA_SIGNATURE_SCHEME -) -> SSlibKey: - """Load SSlibKey from RSA public key file. +def load_public_key_from_file(path: Union[str, Path]) -> SSlibKey: + """Load SSlibKey from a public key file. * Expected key file format is SubjectPublicKeyInfo/PEM - * Signing scheme defaults to 'rsa-pkcs1v15-sha256' + * Signing scheme is detected from the key material (RSA or ECDSA P-256) * Keyid is computed from legacy canonical representation of public key """ - # TODO handle scheme with open(path, "rb") as f: pem = f.read() @@ -136,12 +156,12 @@ def load_public_key_from_file( def load_signer_from_file( - path: Path, password: Optional[str] = None, scheme=DEFAULT_RSA_SIGNATURE_SCHEME + path: Path, password: Optional[str] = None ) -> CryptoSigner: - """Load CryptoSigner from RSA private key file. + """Load CryptoSigner from a private key file. * Expected key file format is PKCS8/PEM - * Signing scheme defaults to 'rsa-pkcs1v15-sha256' + * Signing scheme is detected from the key material (RSA or ECDSA P-256) * Keyid is computed from legacy canonical representation of public key * If password is None, the key is expected to be unencrypted @@ -149,8 +169,6 @@ def load_signer_from_file( with open(path, "rb") as f: pem = f.read() - # TODO scheme - password_encoded = password.encode() if password is not None else None priv = load_pem_private_key(pem, password_encoded) pub = priv.public_key() @@ -158,10 +176,10 @@ def load_signer_from_file( def load_signer_from_pem(pem: bytes, password: Optional[bytes] = None) -> CryptoSigner: - """Load CryptoSigner from RSA private key file. + """Load CryptoSigner from a private key in PEM format. * Expected key file format is PKCS8/PEM - * Signing scheme defaults to 'rsa-pkcs1v15-sha256' + * Signing scheme is detected from the key material (RSA or ECDSA P-256) * Keyid is computed from legacy canonical representation of public key * If password is None, the key is expected to be unencrypted From de9843ef8d4e9fa332cea6d55fcb8745975721f5 Mon Sep 17 00:00:00 2001 From: Renata Date: Wed, 8 Jul 2026 00:13:31 +0200 Subject: [PATCH 2/6] refactor: remove dead scheme parameter from signing/loading paths Scheme only matters for key generation; loading/signing now uses the key's own (auto-detected) scheme. Removed the dead parameter across taf/keys.py, taf/keystore.py, the api/* signing/loading functions, and their CLI commands. Also fixes bugs found along the way: add_role/add_target_repo dropped the requested scheme, two functions called register_target_files with misaligned positional arguments, and key_cmd_prompt passed scheme into the wrong parameter. --- taf/api/api_workflow.py | 4 --- taf/api/dependencies.py | 7 ----- taf/api/metadata.py | 27 +++++++------------- taf/api/roles.py | 40 +++++++++-------------------- taf/api/targets.py | 41 ++++++++++-------------------- taf/keys.py | 12 ++++----- taf/keystore.py | 35 +++++++++++-------------- taf/tests/test_updater/conftest.py | 4 +-- taf/tools/metadata/__init__.py | 8 ------ taf/tools/roles/__init__.py | 35 ------------------------- taf/tools/targets/__init__.py | 15 ----------- 11 files changed, 56 insertions(+), 172 deletions(-) diff --git a/taf/api/api_workflow.py b/taf/api/api_workflow.py index 71d47c78..1ad13ec1 100644 --- a/taf/api/api_workflow.py +++ b/taf/api/api_workflow.py @@ -10,7 +10,6 @@ ) from taf.auth_repo import AuthenticationRepository from taf.config import load_config -from taf.constants import DEFAULT_RSA_SIGNATURE_SCHEME from taf.exceptions import InvalidConfigError, PushFailedError, TAFError from taf.keys import load_signers from taf.messages import git_commit_message @@ -257,7 +256,6 @@ def manage_repo_and_signers( auth_repo: AuthenticationRepository, roles: Optional[List[str]] = None, keystore: Optional[Union[str, Path]] = None, - scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME, prompt_for_keys: Optional[bool] = False, paths_to_reset_on_error: Optional[List[Union[str, Path]]] = None, load_roles: Optional[bool] = True, @@ -279,7 +277,6 @@ def manage_repo_and_signers( auth_repo (AuthenticationRepository): Already instantiated authentication repository. roles (Optional[List[str]]): List of roles that are expected to be updated. keystore (Optional[Union[str, Path]]): Path to the keystore containing signing keys. - scheme (Optional[str]): The signature scheme. prompt_for_keys (Optional[bool]): If True, prompts for keys if not found. Defaults to False. paths_to_reset_on_error (Optional[List[Union[str, Path]]]): Paths to reset if an error occurs. load_roles (Optional[bool]): If True, loads signing keys of the roles specified using the argument of the same name. @@ -314,7 +311,6 @@ def manage_repo_and_signers( auth_repo, role, keystore=keystore_path, - scheme=scheme, prompt_for_keys=prompt_for_keys, ) auth_repo.add_signers_to_cache({role: keystore_signers}) diff --git a/taf/api/dependencies.py b/taf/api/dependencies.py index 9bc13556..18bb2813 100644 --- a/taf/api/dependencies.py +++ b/taf/api/dependencies.py @@ -12,7 +12,6 @@ from pathlib import Path from taf.auth_repo import AuthenticationRepository -from taf.constants import DEFAULT_RSA_SIGNATURE_SCHEME from taf.exceptions import TAFError from taf.git import GitRepository from taf.log import taf_logger @@ -74,7 +73,6 @@ def add_dependency( dependency_path: Optional[str] = None, dependency_url: Optional[str] = None, library_dir: Optional[str] = None, - scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME, custom: Optional[Dict] = None, prompt_for_keys: Optional[bool] = False, commit: Optional[bool] = True, @@ -97,7 +95,6 @@ def add_dependency( keystore: Location of the keystore files. dependency_path (optional): Path to the dependency repository which is to be added. Can be omitted if dependency_name library_dir (optional): Path to the library's root directory. Determined based on the authentication repository's path if not provided. - scheme (optional): Signing scheme. Set to rsa-pkcs1v15-sha256 by default. custom (optional): Additional data that will be added to dependencies.json if specified. prompt_for_keys (optional): Whether to ask the user to enter their key if it is not located inside the keystore directory. commit (optional): Indicates if the changes should be committed and pushed automatically. @@ -180,7 +177,6 @@ def add_dependency( pin_manager=pin_manager, keystore=keystore, commit=commit, - scheme=scheme, auth_repo=auth_repo, update_snapshot_and_timestamp=True, prompt_for_keys=prompt_for_keys, @@ -206,7 +202,6 @@ def remove_dependency( pin_manager: PinManager, dependency_name: str, keystore: str, - scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME, prompt_for_keys: Optional[bool] = False, commit: Optional[bool] = True, push: Optional[bool] = True, @@ -219,7 +214,6 @@ def remove_dependency( path: Path to the authentication repository. dependency_name: Name of the dependency which should be removed. keystore: Location of the keystore files. - scheme (optional): Signing scheme. Set to rsa-pkcs1v15-sha256 by default. prompt_for_keys (optional): Whether to ask the user to enter their key if it is not located inside the keystore directory. commit (optional): Indicates if the changes should be committed and pushed automatically. push (optional): Flag specifying whether to push to remote @@ -269,7 +263,6 @@ def remove_dependency( pin_manager=pin_manager, keystore=keystore, commit=commit, - scheme=scheme, auth_repo=auth_repo, update_snapshot_and_timestamp=True, prompt_for_keys=prompt_for_keys, diff --git a/taf/api/metadata.py b/taf/api/metadata.py index ff0dec79..22be9088 100644 --- a/taf/api/metadata.py +++ b/taf/api/metadata.py @@ -9,7 +9,6 @@ from taf.api.utils._git import check_if_clean_and_synced from taf.api.api_workflow import manage_repo_and_signers from taf.exceptions import TAFError -from taf.constants import DEFAULT_RSA_SIGNATURE_SCHEME from taf.messages import git_commit_message from taf.tuf.repository import MetadataRepository as TUFRepository from taf.log import Path, taf_logger @@ -28,7 +27,6 @@ def add_key_names( path: str, keys_description: Path, pin_manager: PinManager, - scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME, keystore: Optional[str] = None, commit: Optional[bool] = True, commit_msg: Optional[str] = None, @@ -69,10 +67,9 @@ def add_key_names( with manage_repo_and_signers( auth_repo, - list(parent_roles), - keystore, - scheme, - prompt_for_keys, + roles=list(parent_roles), + keystore=keystore, + prompt_for_keys=prompt_for_keys, load_snapshot_and_timestamp=update_snapshot_and_timestamp, commit=commit, commit_msg=commit_msg, @@ -161,7 +158,6 @@ def update_metadata_expiration_date( roles: List[str], interval: Optional[int] = None, keystore: Optional[str] = None, - scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME, start_date: Optional[datetime] = None, commit: Optional[bool] = True, commit_msg: Optional[str] = None, @@ -181,7 +177,6 @@ def update_metadata_expiration_date( interval: Number of days added to the start date in order to calculate the expiration date. keystore (optional): Keystore directory's path - scheme (optional): Signature scheme. start_date (optional): Date to which expiration interval is added. Set to today if not specified. commit (optional): Indicates if the changes should be committed and pushed automatically. @@ -216,10 +211,9 @@ def update_metadata_expiration_date( with manage_repo_and_signers( auth_repo, - roles, - keystore, - scheme, - prompt_for_keys, + roles=roles, + keystore=keystore, + prompt_for_keys=prompt_for_keys, load_snapshot_and_timestamp=update_snapshot_and_timestamp, commit=commit, commit_msg=commit_msg, @@ -254,7 +248,6 @@ def update_snapshot_and_timestamp( pin_manager: PinManager, keystore: Optional[str] = None, roles_to_sync: Optional[List[str]] = None, - scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME, commit: Optional[bool] = True, commit_msg: Optional[str] = None, prompt_for_keys: Optional[bool] = False, @@ -268,7 +261,6 @@ def update_snapshot_and_timestamp( Arguments: path: Authentication repository's location. keystore (optional): Keystore directory's path - scheme (optional): Signature scheme. commit (optional): Indicates if the changes should be committed and pushed automatically. commit_msg (optional): Custom commit messages. prompt_for_keys (optional): Whether to ask the user to enter their key if it is not located inside the keystore directory. @@ -289,10 +281,9 @@ def update_snapshot_and_timestamp( with manage_repo_and_signers( auth_repo, - [], - keystore, - scheme, - prompt_for_keys, + roles=[], + keystore=keystore, + prompt_for_keys=prompt_for_keys, load_snapshot_and_timestamp=True, commit=commit, commit_msg=commit_msg, diff --git a/taf/api/roles.py b/taf/api/roles.py index b719abdf..8b18fbf9 100644 --- a/taf/api/roles.py +++ b/taf/api/roles.py @@ -123,6 +123,7 @@ def add_role( new_role.number = keys_number new_role.threshold = threshold new_role.yubikey = yubikey + new_role.scheme = scheme signers, verification_keys = load_sorted_keys_of_new_roles( roles=new_role, @@ -136,7 +137,6 @@ def add_role( auth_repo, roles=[parent_role], keystore=keystore_path, - scheme=scheme, prompt_for_keys=prompt_for_keys, load_roles=True, load_snapshot_and_timestamp=True, @@ -244,7 +244,6 @@ def add_roles( pin_manager: PinManager, keystore: Optional[str] = None, roles_key_infos: Optional[str] = None, - scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME, prompt_for_keys: Optional[bool] = False, commit: Optional[bool] = True, push: Optional[bool] = True, @@ -258,7 +257,7 @@ def add_roles( path: Path to the authentication repository. keystore (optional): Location of the keystore files. roles_key_infos: Path to a json file which contains information about repository's roles and keys. - scheme (optional): Signing scheme. Set to rsa-pkcs1v15-sha256 by default. + Each role's signing scheme can be specified per-role in this file (defaults to rsa-pkcs1v15-sha256). prompt_for_keys (optional): Whether to ask the user to enter their key if it is not located inside the keystore directory. commit (optional): Indicates if the changes should be committed and pushed automatically. push (optional): Flag specifying whether to push to remote. @@ -321,7 +320,6 @@ def add_roles( auth_repo, roles=roles_to_load, keystore=keystore_path, - scheme=scheme, prompt_for_keys=prompt_for_keys, load_snapshot_and_timestamp=True, commit_msg=commit_msg, @@ -353,7 +351,6 @@ def add_signing_key( pub_key_path: Optional[str] = None, pub_key: Optional[SSlibKey] = None, keystore: Optional[str] = None, - scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME, commit: Optional[bool] = True, prompt_for_keys: Optional[bool] = False, push: Optional[bool] = True, @@ -370,7 +367,6 @@ def add_signing_key( pub_key_path (optional): path to the file containing the public component of the new key. If not provided, it will be necessary to ender the key when prompted. keystore (optional): Location of the keystore files. - scheme (optional): Signing scheme. Set to rsa-pkcs1v15-sha256 by default. prompt_for_keys (optional): Whether to ask the user to enter their key if it is not located inside the keystore directory. commit (optional): Indicates if the changes should be committed and pushed automatically. push (optional): Flag specifying whether to push to remote. @@ -384,7 +380,7 @@ def add_signing_key( """ pub_key = pub_key or _load_pub_key_from_file( - pub_key_path, prompt_for_keys=prompt_for_keys, scheme=scheme + pub_key_path, prompt_for_keys=prompt_for_keys ) roles_keys = {role: [pub_key] for role in roles} @@ -395,10 +391,9 @@ def add_signing_key( with manage_repo_and_signers( auth_repo, - roles, - keystore, - scheme, - prompt_for_keys, + roles=roles, + keystore=keystore, + prompt_for_keys=prompt_for_keys, load_snapshot_and_timestamp=True, load_parents=True, load_roles=False, @@ -436,7 +431,6 @@ def revoke_signing_key( key_id: str, roles: Optional[List[str]] = None, keystore: Optional[str] = None, - scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME, commit: Optional[bool] = True, prompt_for_keys: Optional[bool] = False, push: Optional[bool] = True, @@ -452,7 +446,6 @@ def revoke_signing_key( roles: A list of roles whose signing keys need to be extended. key_id: id of the key to be removed keystore (optional): Location of the keystore files. - scheme (optional): Signing scheme. Set to rsa-pkcs1v15-sha256 by default. prompt_for_keys (optional): Whether to ask the user to enter their key if it is not located inside the keystore directory. commit (optional): Indicates if the changes should be committed and pushed automatically. push (optional): Flag specifying whether to push to remote. @@ -473,10 +466,9 @@ def revoke_signing_key( with manage_repo_and_signers( auth_repo, - roles_to_update, - keystore, - scheme, - prompt_for_keys, + roles=roles_to_update, + keystore=keystore, + prompt_for_keys=prompt_for_keys, load_snapshot_and_timestamp=True, load_parents=True, load_roles=False, @@ -521,7 +513,6 @@ def rotate_signing_key( pub_key_path: Optional[str] = None, roles: Optional[List[str]] = None, keystore: Optional[str] = None, - scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME, prompt_for_keys: Optional[bool] = False, push: Optional[bool] = True, revoke_commit_msg: Optional[str] = None, @@ -541,7 +532,6 @@ def rotate_signing_key( pub_key_path (optional): path to the file containing the public component of the new key. If not provided, it will be necessary to ender the key when prompted. keystore (optional): Location of the keystore files. - scheme (optional): Signing scheme. Set to rsa-pkcs1v15-sha256 by default. prompt_for_keys (optional): Whether to ask the user to enter their key if it is not located inside the keystore directory. commit (optional): Indicates if the changes should be committed and pushed automatically. push (optional): Flag specifying whether to push to remote. @@ -555,9 +545,7 @@ def rotate_signing_key( None """ - pub_key = _load_pub_key_from_file( - pub_key_path, prompt_for_keys=prompt_for_keys, scheme=scheme - ) + pub_key = _load_pub_key_from_file(pub_key_path, prompt_for_keys=prompt_for_keys) auth_repo = AuthenticationRepository(path=path, pin_manager=pin_manager) keys_name_mappings = read_keys_name_mapping(keys_description) auth_repo.add_key_names(keys_name_mappings) @@ -570,7 +558,6 @@ def rotate_signing_key( key_id=key_id, roles=roles, keystore=keystore, - scheme=scheme, commit=commit, prompt_for_keys=prompt_for_keys, push=False, @@ -583,7 +570,6 @@ def rotate_signing_key( roles=roles, pub_key=pub_key, keystore=keystore, - scheme=scheme, commit=commit, prompt_for_keys=prompt_for_keys, push=push, @@ -706,7 +692,7 @@ def _enter_role_info( return role_info -def _load_pub_key_from_file(pub_key_path, prompt_for_keys, scheme) -> SSlibKey: +def _load_pub_key_from_file(pub_key_path, prompt_for_keys) -> SSlibKey: pub_key_pem = None if pub_key_path is not None: pub_key_pem_path = Path(pub_key_path) @@ -714,7 +700,7 @@ def _load_pub_key_from_file(pub_key_path, prompt_for_keys, scheme) -> SSlibKey: pub_key_pem = Path(pub_key_path).read_text() if pub_key_pem is None and prompt_for_keys: - pub_key_pem = new_public_key_cmd_prompt(scheme)["keyval"]["public"] + pub_key_pem = new_public_key_cmd_prompt()["keyval"]["public"] if pub_key_pem is None: raise TAFError("Public key not provided or invalid") @@ -1100,7 +1086,6 @@ def remove_paths( pin_manager: PinManager, paths: List[str], keystore: str, - scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME, commit: Optional[bool] = True, commit_msg: Optional[str] = None, prompt_for_keys: Optional[bool] = False, @@ -1144,7 +1129,6 @@ def remove_paths( auth_repo, roles=list(paths_to_remove_from_roles.keys()), keystore=keystore, - scheme=scheme, prompt_for_keys=prompt_for_keys, load_roles=False, load_parents=True, diff --git a/taf/api/targets.py b/taf/api/targets.py index 0d24d53e..c9fa8a75 100644 --- a/taf/api/targets.py +++ b/taf/api/targets.py @@ -118,7 +118,7 @@ def add_target_repo( threshold=threshold, yubikey=yubikey, keystore=keystore, - scheme=DEFAULT_RSA_SIGNATURE_SCHEME, + scheme=scheme, commit=True, push=False, auth_repo=auth_repo, @@ -148,7 +148,6 @@ def add_target_repo( pin_manager=pin_manager, keystore=keystore, commit=commit, - scheme=scheme, auth_repo=auth_repo, update_snapshot_and_timestamp=True, prompt_for_keys=prompt_for_keys, @@ -337,7 +336,6 @@ def register_target_files( keystore: Optional[str] = None, roles_key_infos: Optional[str] = None, commit: Optional[bool] = True, - scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME, auth_repo: Optional[AuthenticationRepository] = None, update_snapshot_and_timestamp: Optional[bool] = True, prompt_for_keys: Optional[bool] = False, @@ -355,7 +353,6 @@ def register_target_files( path: Authentication repository's path. keystore: Location of the keystore files. roles_key_infos: A dictionary whose keys are role names, while values contain information about the keys. - scheme (optional): Signing scheme. Set to rsa-pkcs1v15-sha256 by default. auth_repo (optional): If auth repository is already initialized, it can be passed and used. write (optional): Write metadata updates to disk if set to True commit (optional): Indicates if the changes should be committed and pushed automatically. @@ -407,10 +404,9 @@ def register_target_files( roles_to_load.append(role) with manage_repo_and_signers( auth_repo, - roles_to_load, - keystore, - scheme, - prompt_for_keys, + roles=roles_to_load, + keystore=keystore, + prompt_for_keys=prompt_for_keys, load_snapshot_and_timestamp=update_snapshot_and_timestamp, load_parents=False, load_roles=True, @@ -449,7 +445,6 @@ def remove_target_repo( keystore: str, prompt_for_keys: Optional[bool] = False, push: Optional[bool] = True, - scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME, keys_description: Optional[str] = None, ) -> None: """ @@ -491,7 +486,6 @@ def remove_target_repo( pin_manager=pin_manager, keystore=keystore, commit=True, - scheme=scheme, auth_repo=auth_repo, update_snapshot_and_timestamp=True, prompt_for_keys=prompt_for_keys, @@ -579,7 +573,6 @@ def update_target_repos_from_repositories_json( library_dir: str, keystore: str, add_branch: Optional[bool] = True, - scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME, commit: Optional[bool] = True, prompt_for_keys: Optional[bool] = False, push: Optional[bool] = True, @@ -592,7 +585,6 @@ def update_target_repos_from_repositories_json( library_dir: Path to the library's root directory. Determined based on the authentication repository's path if not provided. keystore: Location of the keystore files. add_branch: Indicates whether to add the current branch's name to the target file. - scheme (optional): Signing scheme. Set to rsa-pkcs1v15-sha256 by default. commit (optional): Indicates if the changes should be committed and pushed automatically. prompt_for_keys (optional): Whether to ask the user to enter their key if it is not located inside the keystore directory. push (optional): Flag specifying whether to push to remote @@ -616,12 +608,10 @@ def update_target_repos_from_repositories_json( ) register_target_files( - repo_path, - pin_manager, - keystore, - None, - commit, - scheme, + path=repo_path, + pin_manager=pin_manager, + keystore=keystore, + commit=commit, prompt_for_keys=prompt_for_keys, push=push, update_snapshot_and_timestamp=True, @@ -646,7 +636,6 @@ def update_and_sign_targets( target_types: list, keystore: str, roles_key_infos: str, - scheme: str, commit: Optional[bool] = True, prompt_for_keys: Optional[bool] = False, push: Optional[bool] = True, @@ -660,7 +649,6 @@ def update_and_sign_targets( target_types: Types of target repositories whose corresponding target files should be updated and signed. keystore: Location of the keystore files. roles_key_infos: A dictionary whose keys are role names, while values contain information about the keys. - scheme (optional): Signing scheme. Set to rsa-pkcs1v15-sha256 by default. commit (optional): Indicates if the changes should be committed and pushed automatically. prompt_for_keys (optional): Whether to ask the user to enter their key if it is not located inside the keystore directory. @@ -706,13 +694,12 @@ def update_and_sign_targets( taf_logger.log("NOTICE", f"Updated {target_name} target file") register_target_files( - repo_path, - pin_manager, - keystore, - roles_key_infos, - commit, - push, - scheme, + path=repo_path, + pin_manager=pin_manager, + keystore=keystore, + roles_key_infos=roles_key_infos, + commit=commit, + push=push, prompt_for_keys=prompt_for_keys, reset_updated_targets_on_error=True, update_snapshot_and_timestamp=True, diff --git a/taf/keys.py b/taf/keys.py index e20f1117..424b8270 100644 --- a/taf/keys.py +++ b/taf/keys.py @@ -185,17 +185,17 @@ def _sort_roles(roles): def _load_signer_from_keystore( - taf_repo, keystore_path, key_name, num_of_signatures, scheme, role + taf_repo, keystore_path, key_name, num_of_signatures, role ) -> CryptoSigner: if keystore_path is None: return None if (keystore_path / key_name).is_file(): try: signer = load_signer_from_private_keystore( - keystore=keystore_path, key_name=key_name, scheme=scheme + keystore=keystore_path, key_name=key_name ) # load only valid keys - if taf_repo.is_valid_metadata_key(role, signer.public_key, scheme=scheme): + if taf_repo.is_valid_metadata_key(role, signer.public_key): return signer except KeystoreError: pass @@ -270,7 +270,6 @@ def load_signers( taf_repo: TUFRepository, role: str, keystore: Optional[str] = None, - scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME, prompt_for_keys: Optional[bool] = False, key_id_pins: Optional[Dict] = None, use_yubikeys_to_sign: bool = False, @@ -318,7 +317,7 @@ def load_signers( if num_of_signatures < len(keystore_files): key_name = keystore_files[num_of_signatures] signer = _load_signer_from_keystore( - taf_repo, keystore_path, key_name, num_of_signatures, scheme, role + taf_repo, keystore_path, key_name, num_of_signatures, role ) if signer is not None: signers_keystore.append(signer) @@ -370,7 +369,7 @@ def load_signers( if prompt_for_keys and click.confirm(f"Manually enter {role} key?"): keys = [signer.public_key for signer in signers_keystore] - key = key_cmd_prompt(key_name, role, taf_repo, keys, scheme) + key = key_cmd_prompt(key_name, role, taf_repo, keys) signer = load_signer_from_pem(key) signers_keystore.append(key) num_of_signatures += 1 @@ -570,7 +569,6 @@ def _invalid_key_message(key_name, keystore): signer = load_signer_from_private_keystore( keystore_path, key_name, - scheme=scheme, password=password, ) except KeystoreError: diff --git a/taf/keystore.py b/taf/keystore.py index 55b9e69b..fbac811a 100644 --- a/taf/keystore.py +++ b/taf/keystore.py @@ -12,7 +12,6 @@ load_signer_from_pem, ) -from taf.constants import DEFAULT_RSA_SIGNATURE_SCHEME from taf.exceptions import KeystoreError from taf.tuf.repository import MetadataRepository as TUFRepository @@ -54,13 +53,12 @@ def key_cmd_prompt( role: str, taf_repo: TUFRepository, loaded_keys: Optional[List] = None, - scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME, ) -> CryptoSigner: - def _enter_and_check_key(key_name, role, loaded_keys, scheme): + def _enter_and_check_key(key_name, role, loaded_keys): pem = getpass(f"Enter {key_name} private key without its header and footer\n") pem = _form_private_pem(pem) try: - signer = load_signer_from_pem(pem, scheme) + signer = load_signer_from_pem(pem) except Exception: print("Invalid key") return None @@ -74,23 +72,23 @@ def _enter_and_check_key(key_name, role, loaded_keys, scheme): return signer while True: - pem = _enter_and_check_key(key_name, role, loaded_keys, scheme) + pem = _enter_and_check_key(key_name, role, loaded_keys) if pem is not None: return pem -def new_public_key_cmd_prompt(scheme: Optional[str]) -> SSlibKey: - def _enter_and_check_key(scheme): +def new_public_key_cmd_prompt() -> SSlibKey: + def _enter_and_check_key(): pem = getpass("Enter public key without its header and footer\n") pem = _from_public_pem(pem) try: - return load_public_key_from_file(pem, scheme) + return load_public_key_from_file(pem) except Exception: print("Invalid key") return None while True: - key = _enter_and_check_key(scheme) + key = _enter_and_check_key() if key is not None: return key @@ -98,17 +96,16 @@ def _enter_and_check_key(scheme): def load_signer_from_private_keystore( keystore: str, key_name: str, - scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME, password: Optional[str] = None, ) -> CryptoSigner: key_path = Path(keystore, key_name).expanduser().resolve() if not key_path.is_file(): raise KeystoreError(f"{str(key_path)} does not exist") - def _read_key(path, password, scheme): - def _read_key_or_keystore_error(path, password, scheme): + def _read_key(path, password): + def _read_key_or_keystore_error(path, password): try: - return load_signer_from_file(path, password or None, scheme=scheme) + return load_signer_from_file(path, password or None) except TypeError: raise except Exception as e: @@ -116,34 +113,32 @@ def _read_key_or_keystore_error(path, password, scheme): try: # try to load with a given password or None - return _read_key_or_keystore_error(path, password, scheme) + return _read_key_or_keystore_error(path, password) except TypeError: password = getpass( f"Enter {key_name} keystore file password and press ENTER" ) try: - return _read_key_or_keystore_error(path, password, scheme) + return _read_key_or_keystore_error(path, password) except Exception: return None except Exception: return None while True: - signer = _read_key(key_path, password, scheme) + signer = _read_key(key_path, password) if signer is not None: return signer if not click.confirm(f"Could not open keystore file {key_path}. Try again?"): raise KeystoreError(f"Could not open keystore file {key_path}") -def read_public_key_from_keystore( - keystore: str, key_name: str, scheme: Optional[str] = DEFAULT_RSA_SIGNATURE_SCHEME -) -> SSlibKey: +def read_public_key_from_keystore(keystore: str, key_name: str) -> SSlibKey: pub_key_path = Path(keystore, f"{key_name}.pub").expanduser().resolve() if not pub_key_path.is_file(): raise KeystoreError(f"{str(pub_key_path)} does not exist") try: - return load_public_key_from_file(str(pub_key_path), scheme) + return load_public_key_from_file(str(pub_key_path)) except ( securesystemslib.exceptions.FormatError, securesystemslib.exceptions.Error, diff --git a/taf/tests/test_updater/conftest.py b/taf/tests/test_updater/conftest.py index 1db2d2cd..677bc9e3 100644 --- a/taf/tests/test_updater/conftest.py +++ b/taf/tests/test_updater/conftest.py @@ -20,7 +20,7 @@ update_metadata_expiration_date, ) from taf.auth_repo import AuthenticationRepository -from taf.constants import DEFAULT_RSA_SIGNATURE_SCHEME, TARGETS_DIRECTORY_NAME +from taf.constants import TARGETS_DIRECTORY_NAME from taf import repositoriesdb, settings from taf.exceptions import GitError from taf.utils import on_rm_error @@ -826,7 +826,6 @@ def update_timestamp_metadata_invalid_signature( auth_repo, [role], keystore=KEYSTORE_PATH, - scheme=DEFAULT_RSA_SIGNATURE_SCHEME, prompt_for_keys=False, load_snapshot_and_timestamp=False, commit=True, @@ -850,7 +849,6 @@ def update_and_sign_metadata_without_clean_check( pin_manager=pin_manager, roles=roles, keystore=KEYSTORE_PATH, - scheme=DEFAULT_RSA_SIGNATURE_SCHEME, prompt_for_keys=False, skip_clean_check=True, update_snapshot_and_timestamp=True, diff --git a/taf/tools/metadata/__init__.py b/taf/tools/metadata/__init__.py index ea77f4b0..e63acc17 100644 --- a/taf/tools/metadata/__init__.py +++ b/taf/tools/metadata/__init__.py @@ -5,7 +5,6 @@ check_expiration_dates, add_key_names, ) -from taf.constants import DEFAULT_RSA_SIGNATURE_SCHEME from taf.exceptions import TAFError from taf.tools.cli import catch_cli_exception, common_repo_edit_options, find_repository from taf.tools.repo import pin_managed @@ -134,11 +133,6 @@ def update_expiration_dates_command(): type=int, help="Number of days added to the start date", ) - @click.option( - "--scheme", - default=DEFAULT_RSA_SIGNATURE_SCHEME, - help="A signature scheme used for signing", - ) @click.option( "--start-date", default=datetime.datetime.now(), @@ -152,7 +146,6 @@ def update_expiration_dates( role, interval, keystore, - scheme, start_date, no_commit, prompt_for_keys, @@ -170,7 +163,6 @@ def update_expiration_dates( roles=role, interval=interval, keystore=keystore, - scheme=scheme, start_date=start_date, commit=not no_commit, prompt_for_keys=prompt_for_keys, diff --git a/taf/tools/roles/__init__.py b/taf/tools/roles/__init__.py index 375a7af6..22f3a725 100644 --- a/taf/tools/roles/__init__.py +++ b/taf/tools/roles/__init__.py @@ -81,16 +81,10 @@ def add_roles_command(): default=".", help="Authentication repository's location. If not specified, set to the current directory", ) - @click.option( - "--scheme", - default=DEFAULT_RSA_SIGNATURE_SCHEME, - help="A signature scheme used for signing", - ) @pin_managed def add_roles( config_file, path, - scheme, keystore, no_commit, prompt_for_keys, @@ -103,7 +97,6 @@ def add_roles( pin_manager=pin_manager, keystore=keystore, roles_key_infos=config_file, - scheme=scheme, prompt_for_keys=prompt_for_keys, commit=not no_commit, keys_description=keys_description, @@ -264,18 +257,12 @@ def remove_paths_command(): @click.option( "--delegated-path", multiple=True, help="A list of paths to be removed" ) - @click.option( - "--scheme", - default=DEFAULT_RSA_SIGNATURE_SCHEME, - help="A signature scheme used for signing", - ) @click.option("--commit-msg", default=None, help="Commit message") @pin_managed def remove_delegated_paths( path, delegated_path, keystore, - scheme, no_commit, prompt_for_keys, pin_manager, @@ -292,7 +279,6 @@ def remove_delegated_paths( pin_manager=pin_manager, paths=delegated_path, keystore=keystore, - scheme=scheme, commit=not no_commit, prompt_for_keys=prompt_for_keys, keys_description=keys_description, @@ -331,11 +317,6 @@ def add_signing_key_command(): default=None, help="Path to the public key corresponding to the private key which should be registered as the role's signing key", ) - @click.option( - "--scheme", - default=DEFAULT_RSA_SIGNATURE_SCHEME, - help="A signature scheme used for signing", - ) @click.option("--commit-msg", default=None, help="Commit message") @pin_managed def adding_signing_key( @@ -343,7 +324,6 @@ def adding_signing_key( role, pub_key_path, keystore, - scheme, no_commit, prompt_for_keys, pin_manager, @@ -361,7 +341,6 @@ def adding_signing_key( roles=role, pub_key_path=pub_key_path, keystore=keystore, - scheme=scheme, commit=not no_commit, prompt_for_keys=prompt_for_keys, keys_description=keys_description, @@ -390,11 +369,6 @@ def revoke_signing_key_command(): multiple=True, help="A list of roles from which to remove the key. If unspecified, the key is removed from all roles by default.", ) - @click.option( - "--scheme", - default=DEFAULT_RSA_SIGNATURE_SCHEME, - help="A signature scheme used for signing", - ) @click.option("--commit-msg", default=None, help="Commit message") @pin_managed def revoke_key( @@ -402,7 +376,6 @@ def revoke_key( role, keyid, keystore, - scheme, no_commit, prompt_for_keys, pin_manager, @@ -417,7 +390,6 @@ def revoke_key( roles=role, key_id=keyid, keystore=keystore, - scheme=scheme, commit=not no_commit, prompt_for_keys=prompt_for_keys, keys_description=keys_description, @@ -450,11 +422,6 @@ def rotate_signing_key_command(): default=None, help="Path to the public key corresponding to the private key which should be registered as the role's signing key", ) - @click.option( - "--scheme", - default=DEFAULT_RSA_SIGNATURE_SCHEME, - help="A signature scheme used for signing", - ) @click.option("--revoke-commit-msg", default=None, help="Revoke key commit message") @click.option( "--add-commit-msg", default=None, help="Add new signing key commit message" @@ -467,7 +434,6 @@ def rotate_key( keyid, pub_key_path, keystore, - scheme, prompt_for_keys, revoke_commit_msg, add_commit_msg, @@ -482,7 +448,6 @@ def rotate_key( roles=role, key_id=keyid, keystore=keystore, - scheme=scheme, prompt_for_keys=prompt_for_keys, pub_key_path=pub_key_path, revoke_commit_msg=revoke_commit_msg, diff --git a/taf/tools/targets/__init__.py b/taf/tools/targets/__init__.py index f453beec..9eb84751 100644 --- a/taf/tools/targets/__init__.py +++ b/taf/tools/targets/__init__.py @@ -293,11 +293,6 @@ def sign_targets_command(): default=".", help="Authentication repository's location. If not specified, set to the current directory", ) - @click.option( - "--scheme", - default=DEFAULT_RSA_SIGNATURE_SCHEME, - help="A signature scheme used for signing", - ) @click.option( "--no-commit", is_flag=True, @@ -309,7 +304,6 @@ def sign( path, keystore, keys_description, - scheme, prompt_for_keys, no_commit, pin_manager, @@ -320,7 +314,6 @@ def sign( pin_manager=pin_manager, keystore=keystore, roles_key_infos=keys_description, - scheme=scheme, update_snapshot_and_timestamp=True, prompt_for_keys=prompt_for_keys, commit=not no_commit, @@ -377,11 +370,6 @@ def update_and_sign_command(): "--keys-description", help="A dictionary containing information about the keys or a path to a json file which stores the needed information", ) - @click.option( - "--scheme", - default=DEFAULT_RSA_SIGNATURE_SCHEME, - help="A signature scheme used for signing", - ) @click.option( "--prompt-for-keys", is_flag=True, @@ -401,7 +389,6 @@ def update_and_sign( target_type, keystore, keys_description, - scheme, prompt_for_keys, no_commit, pin_manager, @@ -415,7 +402,6 @@ def update_and_sign( target_type, keystore=keystore, roles_key_infos=keys_description, - scheme=scheme, prompt_for_keys=prompt_for_keys, commit=not no_commit, ) @@ -425,7 +411,6 @@ def update_and_sign( library_dir, add_branch=True, keystore=keystore, - scheme=scheme, prompt_for_keys=prompt_for_keys, commit=not no_commit, keys_description=keys_description, From 0e338a4975fdeba21ece6e996491c9c94f05333f Mon Sep 17 00:00:00 2001 From: Renata Date: Wed, 8 Jul 2026 07:24:05 +0200 Subject: [PATCH 3/6] feat: apply requested scheme to generated keys, add coverage for it add_role's scheme argument reached the Role model but was never forwarded into actual key generation, so new keys were always tagged rsa-pkcs1v15-sha256. Threads scheme through load_signer_from_pem so it reaches the real key. Adds skip_prompt to add_target_repo (mirroring add_role) and tests for add_role/add_target_repo scheme handling and update_and_sign_targets (previously untested). --- taf/api/targets.py | 4 + taf/keys.py | 4 +- taf/tests/test_api/roles/api/test_roles.py | 32 +++++++- .../test_api/targets/api/test_targets.py | 75 +++++++++++++++++++ taf/tests/test_api/util.py | 11 +++ taf/tuf/keys.py | 35 ++++++--- 6 files changed, 146 insertions(+), 15 deletions(-) diff --git a/taf/api/targets.py b/taf/api/targets.py index c9fa8a75..d6fc9baf 100644 --- a/taf/api/targets.py +++ b/taf/api/targets.py @@ -56,6 +56,7 @@ def add_target_repo( prompt_for_keys: Optional[bool] = False, push: Optional[bool] = True, keys_description: Optional[str] = None, + skip_prompt: Optional[bool] = False, ) -> None: """ Add a new target repository by adding it to repositories.json, creating a delegation (if targets is not @@ -75,6 +76,8 @@ def add_target_repo( commit (optional): Indicates if the changes should be committed and pushed automatically. prompt_for_keys (optional): Whether to ask the user to enter their key if it is not located inside the keystore directory. push (optional): Flag specifying whether to push to remote + skip_prompt (optional): A flag defining if the user will be asked if they want to generate new keys or reuse existing + ones in case keystore files should be used, when a new role has to be created. New keys will be generated by default. Side Effects: Updates metadata and repositories.json, adds a new target file if repository exists and writes changes to disk @@ -123,6 +126,7 @@ def add_target_repo( push=False, auth_repo=auth_repo, prompt_for_keys=prompt_for_keys, + skip_prompt=skip_prompt, ) elif role != "targets": # delegated role paths are not specified for the top-level targets role diff --git a/taf/keys.py b/taf/keys.py index 424b8270..5f4e177d 100644 --- a/taf/keys.py +++ b/taf/keys.py @@ -604,11 +604,11 @@ def _invalid_key_message(key_name, keystore): private_pem = generate_and_write_rsa_keypair( path=Path(keystore, key_name), key_size=length, password=password ) - signer = load_signer_from_pem(private_pem) + signer = load_signer_from_pem(private_pem, scheme=scheme) else: _, private_pem = generate_rsa_keypair(key_size=length) print(f"{role_name} key:\n\n{private_pem.decode()}\n\n") - signer = load_signer_from_pem(private_pem) + signer = load_signer_from_pem(private_pem, scheme=scheme) if signer is not None: return signer, _get_legacy_keyid(signer.public_key) diff --git a/taf/tests/test_api/roles/api/test_roles.py b/taf/tests/test_api/roles/api/test_roles.py index 6adb539e..a81f8434 100644 --- a/taf/tests/test_api/roles/api/test_roles.py +++ b/taf/tests/test_api/roles/api/test_roles.py @@ -10,7 +10,7 @@ ) from taf.messages import git_commit_message from taf.auth_repo import AuthenticationRepository -from taf.tests.test_api.util import check_new_role +from taf.tests.test_api.util import check_new_role, check_role_scheme from taf.yubikey.yubikey_manager import PinManager @@ -74,6 +74,36 @@ def test_add_role_when_delegated_role_is_parent( ) +def test_add_role_applies_requested_scheme_to_generated_key( + auth_repo: AuthenticationRepository, + roles_keystore: str, + pin_manager: PinManager, +): + """add_role must apply the scheme it was called with to the newly + generated key, instead of silently defaulting to rsa-pkcs1v15-sha256 + regardless of what was requested. + """ + requested_scheme = "rsassa-pss-sha256" + ROLE_NAME = "role_with_custom_scheme" + add_role( + path=str(auth_repo.path), + pin_manager=pin_manager, + auth_repo=auth_repo, + role=ROLE_NAME, + parent_role="targets", + paths=["some-scheme-path"], + keys_number=1, + threshold=1, + yubikey=False, + keystore=roles_keystore, + scheme=requested_scheme, + push=False, + skip_prompt=True, + ) + + check_role_scheme(auth_repo, ROLE_NAME, requested_scheme) + + def test_add_multiple_roles( auth_repo: AuthenticationRepository, pin_manager: PinManager, diff --git a/taf/tests/test_api/targets/api/test_targets.py b/taf/tests/test_api/targets/api/test_targets.py index b71f1c6d..bb42a79f 100644 --- a/taf/tests/test_api/targets/api/test_targets.py +++ b/taf/tests/test_api/targets/api/test_targets.py @@ -8,10 +8,12 @@ from taf.api.targets import ( add_target_repo, register_target_files, + update_and_sign_targets, update_target_repos_from_repositories_json, ) from taf.tests.test_api.util import ( check_if_targets_signed, + check_role_scheme, check_target_file, ) from taf.yubikey.yubikey_manager import PinManager @@ -194,6 +196,38 @@ def test_update_target_repos_from_repositories_json( assert commits[0].message.strip() == git_commit_message("update-targets") +def test_update_and_sign_targets( + auth_repo_when_add_repositories_json: AuthenticationRepository, + pin_manager: PinManager, + library: Path, + keystore_delegations: str, +): + """update_and_sign_targets had a bug where it called register_target_files + with positional arguments that no longer lined up after the scheme + parameter was removed, silently corrupting later arguments. This is the + first direct test of this function. + """ + repo_path = library / "auth" + initial_commits_num = len(auth_repo_when_add_repositories_json.list_pygit_commits()) + namespace = library.name + update_and_sign_targets( + str(repo_path), + pin_manager, + str(library.parent), + ["type1"], + keystore_delegations, + None, + push=False, + ) + target_repo_name = f"{namespace}/target1" + target_repo_path = library.parent / target_repo_name + assert check_target_file( + target_repo_path, target_repo_name, auth_repo_when_add_repositories_json + ) + commits = auth_repo_when_add_repositories_json.list_pygit_commits() + assert len(commits) == initial_commits_num + 1 + + def test_add_target_repository_when_not_on_filesystem( auth_repo_when_add_repositories_json: AuthenticationRepository, pin_manager: PinManager, @@ -234,6 +268,47 @@ def test_add_target_repository_when_not_on_filesystem( assert target_repo_name in delegated_paths +def test_add_target_repo_applies_requested_scheme_when_creating_new_role( + auth_repo_when_add_repositories_json: AuthenticationRepository, + pin_manager: PinManager, + library: Path, + roles_keystore: str, +): + """When add_target_repo has to create a brand-new role for the target, + it must forward the scheme it was called with to add_role, instead of + hardcoding the RSA default regardless of what was requested. + + Uses roles_keystore (an ephemeral, per-module copy that gets deleted + after the test module runs) rather than keystore_delegations directly, + since this test generates a brand-new key and keystore_delegations is + a shared, persistent fixture data directory. + """ + requested_scheme = "rsassa-pss-sha256" + ROLE_NAME = "role_for_custom_scheme" + repo_path = str(library / "auth") + namespace = library.name + target_repo_name = f"{namespace}/target_with_new_role" + add_target_repo( + repo_path, + pin_manager, + None, + target_repo_name, + ROLE_NAME, + None, + roles_keystore, + paths=[target_repo_name], + keys_number=1, + threshold=1, + yubikey=False, + scheme=requested_scheme, + push=False, + should_create_new_role=True, + skip_prompt=True, + ) + + check_role_scheme(auth_repo_when_add_repositories_json, ROLE_NAME, requested_scheme) + + def test_add_target_repository_when_on_filesystem( auth_repo_when_add_repositories_json: AuthenticationRepository, pin_manager: PinManager, diff --git a/taf/tests/test_api/util.py b/taf/tests/test_api/util.py index 1cc62572..7cea85b2 100644 --- a/taf/tests/test_api/util.py +++ b/taf/tests/test_api/util.py @@ -75,3 +75,14 @@ def check_new_role( assert auth_repo.find_delegated_roles_parent(role_name) == parent_name roles_paths = auth_repo.get_role_paths(role_name) assert roles_paths == paths + + +def check_role_scheme( + auth_repo: AuthenticationRepository, + role_name: str, + expected_scheme: str, + parent_name: str = "targets", +): + keyid = auth_repo.get_keyids_of_role(role_name)[0] + _, _, scheme = auth_repo.get_key_length_and_scheme_from_metadata(parent_name, keyid) + assert scheme == expected_scheme diff --git a/taf/tuf/keys.py b/taf/tuf/keys.py index a998d741..2e761925 100644 --- a/taf/tuf/keys.py +++ b/taf/tuf/keys.py @@ -107,25 +107,33 @@ def _get_legacy_keyid(key: SSlibKey) -> str: def _from_crypto( pub: Union[RSAPublicKey, EllipticCurvePublicKey], - scheme=DEFAULT_RSA_SIGNATURE_SCHEME, + scheme: Optional[str] = None, ) -> SSlibKey: """Converts pyca/cryptography public key to SSlibKey with default signing scheme and legacy keyid. Detects the key's actual type (RSA or ECDSA P-256) from the key material - itself. This lets callers that only pass the historical RSA default - scheme (because they don't know ahead of time what type of key they're - about to load, e.g. reading a YubiKey's public key) still get back a - correctly scheme-tagged key when it turns out to be an EC key. + itself. This lets callers that pass no scheme, or the historical RSA + default, because they don't know ahead of time what type of key they're + about to load (e.g. reading a YubiKey's public key) still get back a + correctly scheme-tagged key when it turns out to be an EC key. RSA + callers that need a specific sub-scheme (e.g. rsassa-pss-sha256) can + still request one explicitly, since that can't be inferred from the key + bytes alone. """ # securesystemslib does not (yet) check if keytype and scheme are compatible # https://github.com/secure-systems-lab/securesystemslib/issues/766 if isinstance(pub, RSAPublicKey): - pass + if scheme is None: + scheme = DEFAULT_RSA_SIGNATURE_SCHEME elif isinstance(pub, EllipticCurvePublicKey): if not isinstance(pub.curve, SECP256R1): raise ValueError(f"unsupported EC curve '{pub.curve.name}'") - if scheme in (DEFAULT_RSA_SIGNATURE_SCHEME, DEFAULT_ECDSA_SIGNATURE_SCHEME): + if scheme in ( + None, + DEFAULT_RSA_SIGNATURE_SCHEME, + DEFAULT_ECDSA_SIGNATURE_SCHEME, + ): scheme = DEFAULT_ECDSA_SIGNATURE_SCHEME else: raise ValueError(f"scheme '{scheme}' not valid for an ecdsa key") @@ -155,9 +163,7 @@ def load_public_key_from_file(path: Union[str, Path]) -> SSlibKey: return _from_crypto(pub) -def load_signer_from_file( - path: Path, password: Optional[str] = None -) -> CryptoSigner: +def load_signer_from_file(path: Path, password: Optional[str] = None) -> CryptoSigner: """Load CryptoSigner from a private key file. * Expected key file format is PKCS8/PEM @@ -175,18 +181,23 @@ def load_signer_from_file( return CryptoSigner(priv, _from_crypto(pub)) -def load_signer_from_pem(pem: bytes, password: Optional[bytes] = None) -> CryptoSigner: +def load_signer_from_pem( + pem: bytes, password: Optional[bytes] = None, scheme: Optional[str] = None +) -> CryptoSigner: """Load CryptoSigner from a private key in PEM format. * Expected key file format is PKCS8/PEM * Signing scheme is detected from the key material (RSA or ECDSA P-256) + unless explicitly overridden, which only makes sense for RSA, where + the sub-scheme (e.g. rsassa-pss-sha256) can't be inferred from the + key bytes alone * Keyid is computed from legacy canonical representation of public key * If password is None, the key is expected to be unencrypted """ priv = load_pem_private_key(pem, password) pub = priv.public_key() - return CryptoSigner(priv, _from_crypto(pub)) + return CryptoSigner(priv, _from_crypto(pub, scheme)) class YkSigner(Signer): From 01f5fbfe7a81c421a58f320acfe55763a1059594 Mon Sep 17 00:00:00 2001 From: Renata Date: Wed, 8 Jul 2026 07:49:55 +0200 Subject: [PATCH 4/6] refactor: use None as the single scheme-unset sentinel get_sslib_key_from_value and is_valid_metadata_key defaulted to the RSA scheme string, forcing _from_crypto into an awkward tuple check to also treat it as "no opinion" for EC keys. Default to None everywhere instead, so there's exactly one sentinel. --- taf/tuf/keys.py | 29 ++++++++++------------------- taf/tuf/repository.py | 3 +-- 2 files changed, 11 insertions(+), 21 deletions(-) diff --git a/taf/tuf/keys.py b/taf/tuf/keys.py index 2e761925..c1a73795 100644 --- a/taf/tuf/keys.py +++ b/taf/tuf/keys.py @@ -78,9 +78,7 @@ def generate_and_write_rsa_keypair(path, key_size, password) -> bytes: return private_pem -def get_sslib_key_from_value( - key: str, scheme: str = DEFAULT_RSA_SIGNATURE_SCHEME -) -> SSlibKey: +def get_sslib_key_from_value(key: str, scheme: Optional[str] = None) -> SSlibKey: """ Converts a key from its string representation into an SSlibKey object. """ @@ -113,30 +111,23 @@ def _from_crypto( scheme and legacy keyid. Detects the key's actual type (RSA or ECDSA P-256) from the key material - itself. This lets callers that pass no scheme, or the historical RSA - default, because they don't know ahead of time what type of key they're - about to load (e.g. reading a YubiKey's public key) still get back a - correctly scheme-tagged key when it turns out to be an EC key. RSA - callers that need a specific sub-scheme (e.g. rsassa-pss-sha256) can - still request one explicitly, since that can't be inferred from the key - bytes alone. + itself, so callers that pass scheme=None because they don't know ahead + of time what type of key they're about to load (e.g. reading a + YubiKey's public key) still get back a correctly scheme-tagged key + when it turns out to be an EC key. RSA callers that need a specific + sub-scheme (e.g. rsassa-pss-sha256) can still request one explicitly, + since that can't be inferred from the key bytes alone. """ # securesystemslib does not (yet) check if keytype and scheme are compatible # https://github.com/secure-systems-lab/securesystemslib/issues/766 if isinstance(pub, RSAPublicKey): - if scheme is None: - scheme = DEFAULT_RSA_SIGNATURE_SCHEME + scheme = scheme or DEFAULT_RSA_SIGNATURE_SCHEME elif isinstance(pub, EllipticCurvePublicKey): if not isinstance(pub.curve, SECP256R1): raise ValueError(f"unsupported EC curve '{pub.curve.name}'") - if scheme in ( - None, - DEFAULT_RSA_SIGNATURE_SCHEME, - DEFAULT_ECDSA_SIGNATURE_SCHEME, - ): - scheme = DEFAULT_ECDSA_SIGNATURE_SCHEME - else: + if scheme not in (None, DEFAULT_ECDSA_SIGNATURE_SCHEME): raise ValueError(f"scheme '{scheme}' not valid for an ecdsa key") + scheme = DEFAULT_ECDSA_SIGNATURE_SCHEME else: raise ValueError(f"keytype '{type(pub)}' not supported") key = SSlibKey.from_crypto(pub, scheme=scheme) diff --git a/taf/tuf/repository.py b/taf/tuf/repository.py index 3bfda17c..d3f5a5bf 100644 --- a/taf/tuf/repository.py +++ b/taf/tuf/repository.py @@ -29,7 +29,6 @@ except ImportError: yk = YubikeyMissingLibrary() # type: ignore -from taf.constants import DEFAULT_RSA_SIGNATURE_SCHEME from taf.utils import ( default_backend, get_file_details, @@ -1376,7 +1375,7 @@ def get_role_keys(self, role): pass def is_valid_metadata_key( - self, role: str, key: Union[SSlibKey, str], scheme=DEFAULT_RSA_SIGNATURE_SCHEME + self, role: str, key: Union[SSlibKey, str], scheme: Optional[str] = None ) -> bool: """Checks if metadata role contains key id of provided key. From 3328dd695b956e497bd70c682123364df8843f57 Mon Sep 17 00:00:00 2001 From: Renata Date: Thu, 9 Jul 2026 00:13:53 +0200 Subject: [PATCH 5/6] chore: tidy up docstrings and imports --- taf/api/roles.py | 1 - taf/tests/test_api/roles/api/test_roles.py | 4 ---- taf/tests/test_api/targets/api/test_targets.py | 16 +--------------- taf/tests/tuf/test_keys/test_ecdsa_keys.py | 10 +++++----- taf/tuf/keys.py | 5 +++++ 5 files changed, 11 insertions(+), 25 deletions(-) diff --git a/taf/api/roles.py b/taf/api/roles.py index 8b18fbf9..fcac7989 100644 --- a/taf/api/roles.py +++ b/taf/api/roles.py @@ -257,7 +257,6 @@ def add_roles( path: Path to the authentication repository. keystore (optional): Location of the keystore files. roles_key_infos: Path to a json file which contains information about repository's roles and keys. - Each role's signing scheme can be specified per-role in this file (defaults to rsa-pkcs1v15-sha256). prompt_for_keys (optional): Whether to ask the user to enter their key if it is not located inside the keystore directory. commit (optional): Indicates if the changes should be committed and pushed automatically. push (optional): Flag specifying whether to push to remote. diff --git a/taf/tests/test_api/roles/api/test_roles.py b/taf/tests/test_api/roles/api/test_roles.py index a81f8434..74903587 100644 --- a/taf/tests/test_api/roles/api/test_roles.py +++ b/taf/tests/test_api/roles/api/test_roles.py @@ -79,10 +79,6 @@ def test_add_role_applies_requested_scheme_to_generated_key( roles_keystore: str, pin_manager: PinManager, ): - """add_role must apply the scheme it was called with to the newly - generated key, instead of silently defaulting to rsa-pkcs1v15-sha256 - regardless of what was requested. - """ requested_scheme = "rsassa-pss-sha256" ROLE_NAME = "role_with_custom_scheme" add_role( diff --git a/taf/tests/test_api/targets/api/test_targets.py b/taf/tests/test_api/targets/api/test_targets.py index bb42a79f..02cde149 100644 --- a/taf/tests/test_api/targets/api/test_targets.py +++ b/taf/tests/test_api/targets/api/test_targets.py @@ -196,17 +196,12 @@ def test_update_target_repos_from_repositories_json( assert commits[0].message.strip() == git_commit_message("update-targets") -def test_update_and_sign_targets( +def test_update_and_sign_targets_when_target_type_matches( auth_repo_when_add_repositories_json: AuthenticationRepository, pin_manager: PinManager, library: Path, keystore_delegations: str, ): - """update_and_sign_targets had a bug where it called register_target_files - with positional arguments that no longer lined up after the scheme - parameter was removed, silently corrupting later arguments. This is the - first direct test of this function. - """ repo_path = library / "auth" initial_commits_num = len(auth_repo_when_add_repositories_json.list_pygit_commits()) namespace = library.name @@ -274,15 +269,6 @@ def test_add_target_repo_applies_requested_scheme_when_creating_new_role( library: Path, roles_keystore: str, ): - """When add_target_repo has to create a brand-new role for the target, - it must forward the scheme it was called with to add_role, instead of - hardcoding the RSA default regardless of what was requested. - - Uses roles_keystore (an ephemeral, per-module copy that gets deleted - after the test module runs) rather than keystore_delegations directly, - since this test generates a brand-new key and keystore_delegations is - a shared, persistent fixture data directory. - """ requested_scheme = "rsassa-pss-sha256" ROLE_NAME = "role_for_custom_scheme" repo_path = str(library / "auth") diff --git a/taf/tests/tuf/test_keys/test_ecdsa_keys.py b/taf/tests/tuf/test_keys/test_ecdsa_keys.py index 86a9772b..27580a15 100644 --- a/taf/tests/tuf/test_keys/test_ecdsa_keys.py +++ b/taf/tests/tuf/test_keys/test_ecdsa_keys.py @@ -6,7 +6,11 @@ ) from taf.constants import DEFAULT_ECDSA_SIGNATURE_SCHEME, DEFAULT_RSA_SIGNATURE_SCHEME -from taf.tuf.keys import get_sslib_key_from_value, load_signer_from_pem +from taf.tuf.keys import ( + generate_rsa_keypair, + get_sslib_key_from_value, + load_signer_from_pem, +) def _ec_keypair_pem(curve=SECP256R1()): @@ -80,10 +84,6 @@ def test_ecdsa_signer_round_trip(): def test_rsa_scheme_default_still_used_for_rsa_keys(): - """Regression guard: the RSA path must be entirely unaffected by the - EC-detection branch.""" - from taf.tuf.keys import generate_rsa_keypair - _, public_pem = generate_rsa_keypair(key_size=2048) key = get_sslib_key_from_value(public_pem.decode()) diff --git a/taf/tuf/keys.py b/taf/tuf/keys.py index c1a73795..e597de35 100644 --- a/taf/tuf/keys.py +++ b/taf/tuf/keys.py @@ -123,6 +123,11 @@ def _from_crypto( if isinstance(pub, RSAPublicKey): scheme = scheme or DEFAULT_RSA_SIGNATURE_SCHEME elif isinstance(pub, EllipticCurvePublicKey): + # only P-256 is accepted here because securesystemslib.CryptoSigner + # (used for signing) only implements ecdsa-sha2-nistp256; sslib's + # SSlibKey already supports *verifying* nistp384/nistp521 + # signatures, so this is a signing-side limitation, not a + # fundamental one if not isinstance(pub.curve, SECP256R1): raise ValueError(f"unsupported EC curve '{pub.curve.name}'") if scheme not in (None, DEFAULT_ECDSA_SIGNATURE_SCHEME): From aa2d935defb4bef814534b7fb722613d4b0af2a5 Mon Sep 17 00:00:00 2001 From: Renata Date: Thu, 9 Jul 2026 00:32:13 +0200 Subject: [PATCH 6/6] chore: update changelog --- CHANGELOG.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 04ed1533..a24f7ede 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,8 +11,14 @@ and this project adheres to [Semantic Versioning][semver]. ### Changed +- Remove unused `scheme` parameters ([757]) + ### Fixed +- Detect signing scheme from key material instead of assuming RSA ([757]) + + +[757]: https://github.com/openlawlibrary/taf/pull/757 ## [0.39.0]