|
4 | 4 | """TUF client workflow implementation. |
5 | 5 | """ |
6 | 6 |
|
7 | | -import fnmatch |
8 | 7 | import logging |
9 | 8 | import os |
10 | 9 | from typing import Any, Dict, List, Optional |
11 | 10 | from urllib import parse |
12 | 11 |
|
13 | | -from securesystemslib import hash as sslib_hash |
14 | 12 | from securesystemslib import util as sslib_util |
15 | 13 |
|
16 | 14 | from tuf import exceptions |
@@ -351,8 +349,8 @@ def _preorder_depth_first_walk(self, target_filepath) -> Dict: |
351 | 349 | # NOTE: This may be a slow operation if there are many |
352 | 350 | # delegated roles. |
353 | 351 | for child_role in child_roles: |
354 | | - child_role_name = _visit_child_role( |
355 | | - child_role, target_filepath |
| 352 | + child_role_name = child_role.visit_child_role( |
| 353 | + target_filepath |
356 | 354 | ) |
357 | 355 |
|
358 | 356 | if child_role.terminating and child_role_name is not None: |
@@ -400,117 +398,6 @@ def _preorder_depth_first_walk(self, target_filepath) -> Dict: |
400 | 398 | return {"filepath": target_filepath, "fileinfo": target} |
401 | 399 |
|
402 | 400 |
|
403 | | -def _visit_child_role(child_role: Dict, target_filepath: str) -> str: |
404 | | - """ |
405 | | - <Purpose> |
406 | | - Non-public method that determines whether the given 'target_filepath' |
407 | | - is an allowed path of 'child_role'. |
408 | | -
|
409 | | - Ensure that we explore only delegated roles trusted with the target. The |
410 | | - metadata for 'child_role' should have been refreshed prior to this point, |
411 | | - however, the paths/targets that 'child_role' signs for have not been |
412 | | - verified (as intended). The paths/targets that 'child_role' is allowed |
413 | | - to specify in its metadata depends on the delegating role, and thus is |
414 | | - left to the caller to verify. We verify here that 'target_filepath' |
415 | | - is an allowed path according to the delegated 'child_role'. |
416 | | -
|
417 | | - TODO: Should the TUF spec restrict the repository to one particular |
418 | | - algorithm? Should we allow the repository to specify in the role |
419 | | - dictionary the algorithm used for these generated hashed paths? |
420 | | -
|
421 | | - <Arguments> |
422 | | - child_role: |
423 | | - The delegation targets role object of 'child_role', containing its |
424 | | - paths, path_hash_prefixes, keys, and so on. |
425 | | -
|
426 | | - target_filepath: |
427 | | - The path to the target file on the repository. This will be relative to |
428 | | - the 'targets' (or equivalent) directory on a given mirror. |
429 | | -
|
430 | | - <Exceptions> |
431 | | - None. |
432 | | -
|
433 | | - <Side Effects> |
434 | | - None. |
435 | | -
|
436 | | - <Returns> |
437 | | - If 'child_role' has been delegated the target with the name |
438 | | - 'target_filepath', then we return the role name of 'child_role'. |
439 | | -
|
440 | | - Otherwise, we return None. |
441 | | - """ |
442 | | - |
443 | | - child_role_name = child_role.name |
444 | | - child_role_paths = child_role.paths |
445 | | - child_role_path_hash_prefixes = child_role.path_hash_prefixes |
446 | | - |
447 | | - if child_role_path_hash_prefixes is not None: |
448 | | - target_filepath_hash = _get_filepath_hash(target_filepath) |
449 | | - for child_role_path_hash_prefix in child_role_path_hash_prefixes: |
450 | | - if not target_filepath_hash.startswith(child_role_path_hash_prefix): |
451 | | - continue |
452 | | - |
453 | | - return child_role_name |
454 | | - |
455 | | - elif child_role_paths is not None: |
456 | | - # Is 'child_role_name' allowed to sign for 'target_filepath'? |
457 | | - for child_role_path in child_role_paths: |
458 | | - # A child role path may be an explicit path or glob pattern (Unix |
459 | | - # shell-style wildcards). The child role 'child_role_name' is |
460 | | - # returned if 'target_filepath' is equal to or matches |
461 | | - # 'child_role_path'. Explicit filepaths are also considered |
462 | | - # matches. A repo maintainer might delegate a glob pattern with a |
463 | | - # leading path separator, while the client requests a matching |
464 | | - # target without a leading path separator - make sure to strip any |
465 | | - # leading path separators so that a match is made. |
466 | | - # Example: "foo.tgz" should match with "/*.tgz". |
467 | | - if fnmatch.fnmatch( |
468 | | - target_filepath.lstrip(os.sep), child_role_path.lstrip(os.sep) |
469 | | - ): |
470 | | - logger.debug( |
471 | | - "Child role %s is allowed to sign for %s", |
472 | | - repr(child_role_name), |
473 | | - repr(target_filepath), |
474 | | - ) |
475 | | - |
476 | | - return child_role_name |
477 | | - |
478 | | - logger.debug( |
479 | | - "The given target path %s " |
480 | | - "does not match the trusted path or glob pattern: %s", |
481 | | - repr(target_filepath), |
482 | | - repr(child_role_path), |
483 | | - ) |
484 | | - continue |
485 | | - |
486 | | - else: |
487 | | - # 'role_name' should have been validated when it was downloaded. |
488 | | - # The 'paths' or 'path_hash_prefixes' fields should not be missing, |
489 | | - # so we raise a format error here in case they are both missing. |
490 | | - raise exceptions.FormatError( |
491 | | - repr(child_role_name) + " " |
492 | | - 'has neither a "paths" nor "path_hash_prefixes". At least' |
493 | | - " one of these attributes must be present." |
494 | | - ) |
495 | | - |
496 | | - return None |
497 | | - |
498 | | - |
499 | | -def _get_filepath_hash(target_filepath, hash_function="sha256"): |
500 | | - """ |
501 | | - Calculate the hash of the filepath to determine which bin to find the |
502 | | - target. |
503 | | - """ |
504 | | - # The client currently assumes the repository (i.e., repository |
505 | | - # tool) uses 'hash_function' to generate hashes and UTF-8. |
506 | | - digest_object = sslib_hash.digest(hash_function) |
507 | | - encoded_target_filepath = target_filepath.encode("utf-8") |
508 | | - digest_object.update(encoded_target_filepath) |
509 | | - target_filepath_hash = digest_object.hexdigest() |
510 | | - |
511 | | - return target_filepath_hash |
512 | | - |
513 | | - |
514 | 401 | def _ensure_trailing_slash(url: str): |
515 | 402 | """Return url guaranteed to end in a slash""" |
516 | 403 | return url if url.endswith("/") else f"{url}/" |
0 commit comments