Warn (and optionally fail) on a stale trust root with --offline

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

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 `<number><unit>`, 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`:
 

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)

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

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,39 +935,61 @@ 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(
         cls,
         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())