trust: Support operator field, support multiple services

* Change the SigningConfig API so that it returns actual clients
  (like RekorClient): this may seem unusual but makes sense because
  SigningConfig must know which services this client supports, this
  means the code is simplest if it directly returns correct clients
  (e.g. when rekor has two separate clients for v1 and v2). This allows
  keeping version, operator, etc as SigningConfig imlementation details
* There is one exception to previous point: get_oidc_url() returns
  string, not Issuer object. I could make this change as well to be
  consistent, it just requires a small refactor (because currently
  Issuer makes a http request on construction and that seems bad,
  especially for testing)
* Support operator field: This is used to ensure we only return one
  service version per operator

Signed-off-by: Jussi Kukkonen <jkukkonen@google.com>

Author: jku

sigstore/_internal/trust.py

@@ -18,6 +18,7 @@
 
 from __future__ import annotations
 
+from collections import defaultdict
 from collections.abc import Iterable
 from dataclasses import dataclass
 from datetime import datetime, timezone
@@ -56,6 +57,9 @@
     TrustedRoot as _TrustedRoot,
 )
 
+from sigstore._internal.fulcio.client import FulcioClient
+from sigstore._internal.rekor.client import RekorClient
+from sigstore._internal.timestamp import TimestampAuthorityClient
 from sigstore._internal.tuf import DEFAULT_TUF_URL, STAGING_TUF_URL, TrustUpdater
 from sigstore._utils import (
     KeyID,
@@ -66,6 +70,12 @@
 )
 from sigstore.errors import Error, MetadataError, TUFError, VerificationError
 
+# Versions supported by this client
+REKOR_VERSIONS = [1]
+TSA_VERSIONS = [1]
+FULCIO_VERSIONS = [1]
+OIDC_VERSIONS = [1]
+
 
 def _is_timerange_valid(period: TimeRange | None, *, allow_expired: bool) -> bool:
     """
@@ -323,13 +333,6 @@ def __init__(self, inner: _SigningConfig):
         @api private
         """
         self._inner = inner
-        self._verify()
-
-    def _verify(self) -> None:
-        """
-        Performs various feats of heroism to ensure that the signing config
-        is well-formed.
-        """
 
         # must have a recognized media type.
         try:
@@ -346,6 +349,14 @@ def _verify(self) -> None:
         if self._inner.tsa_config.selector != ServiceSelector.ANY:
             raise Error(f"unsupported TSA selector {self._inner.tsa_config.selector}")
 
