diff --git a/CHANGELOG.md b/CHANGELOG.md index 04ed15336..a24f7ede7 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] diff --git a/taf/api/api_workflow.py b/taf/api/api_workflow.py index 71d47c78b..1ad13ec1a 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 9bc13556d..18bb28131 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 ff0dec79f..22be90882 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 b719abdf4..fcac79891 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,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. - 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. @@ -321,7 +319,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 +350,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 +366,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 +379,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 +390,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 +430,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 +445,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 +465,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 +512,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 +531,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 +544,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 +557,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 +569,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 +691,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 +699,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 +1085,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 +1128,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 0d24d53ec..d6fc9baff 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 @@ -118,11 +121,12 @@ 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, 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 @@ -148,7 +152,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 +340,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 +357,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 +408,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 +449,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 +490,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 +577,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 +589,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 +612,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 +640,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 +653,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 +698,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/constants.py b/taf/constants.py index 310c82620..06eea1195 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/keys.py b/taf/keys.py index e20f11170..5f4e177d7 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: @@ -606,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/keystore.py b/taf/keystore.py index 55b9e69ba..fbac811a8 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_api/roles/api/test_roles.py b/taf/tests/test_api/roles/api/test_roles.py index 6adb539e6..749035879 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,32 @@ 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, +): + 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 b71f1c6d6..02cde149e 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,33 @@ def test_update_target_repos_from_repositories_json( assert commits[0].message.strip() == git_commit_message("update-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, +): + 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 +263,38 @@ 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, +): + 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 1cc625723..7cea85b2d 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/tests/test_updater/conftest.py b/taf/tests/test_updater/conftest.py index 1db2d2cd2..677bc9e3f 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/tests/tuf/test_keys/test_ecdsa_keys.py b/taf/tests/tuf/test_keys/test_ecdsa_keys.py new file mode 100644 index 000000000..27580a159 --- /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 ( + generate_rsa_keypair, + 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(): + _, 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/tools/metadata/__init__.py b/taf/tools/metadata/__init__.py index ea77f4b02..e63acc17b 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 375a7af69..22f3a7255 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 f453beec4..9eb847511 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, diff --git a/taf/tuf/keys.py b/taf/tuf/keys.py index 6bca7af23..e597de35f 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]: @@ -74,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. """ @@ -101,12 +103,37 @@ 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: Optional[str] = None, +) -> 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, 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 not isinstance(pub, RSAPublicKey): + 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): + 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) # FIXME: include the 'keyid_hash_algorithms' entry in the key portion @@ -117,17 +144,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() @@ -135,13 +159,11 @@ def load_public_key_from_file( return _from_crypto(pub) -def load_signer_from_file( - path: Path, password: Optional[str] = None, scheme=DEFAULT_RSA_SIGNATURE_SCHEME -) -> CryptoSigner: - """Load CryptoSigner from RSA private key file. +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 - * 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,26 +171,29 @@ 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() return CryptoSigner(priv, _from_crypto(pub)) -def load_signer_from_pem(pem: bytes, password: Optional[bytes] = None) -> CryptoSigner: - """Load CryptoSigner from RSA private key file. +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 defaults to 'rsa-pkcs1v15-sha256' + * 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): diff --git a/taf/tuf/repository.py b/taf/tuf/repository.py index 3bfda17cb..d3f5a5bfd 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.