feat: cache auth check api calls

Kevin Westphal · Kevin Westphal · commit 18e9aa1a2a9d · 2026-05-19T14:32:56.000+02:00
`check_client_api_key` hits `/v1/metadata` and `/v1/api_tokens/self` on every invocation. When a caller runs bursty scans — notably the VSCode extension, which rescans on every save — this is two API round-trips per scan even though the key state rarely changes between them. This commit introduces a cache for auth calls.
diff --git a/changelog.d/20260423_145903_6d7a_cache_api_calls.md b/changelog.d/20260423_145903_6d7a_cache_api_calls.md
@@ -0,0 +1,3 @@
+### Changed
+
+- Successful API key checks are now cached on disk for 5 minutes.
diff --git a/ggshield/cmd/auth/logout.py b/ggshield/cmd/auth/logout.py
@@ -7,6 +7,7 @@
 import ggshield.verticals.hmsl.utils as hmsl_utils
 from ggshield.cmd.utils.common_options import add_common_options
 from ggshield.cmd.utils.context_obj import ContextObj
+from ggshield.core import auth_check_cache
 from ggshield.core.client import create_client
 from ggshield.core.config import Config
 from ggshield.core.config.token_store import get_token_store
@@ -74,6 +75,7 @@ def logout(config: Config, instance_url: str, revoke: bool) -> None:
         revoke_token(config, instance_url)
         hmsl_utils.remove_token_from_disk()
     delete_account_config(config, instance_url)