+        # Create lists of service protos that are valid & supported by this client
+        self._tlogs = self._get_valid_services(
+            self._inner.rekor_tlog_urls, REKOR_VERSIONS
+        )
+        self._tsas = self._get_valid_services(self._inner.tsa_urls, TSA_VERSIONS)
+        self._fulcios = self._get_valid_services(self._inner.ca_urls, FULCIO_VERSIONS)
+        self._oidcs = self._get_valid_services(self._inner.oidc_urls, OIDC_VERSIONS)
+
     @classmethod
     def from_file(
         cls,
@@ -355,55 +366,67 @@ def from_file(
         inner = _SigningConfig().from_json(Path(path).read_bytes())
         return cls(inner)
 
-    @staticmethod
-    def _get_valid_service_url(services: list[Service]) -> str | None:
+    def _get_valid_services(
+        self, services: list[Service], valid_versions: list[int]
+    ) -> list[Service]:
+        """Return supported services, taking SigningConfig restrictions into account"""
+
+        # split logs by operator, only include valid services
+        logs_by_operator: dict[str, list[Service]] = defaultdict(list)
         for service in services:
-            if service.major_api_version != 1:
+            if service.major_api_version not in valid_versions:
                 continue
 
             if not _is_timerange_valid(service.valid_for, allow_expired=False):
                 continue
-            return service.url
-        return None
 
-    def get_tlog_urls(self) -> list[str]:
+            logs_by_operator[service.operator].append(service)
+
+        # return a list of services but make sure we only return logs of one version per operator
+        result: list[Service] = []
+        for logs in logs_by_operator.values():
+            logs.sort(key=lambda s: s.major_api_version)
+            max_version = logs[-1].major_api_version
+
+            while logs and logs[-1].major_api_version == max_version:
+                result.append(logs.pop())
+
+        return result
+
+    def get_tlogs(self) -> list[RekorClient]:
         """
         Returns the rekor transparency logs that client should sign with.
-        Currently only returns a single one but could in future return several
         """
 
-        url = self._get_valid_service_url(self._inner.rekor_tlog_urls)
-        if not url:
+        if not self._tlogs:
             raise Error("No valid Rekor transparency log found in signing config")
-        return [url]
+        return [RekorClient(tlog.url) for tlog in self._tlogs]
 
-    def get_fulcio_url(self) -> str:
+    def get_fulcio(self) -> FulcioClient:
         """
         Returns url for the fulcio instance that client should use to get a
         signing certificate from
         """
-        url = self._get_valid_service_url(self._inner.ca_urls)
-        if not url:
+        if not self._fulcios:
             raise Error("No valid Fulcio CA found in signing config")
-        return url
+        return FulcioClient(self._fulcios[0].url)
 
     def get_oidc_url(self) -> str:
         """
         Returns url for the OIDC provider that client should use to interactively
         authenticate.
         """
-        url = self._get_valid_service_url(self._inner.oidc_urls)
-        if not url:
+        if not self._oidcs:
             raise Error("No valid OIDC provider found in signing config")
-        return url
+        return self._oidcs[0].url
 
-    def get_tsa_urls(self) -> list[str]:
+    def get_tsas(self) -> list[TimestampAuthorityClient]:
         """
-        Returns timestamp authority API end points. Currently returns a single one
-        but may return more in future.
+        Returns timestamp authority API end points.
         """
-        url = self._get_valid_service_url(self._inner.tsa_urls)
-        return [] if url is None else [url]
+        if not self._tsas:
+            raise Error("No valid Rekor transparency log found in signing config")
+        return [TimestampAuthorityClient(s.url) for s in self._tsas]
 
 
 class TrustedRoot:

sigstore/sign.py

@@ -332,12 +332,10 @@ def from_trust_config(cls, trust_config: ClientTrustConfig) -> SigningContext:
         """
         signing_config = trust_config.signing_config
         return cls(
-            fulcio=FulcioClient(signing_config.get_fulcio_url()),
-            rekor=RekorClient(signing_config.get_tlog_urls()[0]),
+            fulcio=signing_config.get_fulcio(),
+            rekor=signing_config.get_tlogs()[0],
             trusted_root=trust_config.trusted_root,
-            tsa_clients=[
-                TimestampAuthorityClient(url) for url in signing_config.get_tsa_urls()
-            ],
+            tsa_clients=signing_config.get_tsas(),
         )
 
     @contextmanager

test/unit/internal/test_trust.py

@@ -21,6 +21,9 @@
 from cryptography.x509 import load_pem_x509_certificate
 from sigstore_protobuf_specs.dev.sigstore.common.v1 import TimeRange
 
+from sigstore._internal.fulcio.client import FulcioClient
+from sigstore._internal.rekor.client import RekorClient
+from sigstore._internal.timestamp import TimestampAuthorityClient
 from sigstore._internal.trust import (
     CertificateAuthority,
     ClientTrustConfig,
@@ -56,12 +59,21 @@ def test_good(self, asset):
             signing_config._inner.media_type
             == SigningConfig.SigningConfigType.SIGNING_CONFIG_0_2.value
         )
-        assert signing_config.get_fulcio_url() == "https://fulcio.example.com"
+
+        fulcio = signing_config.get_fulcio()
+        assert isinstance(fulcio, FulcioClient)
+        assert fulcio.url == "https://fulcio.example.com"
         assert signing_config.get_oidc_url() == "https://oauth2.example.com/auth"
-        assert signing_config.get_tlog_urls() == ["https://rekor.example.com"]
-        assert signing_config.get_tsa_urls() == [
-            "https://timestamp.example.com/api/v1/timestamp"
-        ]
+
+        tlogs = signing_config.get_tlogs()
+        assert len(tlogs) == 1
+        assert isinstance(tlogs[0], RekorClient)
+        assert tlogs[0].url == "https://rekor.example.com/api/v1"
+
+        tsas = signing_config.get_tsas()
+        assert len(tsas) == 1
+        assert isinstance(tsas[0], TimestampAuthorityClient)
+        assert tsas[0].url == "https://timestamp.example.com/api/v1/timestamp"
 
 
 class TestTrustedRoot: