61
61
import fnmatch
62
62
import logging
63
63
import os
64
- from typing import Any , Dict , List , Optional
64
+ from typing import Dict , List , Optional
65
65
from urllib import parse
66
66
67
67
from securesystemslib import hash as sslib_hash
68
68
from securesystemslib import util as sslib_util
69
69
70
70
from tuf import exceptions
71
+ from tuf .api .metadata import TargetFile
71
72
from tuf .ngclient ._internal import requests_fetcher , trusted_metadata_set
72
73
from tuf .ngclient .config import UpdaterConfig
73
74
from tuf .ngclient .fetcher import FetcherInterface
@@ -143,8 +144,8 @@ def refresh(self) -> None:
143
144
self ._load_snapshot ()
144
145
self ._load_targets ("targets" , "root" )
145
146
146
- def get_one_valid_targetinfo (self , target_path : str ) -> Dict :
147
- """Returns target information for 'target_path'.
147
+ def get_one_valid_targetinfo (self , target_path : str ) -> TargetFile :
148
+ """Returns TargetFile instance with information for 'target_path'.
148
149
149
150
The return value can be used as an argument to
150
151
:func:`download_target()` and :func:`updated_targets()`.
@@ -174,8 +175,8 @@ def get_one_valid_targetinfo(self, target_path: str) -> Dict:
174
175
175
176
@staticmethod
176
177
def updated_targets (
177
- targets : List [Dict [ str , Any ] ], destination_directory : str
178
- ) -> List [Dict [ str , Any ] ]:
178
+ targets : List [TargetFile ], destination_directory : str
179
+ ) -> List [TargetFile ]:
179
180
"""Checks whether local cached target files are up to date
180
181
181
182
After retrieving the target information for the targets that should be
@@ -198,17 +199,16 @@ def updated_targets(
198
199
# against each hash listed for its fileinfo. Note: join() discards
199
200
# 'destination_directory' if 'filepath' contains a leading path
200
201
# separator (i.e., is treated as an absolute path).
201
- filepath = target ["filepath" ]
202
- target_fileinfo : "TargetFile" = target ["fileinfo" ]
203
-
204
- target_filepath = os .path .join (destination_directory , filepath )
202
+ target_filepath = os .path .join (
203
+ destination_directory , target .targetname
204
+ )
205
205
206
206
if target_filepath in updated_targetpaths :
207
207
continue
208
208
209
209
try :
210
210
with open (target_filepath , "rb" ) as target_file :
211
- target_fileinfo .verify_length_and_hashes (target_file )
211
+ target .verify_length_and_hashes (target_file )
212
212
# If the file does not exist locally or length and hashes
213
213
# do not match, append to updated targets.
214
214
except (OSError , exceptions .LengthOrHashMismatchError ):
@@ -219,15 +219,15 @@ def updated_targets(
219
219
220
220
def download_target (
221
221
self ,
222
- targetinfo : Dict ,
222
+ targetinfo : TargetFile ,
223
223
destination_directory : str ,
224
224
target_base_url : Optional [str ] = None ,
225
225
):
226
226
"""Downloads the target file specified by 'targetinfo'.
227
227
228
228
Args:
229
- targetinfo: data received from get_one_valid_targetinfo() or
230
- updated_targets().
229
+ targetinfo: TargetFile instance received from
230
+ get_one_valid_targetinfo() or updated_targets().
231
231
destination_directory: existing local directory to download into.
232
232
Note that new directories may be created inside
233
233
destination_directory as required.
@@ -248,15 +248,14 @@ def download_target(
248
248
else :
249
249
target_base_url = _ensure_trailing_slash (target_base_url )
250
250
251
- target_filepath = targetinfo ["filepath" ]
252
- target_fileinfo : "TargetFile" = targetinfo ["fileinfo" ]
251
+ target_filepath = targetinfo .targetname
253
252
full_url = parse .urljoin (target_base_url , target_filepath )
254
253
255
254
with self ._fetcher .download_file (
256
- full_url , target_fileinfo .length
255
+ full_url , targetinfo .length
257
256
) as target_file :
258
257
try :
259
- target_fileinfo .verify_length_and_hashes (target_file )
258
+ targetinfo .verify_length_and_hashes (target_file )
260
259
except exceptions .LengthOrHashMismatchError as e :
261
260
raise exceptions .RepositoryError (
262
261
f"{ target_filepath } length or hashes do not match"
@@ -368,7 +367,7 @@ def _load_targets(self, role: str, parent_role: str) -> None:
368
367
self ._trusted_set .update_delegated_targets (data , role , parent_role )
369
368
self ._persist_metadata (role , data )
370
369
371
- def _preorder_depth_first_walk (self , target_filepath ) -> Dict :
370
+ def _preorder_depth_first_walk (self , target_filepath ) -> TargetFile :
372
371
"""
373
372
Interrogates the tree of target delegations in order of appearance
374
373
(which implicitly order trustworthiness), and returns the matching
@@ -460,7 +459,8 @@ def _preorder_depth_first_walk(self, target_filepath) -> Dict:
460
459
self .config .max_delegations ,
461
460
)
462
461
463
- return {"filepath" : target_filepath , "fileinfo" : target }
462
+ target .targetname = target_filepath
463
+ return target
464
464
465
465
466
466
def _visit_child_role (child_role : Dict , target_filepath : str ) -> str :
0 commit comments