+    auth_check_cache.invalidate()
 
     click.echo(
         f"Successfully logged out for instance {instance_url}\n\n"
diff --git a/ggshield/core/auth_check_cache.py b/ggshield/core/auth_check_cache.py
@@ -0,0 +1,181 @@
+import hashlib
+import logging
+import os
+import tempfile
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional
+
+import yaml
+from pygitguardian.models import RemediationMessages, SecretScanPreferences, TokenScope
+
+from ggshield.core.dirs import get_cache_dir
+
+
+logger = logging.getLogger(__name__)
+
+# How long a successful auth check (metadata + token scopes) stays valid.
+# Short enough that revoked tokens and scope changes propagate quickly;
+# long enough that a burst of scans (e.g. IDE on-save) shares one check.
+TTL_SECONDS = 300
+
+
+def _cache_file() -> Path:
+    # Resolved lazily so GG_CACHE_DIR overrides (tests, sandboxed envs) are honored.
+    return get_cache_dir() / "auth_check.yaml"
+
+
+@dataclass
+class CachedAuthCheck:
+    # If not None, these are the scopes fetched from /v1/api_tokens/self.
+    # None means we haven't fetched scopes yet (e.g. from an auth-login flow
+    # where no specific scopes were required).
+    scopes: Optional[set[TokenScope]]
+    secrets_engine_version: Optional[str]
+    maximum_payload_size: Optional[int]
+    secret_scan_preferences: Optional[SecretScanPreferences]
+    remediation_messages: Optional[RemediationMessages]
+
+
+def _key_hash(instance_url: str, api_key: str) -> str:
+    return hashlib.sha256(f"{instance_url}\0{api_key}".encode("utf-8")).hexdigest()
+
+
+def load(instance_url: str, api_key: str) -> Optional[CachedAuthCheck]:
+    """Return the cached auth check for this (instance, key) pair, or None on miss."""
+    path = _cache_file()
+    try:
+        with path.open("r") as f:
+            data = yaml.safe_load(f)
+    except FileNotFoundError:
+        return None
+    except (OSError, yaml.YAMLError) as e:
+        logger.warning("Could not load auth check cache: %s", repr(e))
+        return None
+
+    if not isinstance(data, dict):
+        return None
+    if data.get("key_hash") != _key_hash(instance_url, api_key):
+        return None
+    if data.get("expires_at", 0) < time.time():
+        return None
+
+    raw_scopes = data.get("scopes")
+    scopes: Optional[set[TokenScope]]
+    if raw_scopes is None:
+        scopes = None
+    else:
+        scopes = set()
+        for scope_str in raw_scopes:
+            try:
+                scopes.add(TokenScope(scope_str))
+            except ValueError:
+                logger.debug("Ignoring unknown cached scope: '%s'", scope_str)
+
+    raw_version = data.get("secrets_engine_version")
+    secrets_engine_version = raw_version if isinstance(raw_version, str) else None
+
+    raw_max_payload = data.get("maximum_payload_size")
+    maximum_payload_size = raw_max_payload if isinstance(raw_max_payload, int) else None
+
+    raw_ssp = data.get("secret_scan_preferences")
+    secret_scan_preferences: Optional[SecretScanPreferences] = None
+    if isinstance(raw_ssp, dict):
+        try:
+            secret_scan_preferences = SecretScanPreferences(**raw_ssp)
+        except TypeError as e:
+            logger.debug("Ignoring malformed cached secret_scan_preferences: %s", e)
+
+    raw_rm = data.get("remediation_messages")
+    remediation_messages: Optional[RemediationMessages] = None
+    if isinstance(raw_rm, dict):
+        try:
+            remediation_messages = RemediationMessages(**raw_rm)
+        except TypeError as e:
+            logger.debug("Ignoring malformed cached remediation_messages: %s", e)
+
+    return CachedAuthCheck(
+        scopes=scopes,
+        secrets_engine_version=secrets_engine_version,
+        maximum_payload_size=maximum_payload_size,
+        secret_scan_preferences=secret_scan_preferences,
+        remediation_messages=remediation_messages,
+    )
+
+
+def store(
+    instance_url: str,
+    api_key: str,
+    scopes: Optional[set[TokenScope]],
+    secrets_engine_version: Optional[str],
+    maximum_payload_size: Optional[int],
+    secret_scan_preferences: Optional[SecretScanPreferences],
+    remediation_messages: Optional[RemediationMessages],
+) -> None:
+    """Record a successful auth check.
+
+    Pass scopes=None if token scopes were not fetched (only metadata was checked).
+    """
+    payload = {
+        "key_hash": _key_hash(instance_url, api_key),
+        "scopes": (None if scopes is None else sorted(s.value for s in scopes)),
+        "secrets_engine_version": secrets_engine_version,
+        "maximum_payload_size": maximum_payload_size,
+        "secret_scan_preferences": (
+            None
+            if secret_scan_preferences is None
+            else {
+                "maximum_document_size": secret_scan_preferences.maximum_document_size,
+                "maximum_documents_per_scan": secret_scan_preferences.maximum_documents_per_scan,
+            }
+        ),
+        "remediation_messages": (
+            None
+            if remediation_messages is None
+            else {
+                "pre_commit": remediation_messages.pre_commit,
+                "pre_push": remediation_messages.pre_push,
+                "pre_receive": remediation_messages.pre_receive,
+            }
+        ),
+        "expires_at": int(time.time()) + TTL_SECONDS,
+    }
+
+    path = _cache_file()
+    try:
+        path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+        # Re-apply on a pre-existing dir, since mkdir's mode is ignored when the
+        # dir already exists. Keeps the auth-check file out of reach of other
+        # local users on shared POSIX hosts.
+        try:
+            os.chmod(path.parent, 0o700)
+        except OSError as e:
+            logger.debug("Could not tighten cache dir permissions: %s", repr(e))
+        # Atomic write: a concurrent ggshield process would otherwise be able to
+        # observe a truncated YAML file. tempfile in the same directory so
+        # os.replace stays on one filesystem.
+        fd, tmp_path = tempfile.mkstemp(
+            prefix=".auth_check.", suffix=".tmp", dir=path.parent
+        )
+        try:
+            os.chmod(tmp_path, 0o600)
+            with os.fdopen(fd, "w") as f:
+                yaml.dump(payload, f, indent=2, default_flow_style=False)
+            os.replace(tmp_path, path)
+        except Exception:
+            try:
+                os.unlink(tmp_path)
+            except OSError:
+                pass
+            raise
+    except OSError as e:
+        logger.warning("Could not save auth check cache: %s", repr(e))
+
+
+def invalidate() -> None:
+    """Drop the cached auth check, typically after a 401 from any API call."""
+    try:
+        _cache_file().unlink(missing_ok=True)
+    except OSError as e:
+        logger.warning("Could not invalidate auth check cache: %s", repr(e))
diff --git a/ggshield/core/client.py b/ggshield/core/client.py
@@ -9,7 +9,7 @@
 from requests import Session
 from requests.adapters import HTTPAdapter
 
-from . import ui
+from . import auth_check_cache, ui
 from .config import Config
 from .constants import DEFAULT_INSTANCE_URL
 from .errors import (
@@ -130,32 +130,60 @@ def check_client_api_key(client: GGClient, required_scopes: set[TokenScope]) ->
     (either it is invalid or unset). Raises UnexpectedError if the API is down.
 
     If required_scopes is not empty, also checks that the API key has the required scopes.
+
+    Successful checks are cached on disk for a short TTL so bursty callers (e.g. the
+    GitGuardian VSCode extension) do not re-hit /v1/metadata and
+    /v1/api_tokens/self on every invocation.
     """
-    try:
-        response = client.read_metadata()
-    except requests.exceptions.ConnectionError as e:
-        raise ServiceUnavailableError(
-            message="Failed to connect to GitGuardian server. Check your"
-            f" instance URL settings.\nDetails: {e}.",
-        )
+    cached = auth_check_cache.load(client.base_uri, client.api_key)
+    if cached is not None:
+        # Restore the full set of side effects read_metadata() would normally
+        # apply to the client.
+        if cached.secrets_engine_version is not None:
+            client.secrets_engine_version = cached.secrets_engine_version
+        if cached.maximum_payload_size is not None:
+            client.maximum_payload_size = cached.maximum_payload_size
+        if cached.secret_scan_preferences is not None:
+            client.secret_scan_preferences = cached.secret_scan_preferences
+        if cached.remediation_messages is not None:
+            client.remediation_messages = cached.remediation_messages
+
+    if cached is not None and (
+        not required_scopes
+        or (cached.scopes is not None and required_scopes <= cached.scopes)
+    ):
+        return
+
+    if cached is None:
+        try:
+            response = client.read_metadata()
+        except requests.exceptions.ConnectionError as e:
+            raise ServiceUnavailableError(
+                message="Failed to connect to GitGuardian server. Check your"
+                f" instance URL settings.\nDetails: {e}.",
+            )
 
-    if response is None:
-        # None means success
-        pass
-    elif response.status_code == 401:
-        raise APIKeyCheckError(client.base_uri, "Invalid GitGuardian API key.")
-    elif response.status_code == 404:
-        raise UnexpectedError(
-            "The server returned a 404 error. Check your instance URL settings.",
-        )
-    elif response.status_code is not None and 500 <= response.status_code < 600:
-        raise ServiceUnavailableError(
-            message=f"GitGuardian server is not responding.\nDetails: {response.detail}",
-        )
-    else:
-        raise UnexpectedError(
-            f"GitGuardian server is not responding as expected.\nDetails: {response.detail}"
-        )
+        if response is None:
+            # None means success
+            pass
+        elif response.status_code == 401:
+            raise APIKeyCheckError(client.base_uri, "Invalid GitGuardian API key.")
+        elif response.status_code == 404:
+            raise UnexpectedError(
+                "The server returned a 404 error. Check your instance URL settings.",
+            )
+        elif response.status_code is not None and 500 <= response.status_code < 600:
+            raise ServiceUnavailableError(
+                message=f"GitGuardian server is not responding.\nDetails: {response.detail}",
+            )
+        else:
+            raise UnexpectedError(
+                f"GitGuardian server is not responding as expected.\nDetails: {response.detail}"
+            )
+
+    api_scopes: Optional[set[TokenScope]] = (
+        cached.scopes if cached is not None else None
+    )
 
     # Check token scopes if required_scopes is not empty
     if required_scopes:
@@ -164,6 +192,10 @@ def check_client_api_key(client: GGClient, required_scopes: set[TokenScope]) ->
         if not isinstance(response, (Detail, APITokensResponse)):
             raise UnexpectedError("Unexpected api_tokens response")
         elif isinstance(response, Detail):
+            if response.status_code == 401:
+                # Drop the cache for re-verification
+                auth_check_cache.invalidate()
+                raise APIKeyCheckError(client.base_uri, "Invalid GitGuardian API key.")
             raise UnexpectedError(response.detail)
 
         # Build set of API scopes, ignoring unknown ones for forward compatibility
@@ -177,3 +209,13 @@ def check_client_api_key(client: GGClient, required_scopes: set[TokenScope]) ->
         missing_scopes = required_scopes - api_scopes
         if missing_scopes:
             raise MissingScopesError(list(missing_scopes))
+
+    auth_check_cache.store(
+        client.base_uri,
+        client.api_key,
+        api_scopes,
+        client.secrets_engine_version,
+        client.maximum_payload_size,
+        client.secret_scan_preferences,
+        client.remediation_messages,
+    )
diff --git a/ggshield/core/errors.py b/ggshield/core/errors.py
@@ -13,6 +13,7 @@
 from marshmallow import ValidationError
 from pygitguardian.models import Detail, TokenScope
 
+from ggshield.core import auth_check_cache
 from ggshield.core.text_utils import pluralize
 from ggshield.utils.git_shell import InvalidGitRefError
 
@@ -213,6 +214,9 @@ def handle_api_error(detail: Detail) -> None:
     logger.error("status_code=%s", detail.status_code)
     logger.debug("detail=%s", detail.detail)
     if detail.status_code == 401:
+        # The cached metadata/scopes check must not keep masking a revoked
+        # or rotated API key within its TTL.
+        auth_check_cache.invalidate()
         raise click.UsageError(detail.detail)
     if detail.status_code is None:
         raise UnexpectedError(f"Scanning failed: {detail.detail}")
diff --git a/tests/unit/cmd/auth/test_logout.py b/tests/unit/cmd/auth/test_logout.py
@@ -5,6 +5,7 @@
 from requests.exceptions import ConnectionError
 
 from ggshield.__main__ import cli
+from ggshield.core import auth_check_cache
 from ggshield.core.config import Config
 from ggshield.core.config.token_store import KeyringTokenStore, reset_token_store
 from ggshield.core.constants import DEFAULT_INSTANCE_URL
@@ -192,6 +193,34 @@ def test_logout_deletes_from_keyring(self, monkeypatch, cli_fs_runner):
         mock_store.delete_token.assert_called_once_with(DEFAULT_INSTANCE_URL)
         reset_token_store()
 
+    def test_logout_invalidates_auth_check_cache(self, monkeypatch, cli_fs_runner):
+        """
+        GIVEN a saved instance and a populated auth-check cache
+        WHEN running the logout command
+        THEN the auth-check cache file is removed so a stale "still valid" hit
+             cannot survive across the logout
+        """
+        post_mock = Mock(return_value=Mock(status_code=204, ok=True))
+        monkeypatch.setattr("ggshield.core.client.GGClient.post", post_mock)
+
+        add_instance_config()
+
+        auth_check_cache.store(
+            "https://api.gitguardian.com",
+            "test-api-key",
+            None,
+            None,
+            None,
+            None,
+            None,
+        )
+        assert auth_check_cache._cache_file().exists()
+
+        exit_code, output = self.run_cmd(cli_fs_runner)
+
+        assert exit_code == ExitCode.SUCCESS, output
+        assert not auth_check_cache._cache_file().exists()
+
     @staticmethod
     def run_cmd(
         cli_fs_runner,
diff --git a/tests/unit/core/test_auth_check_cache.py b/tests/unit/core/test_auth_check_cache.py
diff --git a/tests/unit/core/test_client.py b/tests/unit/core/test_client.py

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+### Changed`
	`2`	`+`
	`3`	`+- Successful API key checks are now cached on disk for 5 minutes.`