From c989c74c25f95b7e61b617029f693f8fe8babea4 Mon Sep 17 00:00:00 2001 From: Michele Angrisano Date: Thu, 13 Aug 2026 12:16:25 +0200 Subject: [PATCH] feat: warn (and optionally fail) on a stale trust root with --offline Adds staleness checks in offline mode based on the cached TUF timestamp metadata expiry, configurable via --offline-staleness-warn / --offline-staleness-error (and the SIGSTORE_OFFLINE_STALENESS_* environment variables). Closes #1175 Signed-off-by: Michele Angrisano --- CHANGELOG.md | 9 ++++ docs/advanced/offline.md | 27 +++++++++++ sigstore/_cli.py | 79 ++++++++++++++++++++++++++++++-- sigstore/_internal/tuf.py | 50 +++++++++++++++++++- sigstore/models.py | 37 +++++++++++++-- test/unit/internal/test_trust.py | 52 +++++++++++++++++++++ 6 files changed, 246 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e6baab377..60b440cc4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,15 @@ All versions prior to 0.9.0 are untracked. ## [Unreleased] +### Added + +* `sigstore verify` now warns (and can fail) when used with `--offline` and a + stale cached trust root. The thresholds are configurable via + `--offline-staleness-warn` / `--offline-staleness-error` (or the + `SIGSTORE_OFFLINE_STALENESS_WARN` / `SIGSTORE_OFFLINE_STALENESS_ERROR` + environment variables); pass `off` to disable either level. + ([#1175](https://github.com/sigstore/sigstore-python/issues/1175)) + ## [4.5.0] ### Fixed diff --git a/docs/advanced/offline.md b/docs/advanced/offline.md index fb3f3e64b..e81d6b6c0 100644 --- a/docs/advanced/offline.md +++ b/docs/advanced/offline.md @@ -32,6 +32,33 @@ $ sigstore verify identity foo.txt \ --cert-oidc-issuer 'https://github.com/login/oauth' ``` +## Detecting a stale trust root + +Because `--offline` skips the TUF update, the cached trust root can silently +become stale. To surface this, `sigstore-python` inspects the cached TUF +timestamp metadata and: + +* emits a **warning** once the trust root has been expired for longer than + 24 hours; +* fails with an **error** once it has been expired for longer than 7 days. + +Both thresholds are configurable, either per-invocation or via the +`SIGSTORE_OFFLINE_STALENESS_WARN` / `SIGSTORE_OFFLINE_STALENESS_ERROR` +environment variables: + +```bash +$ sigstore verify identity foo.txt \ + --offline \ + --offline-staleness-warn 12h \ + --offline-staleness-error 30d \ + --cert-identity 'hamilcar@example.com' \ + --cert-oidc-issuer 'https://github.com/login/oauth' +``` + +Durations are written as ``, where `unit` is one of `s`, `m`, +`h`, `d`, or `w` (e.g. `24h`, `7d`, `2w`). Pass `off` (or `0`) to disable a +level entirely. + Alternatively, users may choose to bypass TUF entirely by passing an entire trust configuration to `sigstore-python` via `--trust-config`: diff --git a/sigstore/_cli.py b/sigstore/_cli.py index 7573b6436..10c6cb0ea 100644 --- a/sigstore/_cli.py +++ b/sigstore/_cli.py @@ -22,6 +22,7 @@ import sys from concurrent import futures from dataclasses import dataclass +from datetime import timedelta from pathlib import Path from typing import Any, NoReturn, TypeAlias, Union @@ -37,6 +38,10 @@ from sigstore._internal.fulcio.client import ExpiredCertificate from sigstore._internal.rekor import _hashedrekord_from_parts from sigstore._internal.rekor.client import RekorClient +from sigstore._internal.tuf import ( + DEFAULT_OFFLINE_STALENESS_ERROR, + DEFAULT_OFFLINE_STALENESS_WARN, +) from sigstore._utils import sha256_digest from sigstore.dsse import StatementBuilder, Subject from sigstore.dsse._predicate import ( @@ -114,6 +119,37 @@ def _invalid_arguments(args: argparse.Namespace, message: str) -> NoReturn: raise ValueError("unreachable") +def _parse_duration(value: str) -> timedelta | None: + """Parse a duration like '24h' or '7d' into a timedelta. + + '0', 'off', or 'none' disables the check (returns None). + Units: s, m, h, d, w. + """ + v = value.strip().lower() + if v in {"0", "off", "none"}: + return None + units = {"s": "seconds", "m": "minutes", "h": "hours", "d": "days", "w": "weeks"} + unit = v[-1:] + try: + amount = int(v[:-1]) + except ValueError: + raise argparse.ArgumentTypeError( + f"invalid duration {value!r} (e.g. '24h', '7d')" + ) from None + if unit not in units or amount < 0: + raise argparse.ArgumentTypeError( + f"invalid duration {value!r} (use s/m/h/d/w, e.g. '24h')" + ) + return timedelta(**{units[unit]: amount}) + + +def _staleness_env(envvar: str, default: timedelta | None) -> timedelta | None: + val = os.getenv(envvar) + if val is None: + return default + return _parse_duration(val) + + def _boolify_env(envvar: str) -> bool: """ An `argparse` helper for turning an environment variable into a boolean. @@ -193,6 +229,24 @@ def _add_shared_verification_options(group: argparse._ArgumentGroup) -> None: default=_boolify_env("SIGSTORE_OFFLINE"), help="Perform offline verification; requires a Sigstore bundle", ) + group.add_argument( + "--offline-staleness-warn", + metavar="DURATION", + type=_parse_duration, + default=_staleness_env( + "SIGSTORE_OFFLINE_STALENESS_WARN", DEFAULT_OFFLINE_STALENESS_WARN + ), + help="With --offline, warn if the cached trust root is older than this (e.g. 24h, 7d; 'off' to disable)", + ) + group.add_argument( + "--offline-staleness-error", + metavar="DURATION", + type=_parse_duration, + default=_staleness_env( + "SIGSTORE_OFFLINE_STALENESS_ERROR", DEFAULT_OFFLINE_STALENESS_ERROR + ), + help="With --offline, fail if the cached trust root is older than this (e.g. 7d; 'off' to disable)", + ) def _add_shared_oidc_options( @@ -1260,15 +1314,34 @@ def _get_trust_config(args: argparse.Namespace) -> ClientTrustConfig: """ # Not all commands provide --offline offline = getattr(args, "offline", False) + staleness_warn = getattr( + args, "offline_staleness_warn", DEFAULT_OFFLINE_STALENESS_WARN + ) + staleness_error = getattr( + args, "offline_staleness_error", DEFAULT_OFFLINE_STALENESS_ERROR + ) if args.trust_config: trust_config = ClientTrustConfig.from_json(args.trust_config.read_text()) elif args.instance: - trust_config = ClientTrustConfig.from_tuf(args.instance, offline=offline) + trust_config = ClientTrustConfig.from_tuf( + args.instance, + offline=offline, + staleness_warn=staleness_warn, + staleness_error=staleness_error, + ) elif args.staging: - trust_config = ClientTrustConfig.staging(offline=offline) + trust_config = ClientTrustConfig.staging( + offline=offline, + staleness_warn=staleness_warn, + staleness_error=staleness_error, + ) else: - trust_config = ClientTrustConfig.production(offline=offline) + trust_config = ClientTrustConfig.production( + offline=offline, + staleness_warn=staleness_warn, + staleness_error=staleness_error, + ) # Enforce rekor version if --rekor-version is used trust_config.force_tlog_version = getattr(args, "rekor_version", None) diff --git a/sigstore/_internal/tuf.py b/sigstore/_internal/tuf.py index 3882b9291..b89e67020 100644 --- a/sigstore/_internal/tuf.py +++ b/sigstore/_internal/tuf.py @@ -19,12 +19,14 @@ from __future__ import annotations import logging +from datetime import datetime, timedelta, timezone from functools import lru_cache from pathlib import Path from urllib import parse import platformdirs from tuf.api import exceptions as TUFExceptions +from tuf.api.metadata import Metadata from tuf.ngclient import Updater, UpdaterConfig # type: ignore[attr-defined] from sigstore import __version__ @@ -35,6 +37,8 @@ DEFAULT_TUF_URL = "https://tuf-repo-cdn.sigstore.dev" STAGING_TUF_URL = "https://tuf-repo-cdn.sigstage.dev" +DEFAULT_OFFLINE_STALENESS_WARN = timedelta(hours=24) +DEFAULT_OFFLINE_STALENESS_ERROR = timedelta(days=7) def _get_dirs(url: str) -> tuple[Path, Path]: @@ -64,10 +68,18 @@ class TrustUpdater: TrustUpdater expects to find an initial root.json in either the local metadata directory for this URL, or (as special case for the sigstore.dev production and staging instances) in the application resources. + staleness_warn and staleness_error are used to determine how long the local + metadata can be used before warning or erroring, respectively. These are + only used in offline mode, and are ignored when online. """ def __init__( - self, url: str, offline: bool = False, bootstrap_root: Path | None = None + self, + url: str, + offline: bool = False, + bootstrap_root: Path | None = None, + staleness_warn: timedelta | None = DEFAULT_OFFLINE_STALENESS_WARN, + staleness_error: timedelta | None = DEFAULT_OFFLINE_STALENESS_ERROR, ) -> None: """ Create a new `TrustUpdater`, pulling from the given `url`. @@ -103,6 +115,7 @@ def __init__( _logger.warning( "TUF repository is loaded in offline mode; updates will not be performed" ) + self._warn_if_stale(staleness_warn, staleness_error) else: # Initialize and update the toplevel TUF metadata try: @@ -128,6 +141,41 @@ def __init__( except Exception as e: raise TUFError("Failed to refresh TUF metadata") from e + def _warn_if_stale( + self, + staleness_warn: timedelta | None = DEFAULT_OFFLINE_STALENESS_WARN, + staleness_error: timedelta | None = DEFAULT_OFFLINE_STALENESS_ERROR, + ) -> None: + timestamp_path = self._metadata_dir / "timestamp.json" + if not timestamp_path.exists(): + _logger.debug( + "no cached TUF timestamp; cannot assess root trust freshness." + ) + return + try: + expires = Metadata.from_file(str(timestamp_path)).signed.expires + except TUFExceptions.RepositoryError as e: + _logger.debug("could not read cached TUF timestamp metadata: %s", e) + return + overdue = datetime.now(timezone.utc) - expires + overdue_str = str(overdue).split(".")[0] + if staleness_error is not None and overdue > staleness_error: + raise TUFError( + f"Trust root is stale: the cached TUF timestamp metadata expired " + f"{overdue_str} ago (exceeds the offline hard limit of {staleness_error}). " + f"Refresh the trust root by running without --offline." + ) + elif staleness_warn is not None and overdue > staleness_warn: + _logger.warning( + "Trust root may be stale: the cached TUF timestamp metadata expired " + "%s ago (exceeds the offline staleness threshold of %s). " + "Consider refreshing it by running without --offline.", + overdue_str, + staleness_warn, + ) + else: + return + @lru_cache() def get_trusted_root_path(self) -> str: """Return local path to currently valid trusted root file""" diff --git a/sigstore/models.py b/sigstore/models.py index 2237b772e..457a2137b 100644 --- a/sigstore/models.py +++ b/sigstore/models.py @@ -22,6 +22,7 @@ import logging from collections import defaultdict from collections.abc import Iterable +from datetime import timedelta from enum import Enum from pathlib import Path from textwrap import dedent @@ -61,7 +62,13 @@ KeyringPurpose, RekorKeyring, ) -from sigstore._internal.tuf import DEFAULT_TUF_URL, STAGING_TUF_URL, TrustUpdater +from sigstore._internal.tuf import ( + DEFAULT_OFFLINE_STALENESS_ERROR, + DEFAULT_OFFLINE_STALENESS_WARN, + DEFAULT_TUF_URL, + STAGING_TUF_URL, + TrustUpdater, +) from sigstore._utils import KeyID, cert_is_leaf, cert_is_root_ca, is_timerange_valid from sigstore.errors import Error, MetadataError, TUFError, VerificationError @@ -928,25 +935,39 @@ def from_json(cls, raw: str) -> ClientTrustConfig: def production( cls, offline: bool = False, + staleness_warn: timedelta | None = DEFAULT_OFFLINE_STALENESS_WARN, + staleness_error: timedelta | None = DEFAULT_OFFLINE_STALENESS_ERROR, ) -> ClientTrustConfig: """Create new trust config from Sigstore production TUF repository. If `offline`, will use data in local TUF cache. Otherwise will update the data from remote TUF repository. """ - return cls.from_tuf(DEFAULT_TUF_URL, offline) + return cls.from_tuf( + DEFAULT_TUF_URL, + offline, + staleness_warn=staleness_warn, + staleness_error=staleness_error, + ) @classmethod def staging( cls, offline: bool = False, + staleness_warn: timedelta | None = DEFAULT_OFFLINE_STALENESS_WARN, + staleness_error: timedelta | None = DEFAULT_OFFLINE_STALENESS_ERROR, ) -> ClientTrustConfig: """Create new trust config from Sigstore staging TUF repository. If `offline`, will use data in local TUF cache. Otherwise will update the data from remote TUF repository. """ - return cls.from_tuf(STAGING_TUF_URL, offline) + return cls.from_tuf( + STAGING_TUF_URL, + offline, + staleness_warn=staleness_warn, + staleness_error=staleness_error, + ) @classmethod def from_tuf( @@ -954,13 +975,21 @@ def from_tuf( url: str, offline: bool = False, bootstrap_root: Path | None = None, + staleness_warn: timedelta | None = DEFAULT_OFFLINE_STALENESS_WARN, + staleness_error: timedelta | None = DEFAULT_OFFLINE_STALENESS_ERROR, ) -> ClientTrustConfig: """Create a new trust config from a TUF repository. If `offline`, will use data in local TUF cache. Otherwise will update the trust config from remote TUF repository. """ - updater = TrustUpdater(url, offline, bootstrap_root) + updater = TrustUpdater( + url, + offline, + bootstrap_root, + staleness_warn=staleness_warn, + staleness_error=staleness_error, + ) tr_path = updater.get_trusted_root_path() inner_tr = trustroot_v1.TrustedRoot.from_json(Path(tr_path).read_bytes()) diff --git a/test/unit/internal/test_trust.py b/test/unit/internal/test_trust.py index 4340ee007..9e2b45920 100644 --- a/test/unit/internal/test_trust.py +++ b/test/unit/internal/test_trust.py @@ -13,6 +13,7 @@ # limitations under the License. +import logging import os from datetime import datetime, timedelta, timezone @@ -23,6 +24,7 @@ ServiceConfiguration, ServiceSelector, ) +from tuf.api.metadata import Metadata, Timestamp from sigstore._internal.fulcio.client import FulcioClient from sigstore._internal.rekor.client import RekorClient @@ -32,6 +34,7 @@ CertificateAuthority, KeyringPurpose, ) +from sigstore._internal.tuf import STAGING_TUF_URL, TrustUpdater from sigstore._utils import is_timerange_valid from sigstore.errors import Error, TUFError from sigstore.models import ( @@ -264,6 +267,55 @@ def test_trust_root_tuf_offline(mock_staging_tuf, tuf_dirs): assert fail_reqs == {} +def _write_timestamp(metadata_dir, expires): + """Write a minimal TUF timestamp.json with the given expiry into metadata_dir.""" + metadata_dir.mkdir(parents=True, exist_ok=True) + timestamp = Timestamp() + timestamp.expires = expires + Metadata(signed=timestamp).to_file(str(metadata_dir / "timestamp.json")) + + +def test_trust_updater_offline_stale_warns(tuf_dirs, caplog): + data_dir, _ = tuf_dirs + # expired 2 days ago: past the 24h warn window, within the 7d error window + _write_timestamp(data_dir, datetime.now(timezone.utc) - timedelta(days=2)) + with caplog.at_level(logging.WARNING): + TrustUpdater(STAGING_TUF_URL, offline=True) + assert "Trust root may be stale" in caplog.text + + +def test_trust_updater_offline_stale_errors(tuf_dirs): + data_dir, _ = tuf_dirs + # expired 10 days ago: past the 7d error window + _write_timestamp(data_dir, datetime.now(timezone.utc) - timedelta(days=10)) + with pytest.raises(TUFError, match="Trust root is stale"): + TrustUpdater(STAGING_TUF_URL, offline=True) + + +def test_trust_updater_offline_fresh_does_not_warn(tuf_dirs, caplog): + data_dir, _ = tuf_dirs + _write_timestamp(data_dir, datetime.now(timezone.utc) + timedelta(days=5)) + with caplog.at_level(logging.WARNING): + TrustUpdater(STAGING_TUF_URL, offline=True) + assert "stale" not in caplog.text + + +def test_trust_updater_offline_missing_timestamp_is_quiet(tuf_dirs, caplog): + # no timestamp.json: cannot assess freshness -> no warning, no error + with caplog.at_level(logging.WARNING): + TrustUpdater(STAGING_TUF_URL, offline=True) + assert "stale" not in caplog.text + + +def test_trust_updater_offline_error_disabled_only_warns(tuf_dirs, caplog): + data_dir, _ = tuf_dirs + _write_timestamp(data_dir, datetime.now(timezone.utc) - timedelta(days=10)) + with caplog.at_level(logging.WARNING): + # staleness_error=None disables the hard error; only the warning fires + TrustUpdater(STAGING_TUF_URL, offline=True, staleness_error=None) + assert "Trust root may be stale" in caplog.text + + def test_is_timerange_valid(): def range_from(offset_lower=0, offset_upper=0): base = datetime.now(timezone.utc)