Cache ID token only as long as it is valid (#154)

action.yml

@@ -43,5 +43,4 @@ runs:
         GHA_SIGSTORE_CONFORMANCE_INTERNAL_BE_CAREFUL_DEBUG: "${{ inputs.internal-be-careful-debug }}"
         GHA_SIGSTORE_CONFORMANCE_SKIP_SIGNING: "${{ inputs.skip-signing }}"
         GHA_SIGSTORE_CONFORMANCE_XFAIL: "${{ inputs.xfail }}"
-        GHA_SIGSTORE_GITHUB_TOKEN: "${{ github.token }}"
       shell: bash

test/conftest.py

@@ -1,3 +1,4 @@
+import functools
 import json
 import os
 import shutil
@@ -7,7 +8,6 @@
 from base64 import b64decode
 from collections.abc import Callable
 from datetime import datetime, timedelta
-from functools import lru_cache
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from typing import TypeVar
@@ -61,6 +61,13 @@ def pytest_addoption(parser) -> None:
         action="store_true",
         help="run tests against staging",
     )
+    parser.addoption(
+        "--min-id-token-validity",
+        action="store",
+        help="Minimum validity of the identity token in seconds",
+        type=lambda x: timedelta(seconds=int(x)),
+        default=timedelta(seconds=20),
+    )
 
 
 def pytest_runtest_setup(item):
@@ -83,13 +90,46 @@ def pytest_internalerror(excrepr, excinfo):
     return False
 
 
+def _jwt_cache():
+    def _decorator(fn: Callable[[any], str]):
+        @functools.wraps(fn)
+        def _wrapped(pytestconfig):
+            if pytestconfig.getoption("--skip-signing"):
+                return ""
+
+            # Cache the token for the duration of the test run,
+            # as long as the returned token is not yet expired
+            if hasattr(_wrapped, "token"):
+                min_validity = pytestconfig.getoption("--min-id-token-validity")
+                if _is_valid_at(_wrapped.token, datetime.now() + min_validity):
+                    return _wrapped.token
+
+            token = fn(pytestconfig)
+            _wrapped.token = token
+            return token
+
+        return _wrapped
+
+    return _decorator
+
+
+def _is_valid_at(token: str, reference_time: datetime) -> bool:
+    # split token, b64 decode (with padding), parse as json, validate expiry
+    payload = token.split(".")[1]
+    payload += "=" * (4 - len(payload) % 4)
+    payload_json = json.loads(b64decode(payload))
+
+    expiry = datetime.fromtimestamp(payload_json["exp"])
+    return reference_time < expiry
+
+
 @pytest.fixture
-@lru_cache
+@_jwt_cache()
 def identity_token(pytestconfig) -> str:
     # following code is modified from extremely-dangerous-public-oidc-beacon download-token.py.
     # Caching can be made smarter (to return the cached token only if it is valid) if token
     # starts going invalid during runs
-    MIN_VALIDITY = timedelta(seconds=20)
+    MIN_VALIDITY = pytestconfig.getoption("--min-id-token-validity")
     MAX_RETRY_TIME = timedelta(minutes=5 if os.getenv("CI") else 1)
     RETRY_SLEEP_SECS = 30 if os.getenv("CI") else 5
     GIT_URL = "https://github.com/sigstore-conformance/extremely-dangerous-public-oidc-beacon.git"
@@ -98,18 +138,6 @@ def git_clone(url: str, dir: str) -> None:
         base_cmd = ["git", "clone", "--quiet", "--branch", "current-token", "--depth", "1"]
         subprocess.run(base_cmd + [url, dir], check=True)
 
-    def is_valid_at(token: str, reference_time: datetime) -> bool:
-        # split token, b64 decode (with padding), parse as json, validate expiry
-        payload = token.split(".")[1]
-        payload += "=" * (4 - len(payload) % 4)
-        payload_json = json.loads(b64decode(payload))
-
-        expiry = datetime.fromtimestamp(payload_json["exp"])
-        return reference_time < expiry
-
-    if pytestconfig.getoption("--skip-signing"):
-        return ""
-
     start_time = datetime.now()
     while datetime.now() <= start_time + MAX_RETRY_TIME:
         with TemporaryDirectory() as tempdir:
@@ -118,7 +146,7 @@ def is_valid_at(token: str, reference_time: datetime) -> bool:
             with Path(tempdir, "oidc-token.txt").open() as f:
                 token = f.read().rstrip()
 
-            if is_valid_at(token, datetime.now() + MIN_VALIDITY):
+            if _is_valid_at(token, datetime.now() + MIN_VALIDITY):
                 return token
 
         print(f"Current token expires too early, retrying in {RETRY_SLEEP_SECS} seconds.")

test/test_simple.py

@@ -1,6 +1,6 @@
 import pytest  # type: ignore
 
-from test.conftest import _MakeMaterials
+from test.conftest import _MakeMaterials, identity_token
 
 from .client import SigstoreClient