trust: Support operator field, support multiple services (#1407)

* 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>

* trust: zero is a valid number of TSAs for now

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

* trust: Support ServiceSelector fully

I don't think we'll be seeing anything else than ANY for a while
but for completeness, support all selector modes for TSA and Rekor.

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

* trust: Improve docstrings

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

* trust: Refactor based on review comments

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

* trust: Refactor service configuration

* Abstract the ServiceConfiguration handling as suggested in review
  (so both tsa and rekor are handled in the same way)
* This creates some issues as TSAs are still optional... I decided that
  it is reasonable to require "ANY" selector to be used with at least
  one service, meaning I have to change the placeholder signingconfig
  for production.

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

* CHANGELOG: Add entry for SigningConfig API changes

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

* trust: Fix multiple issues in _get_valid_services()

* we want a single service per operator
* selector UNDEFINED should be an error
* variable in this function should not refer to "logs" but "services"
* Method is actually static

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

* trust: Better docstrings

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

* tests: Add tests for service selection

Test the service selection with SigningConfig._get_valid_services()

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

* tests: Refactor based on review comments

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

---------

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

Author: jku

CHANGELOG.md

@@ -50,6 +50,9 @@ All versions prior to 0.9.0 are untracked.
   * ClientTrustConfig now provides methods `production()`, `staging()`and `from_tuf()`
     to get access to current client configuration (trusted keys & certificates,
     URLs and their validity periods). [#1363](https://github.com/sigstore/sigstore-python/pull/1363)
+  * SigningConfig now has methods that return actual clients (like `RekorClient`) instead of
+    just URLs. The returned clients are also filtered according to SigningConfig contents.
+    [#1407](https://github.com/sigstore/sigstore-python/pull/1407)
 * `--trust-config` now requires a file with SigningConfig v0.2, and is able to fully
   configure the used Sigstore instance [#1358]/(https://github.com/sigstore/sigstore-python/pull/1358)
 * By default (when `--trust-config` is not used) the whole trust configuration now

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
@@ -46,6 +47,7 @@
 )
 from sigstore_protobuf_specs.dev.sigstore.trustroot.v1 import (
     Service,
+    ServiceConfiguration,
     ServiceSelector,
     TransparencyLogInstance,
 )
@@ -56,6 +58,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 +71,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,28 +334,34 @@ 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:
             SigningConfig.SigningConfigType(self._inner.media_type)
         except ValueError:
             raise Error(f"unsupported signing config format: {self._inner.media_type}")
 
-        # currently not supporting other select modes
-        # TODO: Support other modes ensuring tsa_urls() and tlog_urls() work
-        if self._inner.rekor_tlog_config.selector != ServiceSelector.ANY:
-            raise Error(
-                f"unsupported tlog selector {self._inner.rekor_tlog_config.selector}"
-            )
-        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, selected by the service
+        # configuration & supported by this client
+        self._tlogs = self._get_valid_services(
+            self._inner.rekor_tlog_urls, REKOR_VERSIONS, self._inner.rekor_tlog_config
+        )
+        if not self._tlogs:
+            raise Error("No valid Rekor transparency log found in signing config")
+
+        self._tsas = self._get_valid_services(
+            self._inner.tsa_urls, TSA_VERSIONS, self._inner.tsa_config
+        )
+
+        self._fulcios = self._get_valid_services(
+            self._inner.ca_urls, FULCIO_VERSIONS, None
+        )
+        if not self._fulcios:
+            raise Error("No valid Fulcio CA found in signing config")
+
+        self._oidcs = self._get_valid_services(
+            self._inner.oidc_urls, OIDC_VERSIONS, None
+        )
 
     @classmethod
     def from_file(
@@ -356,54 +373,73 @@ def from_file(
         return cls(inner)
 
     @staticmethod
-    def _get_valid_service_url(services: list[Service]) -> str | None:
+    def _get_valid_services(
+        services: list[Service],
+        supported_versions: list[int],
+        config: ServiceConfiguration | None,
+    ) -> list[Service]:
+        """Return supported services, taking SigningConfig restrictions into account"""
+
+        # split services by operator, only include valid services
+        services_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 supported_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]:
+            services_by_operator[service.operator].append(service)
+
+        # build a list of services but make sure we only include one service per operator
+        # and use the highest version available for that operator
+        result: list[Service] = []
+        for op_services in services_by_operator.values():
+            op_services.sort(key=lambda s: s.major_api_version)
+            result.append(op_services[-1])
+
+        # Depending on ServiceSelector, prune the result list
+        if not config or config.selector == ServiceSelector.ALL:
+            return result
+
+        if config.selector == ServiceSelector.UNDEFINED:
+            raise ValueError("Undefined is not a valid signing config ServiceSelector")
+
+        # handle EXACT and ANY selectors
+        count = config.count if config.selector == ServiceSelector.EXACT else 1
+        if len(result) < count:
+            raise ValueError(
+                f"Expected {count} services in signing config, found {len(result)}"
+            )
+
+        return result[:count]
+
+    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
+        Returns the rekor transparency log clients to sign with.
         """
+        return [RekorClient(tlog.url) for tlog in self._tlogs]
 
-        url = self._get_valid_service_url(self._inner.rekor_tlog_urls)
-        if not url:
-            raise Error("No valid Rekor transparency log found in signing config")
-        return [url]
-
-    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
+        Returns a Fulcio client to get a signing certificate from
         """
-        url = self._get_valid_service_url(self._inner.ca_urls)
-        if not url:
-            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 clients for urls configured in signing config.
         """
-        url = self._get_valid_service_url(self._inner.tsa_urls)
-        return [] if url is None else [url]
+        return [TimestampAuthorityClient(s.url) for s in self._tsas]
 
 
 class TrustedRoot:

sigstore/_store/https%3A%2F%2Ftuf-repo-cdn.sigstore.dev/signing_config.v0.2.json

@@ -34,6 +34,6 @@
     "selector": "ANY"
   },
   "tsaConfig": {
-    "selector": "ANY"
+    "selector": "ALL"
   }
 }

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

@@ -20,7 +20,15 @@
 from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
 from cryptography.x509 import load_pem_x509_certificate
 from sigstore_protobuf_specs.dev.sigstore.common.v1 import TimeRange
+from sigstore_protobuf_specs.dev.sigstore.trustroot.v1 import (
+    Service,
+    ServiceConfiguration,
+    ServiceSelector,
+)
 
+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,
@@ -32,6 +40,19 @@
 from sigstore._utils import load_pem_public_key
 from sigstore.errors import Error
 
+# Test data for TestSigningcconfig
+_service_v1_op1 = Service("url1", major_api_version=1, operator="op1")
+_service2_v1_op1 = Service("url2", major_api_version=1, operator="op1")
+_service_v2_op1 = Service("url3", major_api_version=2, operator="op1")
+_service_v1_op2 = Service("url4", major_api_version=1, operator="op2")
+_service_v1_op3 = Service("url5", major_api_version=1, operator="op3")
+_service_v1_op4 = Service(
+    "url6",
+    major_api_version=1,
+    operator="op4",
+    valid_for=TimeRange(datetime(3000, 1, 1, tzinfo=timezone.utc)),
+)
+
 
 class TestCertificateAuthority:
     def test_good(self, asset):
@@ -56,12 +77,125 @@ 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"
+
+    @pytest.mark.parametrize(
+        "services, versions, config, expected_result",
+        [
+            pytest.param(
+                [_service_v1_op1],
+                [1],
+                ServiceConfiguration(ServiceSelector.ALL),
+                [_service_v1_op1],
+                id="base case",
+            ),
+            pytest.param(
+                [_service_v1_op1, _service2_v1_op1],
+                [1],
+                ServiceConfiguration(ServiceSelector.ALL),
+                [_service2_v1_op1],
+                id="multiple services, same operator: expect 1 service in result",
+            ),
+            pytest.param(
+                [_service_v1_op1, _service_v1_op2],
+                [1],
+                ServiceConfiguration(ServiceSelector.ALL),
+                [_service_v1_op1, _service_v1_op2],
+                id="2 services, different operator: expect 2 services in result",
+            ),
+            pytest.param(
+                [_service_v1_op1, _service_v1_op2, _service_v1_op4],
+                [1],
+                ServiceConfiguration(ServiceSelector.ALL),
+                [_service_v1_op1, _service_v1_op2],
+                id="3 services, one is not yet valid: expect 2 services in result",
+            ),
+            pytest.param(
+                [_service_v1_op1, _service_v1_op2],
+                [1],
+                ServiceConfiguration(ServiceSelector.ANY),
+                [_service_v1_op1],
+                id="ANY selector: expect 1 service only in result",
+            ),
+            pytest.param(
+                [_service_v1_op1, _service_v1_op2, _service_v1_op3],
+                [1],
+                ServiceConfiguration(ServiceSelector.EXACT, 2),
+                [_service_v1_op1, _service_v1_op2],
+                id="EXACT selector: expect configured number of services in result",
+            ),
+            pytest.param(
+                [_service_v1_op1, _service_v2_op1],
+                [1, 2],
+                ServiceConfiguration(ServiceSelector.ALL),
+                [_service_v2_op1],
+                id="services with different version: expect highest version",
+            ),
+            pytest.param(
+                [_service_v1_op1, _service_v2_op1],
+                [1],
+                ServiceConfiguration(ServiceSelector.ALL),
+                [_service_v1_op1],
+                id="services with different version: expect the supported version",
+            ),
+            pytest.param(
+                [_service_v1_op1, _service_v1_op2],
+                [2],
+                ServiceConfiguration(ServiceSelector.ALL),
+                [],
+                id="No supported versions: expect no results",
+            ),
+            pytest.param(
+                [_service_v1_op1, _service_v2_op1, _service_v1_op2],
+                [1],
+                None,
+                [_service_v1_op1, _service_v1_op2],
+                id="services without ServiceConfiguration: expect all supported",
+            ),
+        ],
+    )
+    def test_get_valid_services(self, services, versions, config, expected_result):
+        result = SigningConfig._get_valid_services(services, versions, config)
+
+        assert result == expected_result
+
+    @pytest.mark.parametrize(
+        "services, versions, config",
+        [
+            (  # ANY selector without services
+                [],
+                [1],
+                ServiceConfiguration(ServiceSelector.ANY),
+            ),
+            (  # EXACT selector without enough services
+                [_service_v1_op1],
+                [1],
+                ServiceConfiguration(ServiceSelector.EXACT, 2),
+            ),
+            (  # UNDEFINED selector
+                [_service_v1_op1],
+                [1],
+                ServiceConfiguration(ServiceSelector.UNDEFINED, 1),
+            ),
+        ],
+    )
+    def test_get_valid_services_fail(self, services, versions, config):
+        with pytest.raises(ValueError):
+            SigningConfig._get_valid_services(services, versions, config)
 
 
 class TestTrustedRoot: