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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 44 additions & 14 deletions taf/api/roles.py
Original file line number Diff line number Diff line change
Expand Up @@ -917,38 +917,68 @@ def _get_roles_key_size(role: str, keystore: str, keys_num: int) -> int:
return get_key_size(key_path)


def _format_role_keys_block(role_label: str, key_ids: List[str], certs_dir: str) -> str:
key_lines = [
str(get_metadata_key_info(certs_dir, key_id)) for key_id in key_ids
]
return f"Role: {role_label}\n" + "\n".join(key_lines)


def _roles_for_keys(auth_repo: AuthenticationRepository) -> Dict[str, set]:

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It was surprising that a similar utility was not already somewhere in the codebase. Let's move this to tuf/repository/MetadataRepository

roles_for_key: Dict[str, set] = defaultdict(set)
for role_name in auth_repo.get_all_roles():

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This works, but is not the most efficient solution. That is fine as we do not expect to ever encounter a situation where we have hundreds of roles.

get_all_roles() walks the full delegation tree and in doing so, it already knows each role's parent, but throws that away and returns a flat list. Then, for each of those role names, get_role_keys(role_name) has to call find_delegated_roles_parent(role). Implementing something similar to find_keysid_roles would be more efficient.

However, this is fine as is, unless you want to tackle the harder implementation as a way of getting more familiar with the system.

role_key_ids = auth_repo.get_role_keys(role=role_name)
if role_key_ids:
for key_id in role_key_ids:
roles_for_key[key_id].add(role_name)
return roles_for_key


@log_on_error(
ERROR,
"Could not list keys of {role:s}: {e}",
"Could not list keys: {e}",
logger=taf_logger,
on_exceptions=TAFError,
reraise=True,
)
def list_keys_of_role(

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it's actually better to split the two up:kKeep list_keys_of_role(path, role) exactly as it was (role required), and add a new function for the "all roles" case list_keys_by_roles. This preserves the existing function's contract, and the existing tests should pass without needing any modifications. And then the CLI could call one or the other.

path: str,
role: str,
role: Optional[str] = None,
) -> List[str]:
"""
Print information about signing keys of role. If a certificate whose name matches
a key's id exists, include information contained by that certificate (like name and valid to/from dates)
Return formatted information about signing keys. If a certificate whose name matches
a key's id exists, include information contained by that certificate (like name and
valid to/from dates).

Arguments:
path: Path to the authentication repository.
role: Name of the role which is to be removed.
Arguments:
path: Path to the authentication repository.
role (optional): Name of the role whose keys should be listed. If omitted, each
unique key is listed once with all roles that reference it.

Side Effects:
None
None

Returns:
None
Returns:
A list of formatted key blocks, each starting with a Role line.
"""
auth_repo = AuthenticationRepository(path=path)
key_ids = auth_repo.get_role_keys(role=role)
if key_ids is None:
raise TAFError(f"Role {role} does not exist")

if role is not None:
key_ids = auth_repo.get_role_keys(role=role)
if key_ids is None:
raise TAFError(f"Role {role} does not exist")
if not key_ids:
return []
return [_format_role_keys_block(role, key_ids, auth_repo.certs_dir)]

roles_for_key = _roles_for_keys(auth_repo)
return [
str(get_metadata_key_info(auth_repo.certs_dir, key_id)) for key_id in key_ids
_format_role_keys_block(
", ".join(sorted(roles_for_key[key_id])),
[key_id],
auth_repo.certs_dir,
)
for key_id in sorted(roles_for_key.keys())
]


