feat: add initial pin manager

taf/api/repository.py

@@ -23,6 +23,7 @@
 from taf.tuf.repository import METADATA_DIRECTORY_NAME
 from taf.utils import ensure_pre_push_hook
 from taf.log import taf_logger
+from taf.yubikey.pin_manager import PinManager
 
 
 @log_on_start(
@@ -38,6 +39,7 @@
 )
 def create_repository(
     path: str,
+    pin_manager: PinManager,
     keystore: Optional[str] = None,
     roles_key_infos: Optional[str] = None,
     commit: Optional[bool] = False,
@@ -61,6 +63,7 @@ def create_repository(
     Returns:
         None
     """
+    import pdb; pdb.set_trace()
     if not _check_if_can_create_repository(Path(path)):
         return
 

taf/tools/repo/__init__.py

@@ -10,6 +10,7 @@
 from taf.tools.cli import catch_cli_exception, find_repository
 from taf.updater.types.update import UpdateType
 from taf.updater.updater import OperationType, UpdateConfig, clone_repository, update_repository, validate_repository
+from taf.yubikey.pin_manager import pin_managed
 
 
 def common_update_options(f):
@@ -121,9 +122,11 @@ def create_repo_command():
                   "committed automatically")
     @click.option("--test", is_flag=True, default=False, help="Indicates if the created repository "
                   "is a test authentication repository")
-    def create(path, keys_description, keystore, no_commit, test):
+    @pin_managed
+    def create(path, keys_description, keystore, no_commit, test, pin_manager):
         create_repository(
             path=path,
+            pin_manager=pin_manager,
             keystore=keystore,
             roles_key_infos=keys_description,
             commit=not no_commit,

taf/tuf/repository.py

@@ -130,6 +130,7 @@ class MetadataRepository(Repository):
 
     def __init__(self, path: Union[Path, str], *args, **kwargs) -> None:
         storage_backend = kwargs.pop("storage", None)
+        pin_manager = kwargs.pop("pin_manager", None)
         super().__init__(*args, **kwargs)
         self.signer_cache: Dict[str, Dict[str, Signer]] = defaultdict(dict)
         self.path = Path(path)
@@ -141,6 +142,7 @@ def __init__(self, path: Union[Path, str], *args, **kwargs) -> None:
         else:
             self.storage_backend = FilesystemBackend()
         self._metadata_to_keep_open: Set[str] = set()
+        self.pin_manager = pin_manager
 
     @property
     def metadata_path(self) -> Path:

taf/yubikey/pin_manager.py

@@ -0,0 +1,36 @@
+import contextlib
+
+
+class PinManager():
+
+    def __init__(self):
+        self._pins = {}
+
+    def set_pin(self, serial_number, pin):
+        self._pins[serial_number] = pin
+
+    def get_pin(self, serial_number):
+        return self._pins.get(serial_number)
+
+    def clear_pins(self):
+        for key in list(self._pins.keys()):
+            self._pins[key] = None
+        self._pins.clear()
+
+
+
+@contextlib.contextmanager
+def manage_pins():
+    pin_manager = PinManager()
+    try:
+        yield pin_manager
+    finally:
+        pin_manager.clear_pins()
+
+
+def pin_managed(func):
+    def wrapper(*args, **kwargs):
+        with manage_pins() as pin_manager:
+            kwargs['pin_manager'] = pin_manager
+            return func(*args, **kwargs)
+    return wrapper