Add priv key uri support to SignerStore

repository_service_tuf_worker/signer.py

@@ -3,10 +3,21 @@
 # SPDX-License-Identifier: MIT
 
 from dynaconf import Dynaconf
-from securesystemslib.signer import Key, Signer
+from securesystemslib.signer import (
+    SIGNER_FOR_URI_SCHEME,
+    CryptoSigner,
+    Key,
+    Signer,
+)
 
 from repository_service_tuf_worker.interfaces import IKeyVault
 
+RSTUF_ONLINE_KEY_URI_FIELD = "x-rstuf-online-key-uri"
+
+# Register non-default securesystemslib file signer
+# secure-systems-lab/securesystemslib#617
+SIGNER_FOR_URI_SCHEME[CryptoSigner.FILE_URI_SCHEME] = CryptoSigner
+
 
 class SignerStore:
     """Generic signer store."""
@@ -16,15 +27,25 @@ def __init__(self, settings: Dynaconf):
         self._signers: dict[str, Signer] = {}
 
     def get(self, key: Key) -> Signer:
-        """Return signer for passed key."""
+        """Return signer for passed key.
+
+        - signer is loaded from the uri included in the passed public key
+          (see SIGNER_FOR_URI_SCHEME for available uri schemes)
+        - RSTUF_KEYVAULT_BACKEND is used as fallback, if no URI is included
+
+        """
 
         if key.keyid not in self._signers:
-            vault = self._settings.get("KEYVAULT")
-            if not isinstance(vault, IKeyVault):
-                raise ValueError(
-                    "RSTUF_KEYVAULT_BACKEND is required for online signing"
-                )
+            if uri := key.unrecognized_fields.get(RSTUF_ONLINE_KEY_URI_FIELD):
+                self._signers[key.keyid] = Signer.from_priv_key_uri(uri, key)
+
+            else:
+                vault = self._settings.get("KEYVAULT")
+                if not isinstance(vault, IKeyVault):
+                    raise ValueError(
+                        "RSTUF_KEYVAULT_BACKEND is required for online signing"
+                    )
 
-            self._signers[key.keyid] = vault.get(key)
+                self._signers[key.keyid] = vault.get(key)
 
         return self._signers[key.keyid]

tests/files/pem/ed25519_private.pem

@@ -0,0 +1,3 @@
+-----BEGIN PRIVATE KEY-----
+MC4CAQAwBQYDK2VwBCIEIGiI3w9x2HZ9UKGi51USN5JN2wtppaYVCRIBTp8ESaj3
+-----END PRIVATE KEY-----

tests/unit/tuf_repository_service_worker/test_signer.py

@@ -2,11 +2,19 @@
 #
 # SPDX-License-Identifier: MIT
 
+from pathlib import Path
+
 import pytest
 from pretend import stub
+from securesystemslib.signer import CryptoSigner, Key
 
 from repository_service_tuf_worker.interfaces import IKeyVault
-from repository_service_tuf_worker.signer import SignerStore
+from repository_service_tuf_worker.signer import (
+    RSTUF_ONLINE_KEY_URI_FIELD,
+    SignerStore,
+)
+
+_FILES = Path(__file__).parent.parent.parent / "files"
 
 
 class TestSigner:
@@ -36,7 +44,7 @@ def get(self, public_key):
 
         fake_id = "fake_id"
         fake_signer = stub()
-        fake_key = stub(keyid=fake_id)
+        fake_key = stub(keyid=fake_id, unrecognized_fields={})
         fake_settings = stub(get=lambda x: FakeKeyVault())
 
         store = SignerStore(fake_settings)
@@ -47,10 +55,35 @@ def get(self, public_key):
 
     def test_get_no_vault(self):
         fake_id = "fake_id"
-        fake_key = stub(keyid=fake_id)
+        fake_key = stub(keyid=fake_id, unrecognized_fields={})
         fake_settings = stub(get=lambda x: None)
 
         store = SignerStore(fake_settings)
 
         with pytest.raises(ValueError):
             store.get(fake_key)
+
+    def test_get_from_file_uri(self):
+        path = _FILES / "pem" / "ed25519_private.pem"
+        uri = f"file:{path}?encrypted=false"
+
+        key_metadata = {
+            "keytype": "ed25519",
+            "scheme": "ed25519",
+            "keyval": {
+                "public": (
+                    "4f66dabebcf30628963786001984c0b7"
+                    "5c175cdcf3bc4855933a2628f0cd0a0f"
+                )
+            },
+            RSTUF_ONLINE_KEY_URI_FIELD: uri,
+        }
+
+        fake_id = "fake_id"
+        key = Key.from_dict(fake_id, key_metadata)
+
+        fake_settings = stub()
+        store = SignerStore(fake_settings)
+        signer = store.get(key)
+
+        assert isinstance(signer, CryptoSigner)