@@ -42,8 +42,7 @@ def __init__(
4242 self ._check_mirrors ()
4343
4444 # Will hold a list of all the gpg keys (public and private)
45- # self._keys: Optional[List[pathlib.Path]] = []
46- self ._keys = self ._key_init ()
45+ self ._keys : Optional [List [pathlib .Path ]] = []
4746
4847 def _load_mirrors (self , cmdline_cache : Optional [pathlib .Path ]) -> Dict [str , Dict ]:
4948 """Load the mirrors file, if one exists."""
@@ -159,12 +158,14 @@ def _check_mirrors(self):
159158 f"Check the url listed in mirrors.yaml in system config. \n { err }
160159 )
161160
162- def key_files (self , config_root : pathlib .Path ):
161+ @property
162+ def keys (self ):
163163 """Return the list of public and private key file paths."""
164+
164165 if self ._keys is None :
165166 raise RuntimeError ("The mirror.keys method was accessed before setup_configs() was called." )
166- key_dir = config_root / self . KEY_STORE_DIR
167- return [ key_dir / info [ "path" ] for info in self ._keys ]
167+
168+ return self ._keys
168169
169170 def setup_configs (self , config_root : pathlib .Path ):
170171 """Setup all mirror configs in the given config_root."""
@@ -233,67 +234,64 @@ def _create_bootstrap_configs(self, config_root: pathlib.Path):
233234 with (config_root / "bootstrap.yaml" ).open ("w" ) as file :
234235 yaml .dump (bootstrap_yaml , file , default_flow_style = False )
235236
236- def _key_init (self ):
237- key_info = {}
238-
239- key = self .mirrors ["buildcache" ].get ("private_key" )
237+ def _load_key (self , key : str , dest : pathlib .Path , name : str ):
238+ """Validate mirror keys, relocate to key_store, and update mirror config with new key paths."""
240239
241- if key is None :
242- return
240+ # key will be saved under key_store/mirror_name.[pub/priv].gpg
243241
244242 # if path, check if abs path, if not, append sys config path in front and check again
245243 path = pathlib .Path (os .path .expandvars (key ))
246-
247244 if not path .is_absolute ():
248245 # try prepending system config path
249246 path = self ._system_config_root / path
250247
251248 if path .is_file ():
252- # use the user-provided file
253- key_info = {"path" : pathlib .Path (f"buildcache.pgp" ), "source" : path }
249+ with open (path , "rb" ) as reader :
250+ binary_key = reader .read ()
251+
252+ # convert base64 key to binary
254253 else :
255- # convert base64 key to binary
256254 try :
257- binary_key = base64 .b64decode (key , validate = True )
258- print (binary_key )
255+ binary_key = base64 .b64decode (key )
259256 except ValueError :
260257 raise MirrorError (
261- f"Key for mirror 'buildcache ' is not valid. \n "
258+ f"Key for mirror '{ name } : ' { path } ' . \n "
262259 f"Must be a path to a GPG public key or a base64 encoded GPG public key. \n "
263260 f"Check the key listed in mirrors.yaml in system config."
264261 )
265262
266- file_type = magic .from_buffer (binary_key , mime = True )
267- if file_type not in "application/x-gnupg-keyring" , "application/pgp-keys" ):
268- raise MirrorError (
269- f"Key for mirror 'buildcache' is not a valid GPG key. \n "
270- f"The file (or base64) was readable, but the data itself was not a PGP key.\n "
271- f"Check the key listed in mirrors.yaml in system config."
272- )
273-
274- key_info = {"path" : pathlib .Path ("buildcache.pgp" ), "source" : binary_key }
263+ # private keys will evaluate as "application/octet-stream"
264+ file_type = magic .from_buffer (binary_key , mime = True )
265+ if file_type not in "application/x-gnupg-keyring" , "application/pgp-keys" , "application/octet-stream" ):
266+ raise MirrorError (
267+ f"Key for mirror { name } \n "
268+ f"The file (or base64) was readable, but the data itself was not a PGP key.\n "
269+ f"Check the key listed in mirrors.yaml in system config."
270+ )
275271
276- return key_info
272+ # copy key to new destination in key store
273+ with open (dest , "wb" ) as writer :
274+ writer .write (binary_key )
277275
276+ self ._keys .append (dest )
277+
278+
278279 def _key_setup (self , key_store : pathlib .Path ):
279- """Validate mirror keys, relocate to key_store, and update mirror config with new key paths. """
280+ """Iterate through mirror keys and load + relocate each one to key_store """
280281
282+ self ._keys = []
281283 key_store .mkdir (exist_ok = True )
282284
283- #for key_info in self._keys:
284-
285- #path = key_store / key_info["path"]
286- path = key_store / self ._keys ["path" ]
287- #source = key_info["source"]
288- source = self ._keys ["source" ]
289-
290- match source :
291- case pathlib .Path ():
292- # copy source -> path
293- shutil .copy2 (source , path )
294- case bytes ():
295- # open path and copy in bytes
296- with open (path , "wb" ) as writer :
297- writer .write (source )
298- case _:
299- raise TypeError (f"Expected Path or bytes, got { type (source ).__name__ } )
285+ for name , mirror in self .mirrors .items ():
286+ if name == "buildcache" :
287+ if mirror .get ("private_key" ):
288+ key = mirror ["private_key" ]
289+ dest = pathlib .Path (key_store / f"{ name } )
290+ self ._load_key (key , dest , name )
291+
292+ if mirror .get ("public_key" ) is None :
293+ continue
294+
295+ key = mirror ["public_key" ]
296+ dest = pathlib .Path (key_store / f"{ name } )
297+ self ._load_key (key , dest , name )
0 commit comments