Method current_access_token

python/openshell/sandbox.py:1024–1078

Return a non-expired access token, refreshing if needed.

(self)

            self._http.close()

    def current_access_token(self) -> str:
        """Return a non-expired access token, refreshing if needed."""
        with self._lock:
            if self._bundle is None:
                self._bundle = _read_oidc_token_bundle(self._gateway_dir)
                if self._bundle is None:
                    raise SandboxError(
                        f"OIDC token for gateway '{self._cluster_name}' is "
                        f"missing or unreadable. Re-authenticate with: "
                        f"openshell gateway login"
                    )
            if self._is_fresh(self._bundle):
                return self._bundle["access_token"]
            # Cached bundle is stale. Before refreshing, re-read disk —
            # another process (CLI, TUI, another SDK client) may have
            # rotated the bundle while we were idle. Adopt the disk
            # bundle when it was refreshed more recently than ours, EVEN
            # WHEN its access token is also stale: otherwise we'd refresh
            # with our in-memory refresh_token, which a rotating IdP may
            # have already invalidated when the other process refreshed
            # (Keycloak with rotation, Entra in strict mode).
            #
            # "More recently" is judged by `expires_at`: a refresh issues
            # a new access token with a forward expiry alongside the
            # (possibly rotated) refresh_token, so the bundle with the
            # later expiry carries the newest refresh_token. This also
            # preserves the write_back=False case, where our in-memory
            # bundle has already rotated past the on-disk one and must
            # NOT be clobbered by the older disk copy.
            disk = _read_oidc_token_bundle(self._gateway_dir)
            if disk is not None and self._expiry(disk) > self._expiry(self._bundle):
                # If the issuer changed under us, the cached token
                # endpoint no longer applies — force re-discovery.
                if _normalize_issuer(disk) != _normalize_issuer(self._bundle):
                    self._token_endpoint = None
                self._bundle = disk
                if self._is_fresh(disk):
                    return disk["access_token"]
            # Truly stale; refresh against the IdP using the freshest
            # bundle we have (disk if it was newer, else in-memory).
            try:
                self._bundle = self._refresh(self._bundle)
            except _InvalidGrantError as exc:
                # We lost a cross-process rotation race: between our disk
                # re-read above and our refresh POST, a peer (CLI, TUI,
                # another SDK client) rotated the refresh_token and the
                # IdP invalidated ours. This is the residual window that
                # neither google-auth nor botocore close without an OS
                # file lock. Rather than lock, recover: re-read disk once
                # and, if a peer wrote a *different* refresh_token, retry
                # with it before surfacing a re-authenticate error.
                self._bundle = self._recover_from_invalid_grant(self._bundle, exc)
            if self._write_back:
                self._write_to_disk(self._bundle)
            return self._bundle["access_token"]

    def _recover_from_invalid_grant(
        self, attempted: dict, exc: _InvalidGrantError