@@ -108,7 +108,7 @@ def __init__(self):
108
108
# target downloads are served from this dict
109
109
self .target_files : Dict [str , RepositoryTarget ] = {}
110
110
111
- # Whether to compute hashes and legth for meta in snapshot/timestamp
111
+ # Whether to compute hashes and length for meta in snapshot/timestamp
112
112
self .compute_metafile_hashes_length = False
113
113
114
114
self .dump_dir = None
@@ -145,6 +145,21 @@ def create_key() -> Tuple[Key, SSlibSigner]:
145
145
sslib_key = generate_ed25519_key ()
146
146
return Key .from_securesystemslib_key (sslib_key ), SSlibSigner (sslib_key )
147
147
148
+ @staticmethod
149
+ def _partition_filename (filename : str ) -> Tuple [str , Optional [str ]]:
150
+ """Partition 'filename' into filename and prefix, if any"""
151
+ p1 , sep , p2 = filename .partition ("." )
152
+ # If the separator is not found, returns a 3-tuple containing
153
+ # the string itself, followed by two empty strings.
154
+ if not sep :
155
+ prefix = None
156
+ filename = p1
157
+ else :
158
+ prefix = p1
159
+ filename = p2
160
+
161
+ return filename , prefix
162
+
148
163
def add_signer (self , role : str , signer : SSlibSigner ):
149
164
if role not in self .signers :
150
165
self .signers [role ] = {}
@@ -185,24 +200,19 @@ def publish_root(self):
185
200
logger .debug ("Published root v%d" , self .root .version )
186
201
187
202
def fetch (self , url : str ) -> Iterator [bytes ]:
188
- if not self .root .consistent_snapshot :
189
- raise NotImplementedError ("non-consistent snapshot not supported" )
190
203
path = parse .urlparse (url ).path
191
204
if path .startswith ("/metadata/" ) and path .endswith (".json" ):
192
205
ver_and_name = path [len ("/metadata/" ) :][: - len (".json" )]
193
- # only consistent_snapshot supported ATM: timestamp is special case
194
- if ver_and_name == "timestamp" :
195
- version = None
196
- role = "timestamp"
197
- else :
198
- version , _ , role = ver_and_name .partition ("." )
206
+ role , version = self ._partition_filename (ver_and_name )
207
+ if version :
199
208
version = int (version )
209
+
200
210
yield self ._fetch_metadata (role , version )
201
211
elif path .startswith ("/targets/" ):
202
212
# figure out target path and hash prefix
203
213
target_path = path [len ("/targets/" ) :]
204
214
dir_parts , sep , prefixed_filename = target_path .rpartition ("/" )
205
- prefix , _ , filename = prefixed_filename . partition ( "." )
215
+ filename , prefix = self . _partition_filename ( prefixed_filename )
206
216
target_path = f"{ dir_parts } { sep } { filename } "
207
217
208
218
yield self ._fetch_target (target_path , prefix )
@@ -245,7 +255,7 @@ def _fetch_metadata(
245
255
elif role == "targets" :
246
256
md = self .md_targets
247
257
else :
248
- md = self .md_delegates [ role ]
258
+ md = self .md_delegates . get ( role )
249
259
250
260
if md is None :
251
261
raise FetcherHTTPError (f"Unknown role { role } " , 404 )
0 commit comments