|
6 | 6 |
|
7 | 7 | import logging |
8 | 8 | import os |
9 | | -from typing import Any, Dict, List, Optional, Union |
| 9 | +from typing import Any, Dict, List, Optional, Set, Tuple, Union |
10 | 10 | from urllib import parse |
11 | 11 |
|
12 | 12 | from securesystemslib import util as sslib_util |
@@ -114,7 +114,10 @@ def get_one_valid_targetinfo( |
114 | 114 | RepositoryError: Metadata failed to verify in some way |
115 | 115 | TODO: download-related errors |
116 | 116 | """ |
117 | | - return self._preorder_depth_first_walk(target_path) |
| 117 | + targetinfo, dummy = self._preorder_depth_first_walk( |
| 118 | + target_path, set(), ("targets", "root"), self.config.max_delegations |
| 119 | + ) |
| 120 | + return targetinfo |
118 | 121 |
|
119 | 122 | @staticmethod |
120 | 123 | def updated_targets( |
@@ -316,76 +319,63 @@ def _load_targets(self, role: str, parent_role: str) -> None: |
316 | 319 | self._persist_metadata(role, data) |
317 | 320 |
|
318 | 321 | def _preorder_depth_first_walk( |
319 | | - self, target_filepath: str |
320 | | - ) -> Union[Dict[str, Any], None]: |
| 322 | + self, |
| 323 | + target_filepath: str, |
| 324 | + visited_role_names: Set[str], |
| 325 | + current_role_pair: List[Tuple[str, ...]], |
| 326 | + number_of_delegations: int, |
| 327 | + ) -> Tuple[Union[Dict[str, Any], None], bool]: |
321 | 328 | """ |
322 | 329 | Interrogates the tree of target delegations in order of appearance |
323 | 330 | (which implicitly order trustworthiness), and returns the matching |
324 | 331 | target found in the most trusted role. |
325 | 332 | """ |
326 | | - |
327 | | - role_names = [("targets", "root")] |
328 | | - visited_role_names = set() |
329 | | - number_of_delegations = self.config.max_delegations |
330 | | - |
| 333 | + targetinfo = None |
| 334 | + terminated = False |
331 | 335 | # Preorder depth-first traversal of the graph of target delegations. |
332 | | - while number_of_delegations > 0 and len(role_names) > 0: |
333 | | - |
334 | | - # Pop the role name from the top of the stack. |
335 | | - role_name, parent_role = role_names.pop(-1) |
336 | | - |
337 | | - # Skip any visited current role to prevent cycles. |
338 | | - if (role_name, parent_role) in visited_role_names: |
339 | | - logger.debug("Skipping visited current role %s", role_name) |
340 | | - continue |
341 | | - |
342 | | - # The metadata for 'role_name' must be downloaded/updated before |
343 | | - # its targets, delegations, and child roles can be inspected. |
344 | | - self._load_targets(role_name, parent_role) |
345 | | - |
346 | | - role_metadata: Targets = self._trusted_set[role_name].signed |
347 | | - target = role_metadata.targets.get(target_filepath) |
348 | | - |
349 | | - if target is not None: |
350 | | - logger.debug("Found target in current role %s", role_name) |
351 | | - return {"filepath": target_filepath, "fileinfo": target} |
352 | | - |
353 | | - # After preorder check, add current role to set of visited roles. |
354 | | - visited_role_names.add((role_name, parent_role)) |
355 | | - |
356 | | - # And also decrement number of visited roles. |
357 | | - number_of_delegations -= 1 |
358 | | - |
359 | | - if role_metadata.delegations is not None: |
360 | | - child_roles_to_visit = [] |
361 | | - # NOTE: This may be a slow operation if there are many |
362 | | - # delegated roles. |
363 | | - for child_role in role_metadata.delegations.roles: |
364 | | - if child_role.is_in_trusted_paths(target_filepath): |
365 | | - logger.debug("Adding child role %s", child_role.name) |
366 | | - |
367 | | - child_roles_to_visit.append( |
368 | | - (child_role.name, role_name) |
369 | | - ) |
370 | | - if child_role.terminating: |
371 | | - logger.debug("Not backtracking to other roles.") |
372 | | - role_names = [] |
373 | | - break |
374 | | - # Push 'child_roles_to_visit' in reverse order of appearance |
375 | | - # onto 'role_names'. Roles are popped from the end of |
376 | | - # the 'role_names' list. |
377 | | - child_roles_to_visit.reverse() |
378 | | - role_names.extend(child_roles_to_visit) |
379 | | - |
380 | | - if number_of_delegations == 0 and len(role_names) > 0: |
381 | | - logger.debug( |
382 | | - "%d roles left to visit, but allowed to visit at most %d.", |
383 | | - len(role_names), |
384 | | - self.config.max_delegations, |
385 | | - ) |
386 | | - |
387 | | - # If this point is reached then target is not found, return None |
388 | | - return None |
| 336 | + if number_of_delegations <= 0: |
| 337 | + return targetinfo, terminated |
| 338 | + |
| 339 | + # Pop the role name from the top of the stack. |
| 340 | + role_name, parent_role = current_role_pair |
| 341 | + |
| 342 | + # The metadata for 'role_name' must be downloaded/updated before |
| 343 | + # its targets, delegations, and child roles can be inspected. |
| 344 | + self._load_targets(role_name, parent_role) |
| 345 | + role_metadata: Targets = self._trusted_set[role_name].signed |
| 346 | + target = role_metadata.targets.get(target_filepath) |
| 347 | + |
| 348 | + if target is not None: |
| 349 | + logger.debug("Found target in current role %s", role_name) |
| 350 | + targetinfo = {"filepath": target_filepath, "fileinfo": target} |
| 351 | + return targetinfo, terminated |
| 352 | + |
| 353 | + # After preorder check, add current role to set of visited roles. |
| 354 | + visited_role_names.add((role_name, parent_role)) |
| 355 | + |
| 356 | + # And also decrement number of visited roles. |
| 357 | + number_of_delegations -= 1 |
| 358 | + if role_metadata.delegations is not None: |
| 359 | + for child_role in role_metadata.delegations.roles: |
| 360 | + # Skip any visited current role to prevent cycles. |
| 361 | + if (child_role.name, parent_role) in visited_role_names: |
| 362 | + continue |
| 363 | + |
| 364 | + if child_role.is_in_trusted_paths(target_filepath): |
| 365 | + |
| 366 | + targetinfo, terminated = self._preorder_depth_first_walk( |
| 367 | + target_filepath, |
| 368 | + visited_role_names, |
| 369 | + (child_role.name, role_name), |
| 370 | + number_of_delegations, |
| 371 | + ) |
| 372 | + |
| 373 | + if child_role.terminating or terminated: |
| 374 | + terminated = True |
| 375 | + logger.debug("Not backtracking to other roles.") |
| 376 | + break |
| 377 | + |
| 378 | + return targetinfo, terminated |
389 | 379 |
|
390 | 380 |
|
391 | 381 | def _ensure_trailing_slash(url: str): |
|
0 commit comments