Expand Down
43 changes: 38 additions & 5 deletions taf/tests/test_api/roles/api/test_roles.py
Original file line number Diff line number Diff line change
Expand Up @@ -243,13 +243,35 @@ def test_remove_role_paths(

def test_list_keys(auth_repo: AuthenticationRepository):
root_keys_infos = list_keys_of_role(str(auth_repo.path), "root")
assert len(root_keys_infos) == 3
assert len(root_keys_infos) == 1
assert root_keys_infos[0].startswith("Role: root\n")
assert root_keys_infos[0].count("Key ID:") == 3

targets_keys_infos = list_keys_of_role(str(auth_repo.path), "targets")
assert len(targets_keys_infos) == 2
assert len(targets_keys_infos) == 1
assert targets_keys_infos[0].startswith("Role: targets\n")
assert targets_keys_infos[0].count("Key ID:") == 2

snapshot_keys_infos = list_keys_of_role(str(auth_repo.path), "snapshot")
assert len(snapshot_keys_infos) == 1
assert snapshot_keys_infos[0].startswith("Role: snapshot\n")
assert snapshot_keys_infos[0].count("Key ID:") == 1

timestamp_keys_infos = list_keys_of_role(str(auth_repo.path), "timestamp")
assert len(timestamp_keys_infos) == 1
assert timestamp_keys_infos[0].startswith("Role: timestamp\n")
assert timestamp_keys_infos[0].count("Key ID:") == 1


def test_list_all_keys(auth_repo: AuthenticationRepository):
all_keys_infos = list_keys_of_role(str(auth_repo.path))
unique_keys = set()
for role_name in auth_repo.get_all_roles():
for key_id in auth_repo.get_role_keys(role=role_name) or []:
unique_keys.add(key_id)
assert len(all_keys_infos) == len(unique_keys)
assert all(info.startswith("Role:") for info in all_keys_infos)
assert all("Key ID:" in info for info in all_keys_infos)


def test_add_signing_key(
Expand All @@ -272,10 +294,20 @@ def test_add_signing_key(
commits = auth_repo.list_pygit_commits()
assert len(commits) == initial_commits_num + 1
assert commits[0].message.strip() == COMMIT_MSG
shared_key_info = next(
info
for info in list_keys_of_role(str(auth_repo.path))
if "Role: snapshot, timestamp" in info or "Role: timestamp, snapshot" in info
)
assert shared_key_info.count("Key ID:") == 1
timestamp_keys_infos = list_keys_of_role(str(auth_repo.path), "timestamp")
assert len(timestamp_keys_infos) == 2
assert len(timestamp_keys_infos) == 1
assert timestamp_keys_infos[0].startswith("Role: timestamp\n")
assert timestamp_keys_infos[0].count("Key ID:") == 2
snapshot_keys_infos = list_keys_of_role(str(auth_repo.path), "snapshot")
assert len(snapshot_keys_infos) == 2
assert len(snapshot_keys_infos) == 1
assert snapshot_keys_infos[0].startswith("Role: snapshot\n")
assert snapshot_keys_infos[0].count("Key ID:") == 2


def test_revoke_signing_key(
Expand All @@ -286,7 +318,8 @@ def test_revoke_signing_key(
key_to_remove = targest_keyids[-1]
initial_commits_num = len(auth_repo.list_pygit_commits())
targets_keys_infos = list_keys_of_role(str(auth_repo.path), "targets")
assert len(targets_keys_infos) == 2
assert len(targets_keys_infos) == 1
assert targets_keys_infos[0].count("Key ID:") == 2
COMMIT_MSG = "Revoke a targets key"
revoke_signing_key(
path=str(auth_repo.path),
Expand Down
32 changes: 29 additions & 3 deletions taf/tests/test_api/roles/cli/test_roles_cmds.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,16 +90,42 @@ def test_roles_add_signing_key_cmd_expect_success(auth_repo, roles_keystore):
],
)
timestamp_keys_infos = list_keys_of_role(str(auth_repo.path), "timestamp")
assert len(timestamp_keys_infos) == 2
# assert len(timestamp_keys_infos) == 2
assert len(timestamp_keys_infos) == 1
assert timestamp_keys_infos[0].count("Key ID:") == 2
snapshot_keys_infos = list_keys_of_role(str(auth_repo.path), "snapshot")
assert len(snapshot_keys_infos) == 2
# assert len(snapshot_keys_infos) == 2
assert len(snapshot_keys_infos) == 1
assert snapshot_keys_infos[0].count("Key ID:") == 2


def test_roles_list_keys_no_role_arg_expect_success(auth_repo, roles_keystore):
runner = CliRunner()

with runner.isolated_filesystem():
result = runner.invoke(
taf,
[
"roles",
"list-keys",
"--path",
f"{str(auth_repo.path)}",
],
)

assert result.exit_code == 0
assert "Role:" in result.output
assert "Key ID:" in result.output
assert result.output.strip().startswith("Role:")


def test_revoke_key_cmd_expect_success(auth_repo, roles_keystore):
runner = CliRunner()

targets_keys_infos = list_keys_of_role(str(auth_repo.path), "targets")
assert len(targets_keys_infos) == 2
# assert len(targets_keys_infos) == 2
assert len(targets_keys_infos) == 1
assert targets_keys_infos[0].count("Key ID:") == 2

with runner.isolated_filesystem():
targest_keyids = auth_repo.get_keyids_of_role("targets")
Expand Down
4 changes: 2 additions & 2 deletions taf/tools/roles/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -494,15 +494,15 @@ def rotate_key(

return rotate_key


def list_keys_command():
@click.command(help="""
List all keys of the specified role. If certs directory exists and contains certificates exported from YubiKeys,
include additional information read from these certificates, like name or organization.
If role isn't specified, list all keys with their roles.
""")
@find_repository
@catch_cli_exception(handle=TAFError)
@click.argument("role")
@click.argument("role", required=False, default=None)
@click.option(
"--path",
default=".",
Expand Down