tuf: use bundled trusted root if available (#542)

tnytown · woodruffw · web-flow · commit 7b390f4789d3 · 2023-03-23T19:22:21.000-04:00
* tuf: use "trusted root" if available

Signed-off-by: Andrew Pan &lt;a@tny.town&gt;

* tuf, test_tuf: adapt tests for bundled root

Signed-off-by: Andrew Pan &lt;a@tny.town&gt;

* keyring, _utils, test_utils: DER cert loading

Signed-off-by: Andrew Pan &lt;a@tny.town&gt;

* tuf: allow missing expiries in key validity period

Signed-off-by: Andrew Pan &lt;a@tny.town&gt;

* tuf, test_tuf: `_get` should never return `None`

Signed-off-by: Andrew Pan &lt;a@tny.town&gt;

* tuf: revert enums

Signed-off-by: Andrew Pan &lt;a@tny.town&gt;

* test_tuf: test_is_timerange_valid

Signed-off-by: Andrew Pan &lt;a@tny.town&gt;

* _utils: _errors -&gt; errors

Signed-off-by: Andrew Pan &lt;a@tny.town&gt;

* CHANGELOG: blurb

Signed-off-by: Andrew Pan &lt;a@tny.town&gt;

* keyring: handle all errors in except cases

Signed-off-by: Andrew Pan &lt;a@tny.town&gt;

* Apply suggestions from code review

Co-authored-by: William Woodruff &lt;william@yossarian.net&gt;
Signed-off-by: Andrew Pan &lt;3821575+tnytown@users.noreply.github.com&gt;

* Implement suggestions from code review

Signed-off-by: Andrew Pan &lt;a@tny.town&gt;

* Apply suggestions from code review

Co-authored-by: William Woodruff &lt;william@yossarian.net&gt;
Signed-off-by: Andrew Pan &lt;3821575+tnytown@users.noreply.github.com&gt;

---------

Signed-off-by: Andrew Pan &lt;a@tny.town&gt;
Signed-off-by: Andrew Pan &lt;3821575+tnytown@users.noreply.github.com&gt;
Co-authored-by: William Woodruff &lt;william@yossarian.net&gt;
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -27,6 +27,9 @@ All versions prior to 0.9.0 are untracked.
   ([#535](https://github.com/sigstore/sigstore-python/pull/535))
 * Revamped error diagnostics reporting. All errors with diagnostics now implement
   `sigstore.errors.Error`.
+* Trust root materials are now retrieved from a single trust bundle,
+  if it is available via TUF
+  ([#542](https://github.com/sigstore/sigstore-python/pull/542))
 * Improved diagnostics around Signed Certificate Timestamp verification failures.
   ([#555](https://github.com/sigstore/sigstore-python/pull/555))
 
diff --git a/sigstore/_internal/keyring.py b/sigstore/_internal/keyring.py
@@ -24,7 +24,14 @@
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.asymmetric import ec, rsa
 
-from sigstore._utils import KeyID, key_id, load_pem_public_key
+from sigstore._utils import (
+    InvalidKeyError,
+    KeyID,
+    UnexpectedKeyFormatError,
+    key_id,
+    load_der_public_key,
+    load_pem_public_key,
+)
 
 
 class KeyringError(Exception):
@@ -61,11 +68,19 @@ class Keyring:
     def __init__(self, keys: List[bytes] = []):
         """
         Create a new `Keyring`, with `keys` as the initial set of signing
-        keys.
+        keys. These `keys` can be in either DER format or PEM encoded.
         """
         self._keyring = {}
         for key_bytes in keys:
-            key = load_pem_public_key(key_bytes)
+            key = None
+
+            try:
+                key = load_pem_public_key(key_bytes)
+            except UnexpectedKeyFormatError as e:
+                raise e
+            except InvalidKeyError:
+                key = load_der_public_key(key_bytes)
+
             self._keyring[key_id(key)] = key
 
     def add(self, key_pem: bytes) -> None:
diff --git a/sigstore/_internal/tuf.py b/sigstore/_internal/tuf.py
@@ -20,12 +20,20 @@
 
 import logging
 import os
+from datetime import datetime, timezone
 from functools import lru_cache
 from pathlib import Path
+from typing import Optional
 from urllib import parse
 
 import appdirs
-from cryptography.x509 import Certificate, load_pem_x509_certificate
+from cryptography.x509 import (
+    Certificate,
+    load_der_x509_certificate,
+    load_pem_x509_certificate,
+)
+from sigstore_protobuf_specs.dev.sigstore.common.v1 import TimeRange
+from sigstore_protobuf_specs.dev.sigstore.trustroot.v1 import TrustedRoot
 from tuf.api import exceptions as TUFExceptions
 from tuf.ngclient import RequestsFetcher, Updater
 
@@ -66,6 +74,28 @@ def _get_dirs(url: str) -> tuple[Path, Path]:
     return (tuf_data_dir / repo_base), (tuf_cache_dir / repo_base)
 
 
+def _is_timerange_valid(period: TimeRange | None, *, allow_expired: bool) -> bool:
+    """
+    Given a `period`, checks that the the current time is not before `start`. If
+    `allow_expired` is `False`, also checks that the current time is not after
+    `end`.
+    """
+    now = datetime.now(timezone.utc)
+
+    # If there was no validity period specified, the key is always valid.
+    if not period:
+        return True
+
+    # Active: if the current time is before the starting period, we are not yet
+    # valid.
+    if now < period.start:
+        return False
+
+    # If we want Expired keys, the key is valid at this point. Otherwise, check
+    # that we are within range.
+    return allow_expired or (period.end is None or now <= period.end)
+
+
 class TrustUpdater:
     """Internal trust root (certificates and keys) downloader.
 
@@ -85,8 +115,6 @@ def __init__(self, url: str) -> None:
         roots, i.e. for the production or staging Sigstore TUF repos.
         """
         self._repo_url = url
-        self._updater: Updater | None = None
-
         self._metadata_dir, self._targets_dir = _get_dirs(url)
 
         # Initialize metadata dir
@@ -125,7 +153,8 @@ def staging(cls) -> TrustUpdater:
         """
         return cls(STAGING_TUF_URL)
 
-    def _setup(self) -> Updater:
+    @lru_cache()
+    def _updater(self) -> Updater:
         """Initialize and update the toplevel TUF metadata"""
         updater = Updater(
             metadata_dir=str(self._metadata_dir),
@@ -144,25 +173,39 @@ def _setup(self) -> Updater:
 
         return updater
 
+    @lru_cache()
+    def _get_trusted_root(self) -> Optional[TrustedRoot]:
+        root_info = self._updater().get_targetinfo("trusted_root.json")
+        if root_info is None:
+            return None
+        path = self._updater().find_cached_target(root_info)
+        if path is None:
+            try:
+                path = self._updater().download_target(root_info)
+            except (
+                TUFExceptions.DownloadError,
+                TUFExceptions.RepositoryError,
+            ) as e:
+                raise TUFError("Failed to download trusted key bundle") from e
+
+        return TrustedRoot().from_json(Path(path).read_bytes())
+
     def _get(self, usage: str, statuses: list[str]) -> list[bytes]:
         """Return all targets with given usage and any of the statuses"""
-        if not self._updater:
-            self._updater = self._setup()
-
         data = []
 
-        targets = self._updater._trusted_set.targets.signed.targets
+        targets = self._updater()._trusted_set.targets.signed.targets
         for target_info in targets.values():
             custom = target_info.unrecognized_fields.get("custom", {}).get("sigstore")
             if (
                 custom
                 and custom.get("status") in statuses
                 and custom.get("usage") == usage
             ):
-                path = self._updater.find_cached_target(target_info)
+                path = self._updater().find_cached_target(target_info)
                 if path is None:
                     try:
-                        path = self._updater.download_target(target_info)
+                        path = self._updater().download_target(target_info)
                     except (
                         TUFExceptions.DownloadError,
                         TUFExceptions.RepositoryError,
@@ -187,7 +230,21 @@ def get_ctfe_keys(self) -> list[bytes]:
 
         May download files from the remote repository.
         """
-        ctfes = self._get("CTFE", ["Active"])
+        ctfes = []
+
+        trusted_root = self._get_trusted_root()
+        if trusted_root:
+            for key in trusted_root.ctlogs:
+                if not _is_timerange_valid(
+                    key.public_key.valid_for, allow_expired=False
+                ):
+                    continue
+                key_bytes = key.public_key.raw_bytes
+                if key_bytes:
+                    ctfes.append(key_bytes)
+        else:
+            ctfes = self._get("CTFE", ["Active"])
+
         if not ctfes:
             raise MetadataError("CTFE keys not found in TUF metadata")
         return ctfes
@@ -197,7 +254,21 @@ def get_rekor_keys(self) -> list[bytes]:
 
         May download files from the remote repository.
         """
-        keys = self._get("Rekor", ["Active"])
+        keys = []
+
+        trusted_root = self._get_trusted_root()
+        if trusted_root:
+            for key in trusted_root.tlogs:
+                if not _is_timerange_valid(
+                    key.public_key.valid_for, allow_expired=False
+                ):
+                    continue
+                key_bytes = key.public_key.raw_bytes
+                if key_bytes:
+                    keys.append(key_bytes)
+        else:
+            keys = self._get("Rekor", ["Active"])
+
         if len(keys) != 1:
             raise MetadataError("Did not find one active Rekor key in TUF metadata")
         return keys
@@ -207,9 +278,29 @@ def get_fulcio_certs(self) -> list[Certificate]:
 
         May download files from the remote repository.
         """
+        certs = []
+
+        trusted_root = self._get_trusted_root()
         # Return expired certificates too: they are expired now but may have
         # been active when the certificate was used to sign.
-        certs = self._get("Fulcio", ["Active", "Expired"])
+        if trusted_root:
+            for ca in trusted_root.certificate_authorities:
+                if not _is_timerange_valid(ca.valid_for, allow_expired=True):
+                    continue
+                certs.extend(
+                    [
+                        load_der_x509_certificate(cert.raw_bytes)
+                        for cert in ca.cert_chain.certificates
+                    ]
+                )
+        else:
+            certs = [
+                load_pem_x509_certificate(c)
+                for c in self._get(
+                    "Fulcio",
+                    ["Active", "Expired"],
+                )
+            ]
         if not certs:
             raise MetadataError("Fulcio certificates not found in TUF metadata")
-        return [load_pem_x509_certificate(c) for c in certs]
+        return certs
diff --git a/sigstore/_utils.py b/sigstore/_utils.py
@@ -27,6 +27,8 @@
 from cryptography.hazmat.primitives.asymmetric import ec, rsa
 from cryptography.x509 import Certificate
 
+from sigstore.errors import Error
+
 if sys.version_info < (3, 11):
     import importlib_resources as resources
 else:
@@ -57,28 +59,52 @@
 """
 
 
-class InvalidKey(Exception):
+class InvalidKeyError(Error):
     """
     Raised when loading a key fails.
     """
 
     pass
 
 
+class UnexpectedKeyFormatError(InvalidKeyError):
+    """
+    Raised when loading a key produces a key of an unexpected type.
+    """
+
+    pass
+
+
 def load_pem_public_key(key_pem: bytes) -> PublicKey:
     """
     A specialization of `cryptography`'s `serialization.load_pem_public_key`
-    with a uniform exception type (`InvalidKey`) and additional restrictions
+    with a uniform exception type (`InvalidKeyError`) and additional restrictions
     on key validity (only RSA and ECDSA keys are valid).
     """
 
     try:
         key = serialization.load_pem_public_key(key_pem)
     except Exception as exc:
-        raise InvalidKey("could not load PEM-formatted public key") from exc
+        raise InvalidKeyError("could not load PEM-formatted public key") from exc
+
+    if not isinstance(key, (rsa.RSAPublicKey, ec.EllipticCurvePublicKey)):
+        raise UnexpectedKeyFormatError(f"invalid key format (not ECDSA or RSA): {key}")
+
+    return key
+
+
+def load_der_public_key(key_der: bytes) -> PublicKey:
+    """
+    The `load_pem_public_key` specialization, but DER.
+    """
+
+    try:
+        key = serialization.load_der_public_key(key_der)
+    except Exception as exc:
+        raise InvalidKeyError("could not load DER-formatted public key") from exc
 
     if not isinstance(key, (rsa.RSAPublicKey, ec.EllipticCurvePublicKey)):
-        raise InvalidKey(f"invalid key format (not ECDSA or RSA): {key}")
+        raise UnexpectedKeyFormatError(f"invalid key format (not ECDSA or RSA): {key}")
 
     return key
 
diff --git a/test/unit/internal/test_tuf.py b/test/unit/internal/test_tuf.py
@@ -14,10 +14,12 @@
 
 
 import os
+from datetime import datetime, timedelta, timezone
 
 import pytest
+from sigstore_protobuf_specs.dev.sigstore.common.v1 import TimeRange
 
-from sigstore._internal.tuf import TrustUpdater
+from sigstore._internal.tuf import TrustUpdater, _is_timerange_valid
 
 
 def test_updater_staging_caches_and_requests(mock_staging_tuf, tuf_dirs):
@@ -82,6 +84,35 @@ def test_updater_staging_caches_and_requests(mock_staging_tuf, tuf_dirs):
     assert fail_reqs == expected_fail_reqs
 
 
+def test_is_timerange_valid():
+    def range_from(offset_lower=0, offset_upper=0):
+        base = datetime.now(timezone.utc)
+        return TimeRange(
+            base + timedelta(minutes=offset_lower),
+            base + timedelta(minutes=offset_upper),
+        )
+
+    # Test None should always be valid
+    assert _is_timerange_valid(None, allow_expired=False)
+    assert _is_timerange_valid(None, allow_expired=True)
+
+    # Test lower bound conditions
+    assert _is_timerange_valid(
+        range_from(-1, 1), allow_expired=False
+    )  # Valid: 1 ago, 1 from now
+    assert not _is_timerange_valid(
+        range_from(1, 1), allow_expired=False
+    )  # Invalid: 1 from now, 1 from now
+
+    # Test upper bound conditions
+    assert not _is_timerange_valid(
+        range_from(-1, -1), allow_expired=False
+    )  # Invalid: 1 ago, 1 ago
+    assert _is_timerange_valid(
+        range_from(-1, -1), allow_expired=True
+    )  # Valid: 1 ago, 1 ago
+
+
 def test_updater_staging_get(mock_staging_tuf, tuf_asset):
     """Test that one of the get-methods returns the expected content"""
     updater = TrustUpdater.staging()
@@ -98,6 +129,7 @@ def test_updater_ctfe_keys_error(monkeypatch):
     updater = TrustUpdater.staging()
     # getter returns no keys.
     monkeypatch.setattr(updater, "_get", lambda usage, statuses: [])
+    monkeypatch.setattr(updater, "_get_trusted_root", lambda: None)
     with pytest.raises(Exception, match="CTFE keys not found in TUF metadata"):
         updater.get_ctfe_keys()
 
@@ -112,6 +144,7 @@ def test_updater_rekor_keys_error(tuf_asset, monkeypatch):
             "_get",
             lambda usage, statuses: [rekor_key, rekor_key],
         )
+    monkeypatch.setattr(updater, "_get_trusted_root", lambda: None)
 
     with pytest.raises(
         Exception, match="Did not find one active Rekor key in TUF metadata"
@@ -122,7 +155,8 @@ def test_updater_rekor_keys_error(tuf_asset, monkeypatch):
 def test_updater_fulcio_certs_error(tuf_asset, monkeypatch):
     updater = TrustUpdater.staging()
     # getter returns no fulcio certs.
-    monkeypatch.setattr(updater, "_get", lambda usage, statuses: None)
+    monkeypatch.setattr(updater, "_get", lambda usage, statuses: [])
+    monkeypatch.setattr(updater, "_get_trusted_root", lambda: None)
     with pytest.raises(
         Exception, match="Fulcio certificates not found in TUF metadata"
     ):
diff --git a/test/unit/test_utils.py b/test/unit/test_utils.py
