Use TUF to download key/cert material

pyproject.toml

@@ -33,6 +33,7 @@ dependencies = [
   "pyOpenSSL >= 22.0.0",
   "requests",
   "securesystemslib",
+  "tuf >= 2.0.0",
 ]
 requires-python = ">=3.7"
 

sigstore/_cli.py

@@ -40,13 +40,9 @@
     RekorClient,
     RekorEntry,
 )
+from sigstore._internal.tuf import TrustUpdater
 from sigstore._sign import Signer
-from sigstore._utils import (
-    SplitCertificateChainError,
-    load_pem_public_key,
-    read_embedded,
-    split_certificate_chain,
-)
+from sigstore._utils import SplitCertificateChainError, split_certificate_chain
 from sigstore._verify import (
     CertificateVerificationFailure,
     RekorEntryMissing,
@@ -60,22 +56,6 @@
 logging.basicConfig(level=os.environ.get("SIGSTORE_LOGLEVEL", "INFO").upper())
 
 
-class _Embedded:
-    """
-    A repr-wrapper for reading embedded resources, needed to help `argparse`
-    render defaults correctly.
-    """
-
-    def __init__(self, name: str) -> None:
-        self._name = name
-
-    def read(self) -> bytes:
-        return read_embedded(self._name)
-
-    def __repr__(self) -> str:
-        return f"{self._name} (embedded)"
-
-
 def _boolify_env(envvar: str) -> bool:
     """
     An `argparse` helper for turning an environment variable into a boolean.
@@ -116,7 +96,7 @@ def _add_shared_instance_options(group: argparse._ArgumentGroup) -> None:
         metavar="FILE",
         type=argparse.FileType("rb"),
         help="A PEM-encoded root public key for Rekor itself (conflicts with --staging)",
-        default=os.getenv("SIGSTORE_REKOR_ROOT_PUBKEY", _Embedded("rekor.pub")),
+        default=os.getenv("SIGSTORE_REKOR_ROOT_PUBKEY"),
     )
 
 
@@ -238,7 +218,7 @@ def _parser() -> argparse.ArgumentParser:
         metavar="FILE",
         type=argparse.FileType("rb"),
         help="A PEM-encoded public key for the CT log (conflicts with --staging)",
-        default=os.getenv("SIGSTORE_CTFE", _Embedded("ctfe.pub")),
+        default=os.getenv("SIGSTORE_CTFE"),
     )
 
     sign.add_argument(
@@ -427,12 +407,21 @@ def _sign(args: argparse.Namespace) -> None:
     elif args.fulcio_url == DEFAULT_FULCIO_URL and args.rekor_url == DEFAULT_REKOR_URL:
         signer = Signer.production()
     else:
-        ct_keyring = CTKeyring([load_pem_public_key(args.ctfe_pem.read())])
+        # Assume "production" keys if none are given as arguments
+        updater = TrustUpdater.production()
+        if args.ctfe_pem is not None:
+            ctfe_keys = [args.ctfe_pem.read()]
+        else:
+            ctfe_keys = updater.get_ctfe_keys()
+        if args.rekor_root_pubkey is not None:
+            rekor_key = args.rekor_root_pubkey.read()
+        else:
+            rekor_key = updater.get_rekor_key()
+
+        ct_keyring = CTKeyring(ctfe_keys)
         signer = Signer(
             fulcio=FulcioClient(args.fulcio_url),
-            rekor=RekorClient(
-                args.rekor_url, args.rekor_root_pubkey.read(), ct_keyring
-            ),
+            rekor=RekorClient(args.rekor_url, rekor_key, ct_keyring),
         )
 
     # The order of precedence is as follows:
@@ -560,10 +549,16 @@ def _verify(args: argparse.Namespace) -> None:
         except SplitCertificateChainError as error:
             args._parser.error(f"Failed to parse certificate chain: {error}")
 
+        if args.rekor_root_pubkey is not None:
+            rekor_key = args.rekor_root_pubkey.read()
+        else:
+            updater = TrustUpdater.production()
+            rekor_key = updater.get_rekor_key()
+
         verifier = Verifier(
             rekor=RekorClient(
                 url=args.rekor_url,
-                pubkey=args.rekor_root_pubkey.read(),
+                pubkey=rekor_key,
                 # We don't use the CT keyring in verification so we can supply an empty keyring
                 ct_keyring=CTKeyring(),
             ),

sigstore/_internal/ctfe.py

@@ -25,12 +25,7 @@
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.asymmetric import ec, rsa
 
-from sigstore._utils import (
-    PublicKey,
-    key_id,
-    load_pem_public_key,
-    read_embedded,
-)
+from sigstore._utils import key_id, load_pem_public_key
 
 
 class CTKeyringError(Exception):
@@ -58,48 +53,16 @@ class CTKeyring:
     This structure exists to facilitate key rotation in a CT log.
     """
 
-    def __init__(self, keys: List[PublicKey] = []):
+    def __init__(self, keys: List[bytes] = []):
         """
         Create a new `CTKeyring`, with `keys` as the initial set of signing
         keys.
         """
         self._keyring = {}
-        for key in keys:
+        for key_bytes in keys:
+            key = load_pem_public_key(key_bytes)
             self._keyring[key_id(key)] = key
 
-    @classmethod
-    def staging(cls) -> CTKeyring:
-        """
-        Returns a `CTKeyring` instance capable of verifying SCTs from
-        Sigstore's staging deployment.
-        """
-        keyring = cls()
-        keyring._add_resource("ctfe.staging.pub")
-        keyring._add_resource("ctfe_2022.staging.pub")
-        keyring._add_resource("ctfe_2022.2.staging.pub")
-
-        return keyring
-
-    @classmethod
-    def production(cls) -> CTKeyring:
-        """
-        Returns a `CTKeyring` instance capable of verifying SCTs from
-        Sigstore's production deployment.
-        """
-        keyring = cls()
-        keyring._add_resource("ctfe.pub")
-        keyring._add_resource("ctfe_2022.pub")
-
-        return keyring
-
-    def _add_resource(self, name: str) -> None:
-        """
-        Adds a key to the current keyring, as identified by its
-        resource name under `sigstore._store`.
-        """
-        key_pem = read_embedded(name)
-        self.add(key_pem)
-
     def add(self, key_pem: bytes) -> None:
         """
         Adds a PEM-encoded key to the current keyring.

sigstore/_internal/rekor/client.py

@@ -33,19 +33,14 @@
 from securesystemslib.formats import encode_canonical
 
 from sigstore._internal.ctfe import CTKeyring
-from sigstore._utils import base64_encode_pem_cert, read_embedded
+from sigstore._internal.tuf import TrustUpdater
+from sigstore._utils import base64_encode_pem_cert
 
 logger = logging.getLogger(__name__)
 
 DEFAULT_REKOR_URL = "https://rekor.sigstore.dev"
 STAGING_REKOR_URL = "https://rekor.sigstage.dev"
 
-_DEFAULT_REKOR_ROOT_PUBKEY = read_embedded("rekor.pub")
-_STAGING_REKOR_ROOT_PUBKEY = read_embedded("rekor.staging.pub")
-
-_DEFAULT_REKOR_CTFE_PUBKEY = read_embedded("ctfe.pub")
-_STAGING_REKOR_CTFE_PUBKEY = read_embedded("ctfe.staging.pub")
-
 
 class RekorBundle(BaseModel):
     """
@@ -411,14 +406,10 @@ def __del__(self) -> None:
         self.session.close()
 
     @classmethod
-    def production(cls) -> RekorClient:
-        return cls(
-            DEFAULT_REKOR_URL, _DEFAULT_REKOR_ROOT_PUBKEY, CTKeyring.production()
-        )
-
-    @classmethod
-    def staging(cls) -> RekorClient:
-        return cls(STAGING_REKOR_URL, _STAGING_REKOR_ROOT_PUBKEY, CTKeyring.staging())
+    def with_updater(cls, updater: TrustUpdater) -> RekorClient:
+        rekor_key = updater.get_rekor_key()
+        ctfe_keys = updater.get_ctfe_keys()
+        return cls(DEFAULT_REKOR_URL, rekor_key, CTKeyring(ctfe_keys))
 
     @property
     def log(self) -> RekorLog:

sigstore/_internal/tuf.py

@@ -0,0 +1,150 @@
+# Copyright 2022 The Sigstore Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import shutil
+from importlib import resources
+from pathlib import Path
+from typing import List, Optional, Tuple
+from urllib import parse
+
+from tuf.ngclient import Updater
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_TUF_URL = "https://sigstore-tuf-root.storage.googleapis.com/"
+STAGING_TUF_URL = "https://tuf-root-staging.storage.googleapis.com/"
+
+
+def _get_dirs(url: str) -> Tuple[Path, Path]:
+    """Return metadata dir and target cache dir for URL"""
+    # NOTE: this is not great for windows: should maybe depend on appdirs?
+    # TODO: there should be URL normalization if URLs come from user
+    dir = parse.quote(url, safe="")
+    md_dir = Path.home() / ".local" / "share" / "sigstore-python" / "tuf" / dir
+    targets_dir = Path.home() / ".cache" / "sigstore-python" / "tuf" / dir
+    return md_dir, targets_dir
+
+
+class TrustUpdater:
+    def __init__(self, url: str) -> None:
+        self._repo_url = url
+        self._updater: Optional[Updater] = None
+
+        self._metadata_dir, self._targets_dir = _get_dirs(url)
+
+        # intialize metadata dir
+        tuf_root = self._metadata_dir / "root.json"
+        if not tuf_root.exists():
+            if self._repo_url == DEFAULT_TUF_URL:
+                fname = "root.json"
+            elif self._repo_url == STAGING_TUF_URL:
+                fname = "staging-root.json"
+            else:
+                raise Exception(f"TUF root not found in {tuf_root}")
+
+            self._metadata_dir.mkdir(parents=True, exist_ok=True)
+            with resources.path("sigstore._store", fname) as res:
+                shutil.copy2(res, tuf_root)
+
+        # intialize targets cache dir
+        # TODO: Pre-populate with any targets we ship with sources
+        self._targets_dir.mkdir(parents=True, exist_ok=True)
+
+        logger.debug("TUF metadata: %s", self._metadata_dir)
+        logger.debug("TUF targets cache: %s", self._targets_dir)
+
+    @classmethod
+    def production(cls) -> "TrustUpdater":
+        return cls(DEFAULT_TUF_URL)
+
+    @classmethod
+    def staging(cls) -> "TrustUpdater":
+        return cls(STAGING_TUF_URL)
+
+    def _setup(self) -> "Updater":
+        """Initialize and update the toplevel TUF metadata"""
+        updater = Updater(
+            metadata_dir=str(self._metadata_dir),
+            metadata_base_url=f"{self._repo_url}",
+            target_base_url=f"{self._repo_url}targets/",
+            target_dir=str(self._targets_dir),
+        )
+
+        # NOTE: we would like to avoid refresh if the toplevel metadata is valid.
+        # https://github.com/theupdateframework/python-tuf/issues/2225
+        updater.refresh()
+        return updater
+
+    def get_ctfe_keys(self) -> List[bytes]:
+        """Return the active CTFE public keys contents"""
+        if not self._updater:
+            self._updater = self._setup()
+
+        ctfes = []
+        assert self._updater._trusted_set.targets
+        targets = self._updater._trusted_set.targets.signed.targets
+        for target_info in targets.values():
+            custom = target_info.unrecognized_fields["custom"]["sigstore"]
+            if custom["status"] == "Active" and custom["usage"] == "CTFE":
+                path = self._updater.find_cached_target(target_info)
+                if path is None:
+                    path = self._updater.download_target(target_info)
+                with open(path, "rb") as f:
+                    ctfes.append(f.read())
+
+        if not ctfes:
+            raise Exception("CTFE keys not found in TUF metadata")
+
+        return ctfes
+
+    def get_rekor_key(self) -> bytes:
+        """Return the rekor public key content"""
+        if not self._updater:
+            self._updater = self._setup()
+
+        assert self._updater._trusted_set.targets
+        targets = self._updater._trusted_set.targets.signed.targets
+        for target, target_info in targets.items():
+            custom = target_info.unrecognized_fields["custom"]["sigstore"]
+            if custom["status"] == "Active" and custom["usage"] == "Rekor":
+                path = self._updater.find_cached_target(target_info)
+                if path is None:
+                    path = self._updater.download_target(target_info)
+                with open(path, "rb") as f:
+                    return f.read()
+
+        raise Exception("Rekor key not found in TUF metadata")
+
+    def get_fulcio_certs(self) -> List[bytes]:
+        """Return the active Fulcio certificate contents"""
+        if not self._updater:
+            self._updater = self._setup()
+
+        certs = []
+        assert self._updater._trusted_set.targets
+        targets = self._updater._trusted_set.targets.signed.targets
+        for target_info in targets.values():
+            custom = target_info.unrecognized_fields["custom"]["sigstore"]
+            if custom["status"] == "Active" and custom["usage"] == "Fulcio":
+                path = self._updater.find_cached_target(target_info)
+                if path is None:
+                    path = self._updater.download_target(target_info)
+                with open(path, "rb") as f:
+                    certs.append(f.read())
+
+        if not certs:
+            raise Exception("Fulcio certificates not found in TUF metadata")
+
+        return certs

sigstore/_sign.py

@@ -33,6 +33,7 @@
 from sigstore._internal.oidc import Identity
 from sigstore._internal.rekor import RekorClient, RekorEntry
 from sigstore._internal.sct import verify_sct
+from sigstore._internal.tuf import TrustUpdater
 from sigstore._utils import sha256_streaming
 
 logger = logging.getLogger(__name__)
@@ -61,14 +62,18 @@ def production(cls) -> Signer:
         """
         Return a `Signer` instance configured against Sigstore's production-level services.
         """
-        return cls(fulcio=FulcioClient.production(), rekor=RekorClient.production())
+        updater = TrustUpdater.production()
+        rekor = RekorClient.with_updater(updater)
+        return cls(fulcio=FulcioClient.production(), rekor=rekor)
 
     @classmethod
     def staging(cls) -> Signer:
         """
         Return a `Signer` instance configured against Sigstore's staging-level services.
         """
-        return cls(fulcio=FulcioClient.staging(), rekor=RekorClient.staging())
+        updater = TrustUpdater.staging()
+        rekor = RekorClient.with_updater(updater)
+        return cls(fulcio=FulcioClient.staging(), rekor=rekor)
 
     def sign(
         self,