62
62
63
63
import logging
64
64
import os
65
- from typing import Any , Dict , List , Optional , Set , Tuple
65
+ from typing import List , Optional , Set , Tuple
66
66
from urllib import parse
67
67
68
68
from securesystemslib import util as sslib_util
69
69
70
70
from tuf import exceptions
71
- from tuf .api .metadata import Targets
71
+ from tuf .api .metadata import TargetFile , Targets
72
72
from tuf .ngclient ._internal import requests_fetcher , trusted_metadata_set
73
73
from tuf .ngclient .config import UpdaterConfig
74
74
from tuf .ngclient .fetcher import FetcherInterface
@@ -146,8 +146,8 @@ def refresh(self) -> None:
146
146
147
147
def get_one_valid_targetinfo (
148
148
self , target_path : str
149
- ) -> Optional [Dict [ str , Any ] ]:
150
- """Returns target information for 'target_path'.
149
+ ) -> Optional [TargetFile ]:
150
+ """Returns TargetFile instance with information for 'target_path'.
151
151
152
152
The return value can be used as an argument to
153
153
:func:`download_target()` and :func:`updated_targets()`.
@@ -174,14 +174,14 @@ def get_one_valid_targetinfo(
174
174
TODO: download-related errors
175
175
176
176
Returns:
177
- A targetinfo dictionary or None
177
+ A TargetFile instance or None.
178
178
"""
179
179
return self ._preorder_depth_first_walk (target_path )
180
180
181
181
@staticmethod
182
182
def updated_targets (
183
- targets : List [Dict [ str , Any ] ], destination_directory : str
184
- ) -> List [Dict [ str , Any ] ]:
183
+ targets : List [TargetFile ], destination_directory : str
184
+ ) -> List [TargetFile ]:
185
185
"""Checks whether local cached target files are up to date
186
186
187
187
After retrieving the target information for the targets that should be
@@ -204,17 +204,14 @@ def updated_targets(
204
204
# against each hash listed for its fileinfo. Note: join() discards
205
205
# 'destination_directory' if 'filepath' contains a leading path
206
206
# separator (i.e., is treated as an absolute path).
207
- filepath = target ["filepath" ]
208
- target_fileinfo : "TargetFile" = target ["fileinfo" ]
209
-
210
- target_filepath = os .path .join (destination_directory , filepath )
207
+ target_filepath = os .path .join (destination_directory , target .path )
211
208
212
209
if target_filepath in updated_targetpaths :
213
210
continue
214
211
215
212
try :
216
213
with open (target_filepath , "rb" ) as target_file :
217
- target_fileinfo .verify_length_and_hashes (target_file )
214
+ target .verify_length_and_hashes (target_file )
218
215
# If the file does not exist locally or length and hashes
219
216
# do not match, append to updated targets.
220
217
except (OSError , exceptions .LengthOrHashMismatchError ):
@@ -225,15 +222,15 @@ def updated_targets(
225
222
226
223
def download_target (
227
224
self ,
228
- targetinfo : Dict ,
225
+ targetinfo : TargetFile ,
229
226
destination_directory : str ,
230
227
target_base_url : Optional [str ] = None ,
231
228
):
232
229
"""Downloads the target file specified by 'targetinfo'.
233
230
234
231
Args:
235
- targetinfo: data received from get_one_valid_targetinfo() or
236
- updated_targets().
232
+ targetinfo: TargetFile instance received from
233
+ get_one_valid_targetinfo() or updated_targets().
237
234
destination_directory: existing local directory to download into.
238
235
Note that new directories may be created inside
239
236
destination_directory as required.
@@ -254,27 +251,26 @@ def download_target(
254
251
else :
255
252
target_base_url = _ensure_trailing_slash (target_base_url )
256
253
257
- target_fileinfo : "TargetFile" = targetinfo ["fileinfo" ]
258
- target_filepath = targetinfo ["filepath" ]
254
+ target_filepath = targetinfo .path
259
255
consistent_snapshot = self ._trusted_set .root .signed .consistent_snapshot
260
256
if consistent_snapshot and self .config .prefix_targets_with_hash :
261
- hashes = list (target_fileinfo .hashes .values ())
257
+ hashes = list (targetinfo .hashes .values ())
262
258
target_filepath = f"{ hashes [0 ]} .{ target_filepath } "
263
259
full_url = parse .urljoin (target_base_url , target_filepath )
264
260
265
261
with self ._fetcher .download_file (
266
- full_url , target_fileinfo .length
262
+ full_url , targetinfo .length
267
263
) as target_file :
268
264
try :
269
- target_fileinfo .verify_length_and_hashes (target_file )
265
+ targetinfo .verify_length_and_hashes (target_file )
270
266
except exceptions .LengthOrHashMismatchError as e :
271
267
raise exceptions .RepositoryError (
272
268
f"{ target_filepath } length or hashes do not match"
273
269
) from e
274
270
275
271
# Store the target file name without the HASH prefix.
276
272
local_filepath = os .path .join (
277
- destination_directory , targetinfo [ "filepath" ]
273
+ destination_directory , targetinfo . path
278
274
)
279
275
sslib_util .persist_temp_file (target_file , local_filepath )
280
276
@@ -380,7 +376,7 @@ def _load_targets(self, role: str, parent_role: str) -> None:
380
376
381
377
def _preorder_depth_first_walk (
382
378
self , target_filepath : str
383
- ) -> Optional [Dict [ str , Any ] ]:
379
+ ) -> Optional [TargetFile ]:
384
380
"""
385
381
Interrogates the tree of target delegations in order of appearance
386
382
(which implicitly order trustworthiness), and returns the matching
@@ -413,7 +409,7 @@ def _preorder_depth_first_walk(
413
409
414
410
if target is not None :
415
411
logger .debug ("Found target in current role %s" , role_name )
416
- return { "filepath" : target_filepath , "fileinfo" : target }
412
+ return target
417
413
418
414
# After preorder check, add current role to set of visited roles.
419
415
visited_role_names .add ((role_name , parent_role ))
0 commit comments