Spaces:

chintakapp
/

prova-core

Running

auth added

0e9b1cb about 1 month ago

19.1 kB

"""Who is calling. Master prompt 2 and 8: every request is verified, every time.
The brain owns login (Karnak's call). Google proves who you are once; the brain then hands out
its own credentials, so any front end (Android, web, a script) uses the same small API and
Supabase is only ever the database:
* ``POST /auth/google``: a Google ID token in, an **access token** (JWT, 3 hours) and a **refresh
  token** (30 days, rotated on every use) out.
* Every other request carries the access token. Checking it needs no database, so strangers and
  expired tokens cost nothing.
* ``user_id`` comes from the token's ``sub`` and from nowhere else (CLAUDE.md). No route reads a
  user id from a request body.
* The email must be on ``ALLOWED_EMAIL``, at login and again on every request.
* Nothing is ever accepted when ``AUTH_SECRET``, ``GOOGLE_CLIENT_IDS`` or ``ALLOWED_EMAIL`` is
  missing.
* The two cron routes use a shared-secret header instead, compared in constant time.
"""

import asyncio
import base64
import hashlib
import hmac
import logging
import secrets
import time
import uuid
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import Any, NoReturn, Protocol

import jwt

log = logging.getLogger(__name__)

ACCESS_TTL_SECONDS = 3 * 60 * 60
REFRESH_TTL_SECONDS = 30 * 24 * 60 * 60
#: A session can be renewed by refreshing, but never lives longer than this in total.
SESSION_MAX_SECONDS = 90 * 24 * 60 * 60
#: How long a failed attempt waits before it is answered. Slows guessing; costs nothing.
FAILURE_DELAY_SECONDS = 1.0
MIN_SECRET_LENGTH = 32

ISSUER = "prova-brain"
AUDIENCE = "prova"
GOOGLE_JWKS_URL = "https://www.googleapis.com/oauth2/v3/certs"
GOOGLE_ISSUERS = ("https://accounts.google.com", "accounts.google.com")
CRON_HEADER = "x-prova-cron-secret"


class AuthError(Exception):
    def __init__(self, status: int, message: str) -> None:
        super().__init__(message)
        self.status = status


@dataclass(frozen=True)
class User:
    id: str
    email: str = ""
    #: Unix time the access token stops working. Set by the verifier, unset for the dev user.
    expires_at: int | None = None


def user_id_for(google_sub: str) -> str:
    """A stable UUID per Google account, so the same person always lands on the same rows."""
    return str(uuid.uuid5(uuid.NAMESPACE_URL, f"prova:google:{google_sub}"))


def _flag(value: str) -> bool:
    return value.strip().lower() in {"1", "true", "yes", "on"}


def _csv(value: str, *, lower: bool = False) -> tuple[str, ...]:
    parts = (part.strip() for part in value.split(","))
    return tuple(part.lower() if lower else part for part in parts if part)


@dataclass(frozen=True)
class AuthConfig:
    #: Signs the access and refresh tokens. 32+ random characters, a Space secret.
    secret: str = ""
    #: The Google OAuth client ids whose ID tokens are accepted (the Web client id; one per front
    #: end if a web app gets its own).
    google_client_ids: tuple[str, ...] = ()
    #: Comma-separated in the environment. Master prompt 8 allows exactly one; a list costs nothing.
    allowed_emails: tuple[str, ...] = ()
    #: Accept a Google token that carries no nonce. A switch for testing with Postman, where no
    #: app can bind a nonce; leave it off in normal use.
    allow_no_nonce: bool = False
    cron_secret: str = ""

    @classmethod
    def from_env(cls, env: dict[str, str]) -> "AuthConfig":
        return cls(
            secret=env.get("AUTH_SECRET", "").strip(),
            google_client_ids=_csv(env.get("GOOGLE_CLIENT_IDS", "")),
            allowed_emails=_csv(env.get("ALLOWED_EMAIL", ""), lower=True),
            allow_no_nonce=_flag(env.get("AUTH_ALLOW_NO_NONCE", "")),
            cron_secret=env.get("CRON_SECRET", ""),
        )

    def problems(self) -> list[str]:
        """What is missing. Empty means login works."""
        found = []
        if len(self.secret) < MIN_SECRET_LENGTH:
            found.append(f"AUTH_SECRET (at least {MIN_SECRET_LENGTH} characters)")
        if not self.google_client_ids:
            found.append("GOOGLE_CLIENT_IDS")
        if not self.allowed_emails:
            found.append("ALLOWED_EMAIL")
        return found

    @property
    def configured(self) -> bool:
        return not self.problems()

    def require_configured(self) -> None:
        # Fail closed: an unconfigured login must never mean "everyone".
        if problems := self.problems():
            raise AuthError(503, "login is not configured: " + ", ".join(problems))


def _derive_key(secret: str, purpose: str) -> bytes:
    """One key per purpose, so an access-token key can never validate a refresh token."""
    return hmac.new(secret.encode(), f"prova/{purpose}/v1".encode(), hashlib.sha256).digest()


def _sha256_hex(value: str) -> str:
    return hashlib.sha256(value.encode()).hexdigest()


def _b64(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).decode().rstrip("=")


# --- Access tokens ---------------------------------------------------------------------------


def issue_access_token(config: AuthConfig, user: User, *, session_id: str, now: float) -> str:
    config.require_configured()
    claims = {
        "iss": ISSUER,
        "aud": AUDIENCE,
        "sub": user.id,
        "email": user.email,
        "sid": session_id,
        "typ": "access",
        "iat": int(now),
        "exp": int(now) + ACCESS_TTL_SECONDS,
        "jti": secrets.token_hex(8),
    }
    return jwt.encode(claims, _derive_key(config.secret, "access"), algorithm="HS256")


class Verifier:
    """Checks the access token on every request. Pure computation: no database, no network."""

    def __init__(self, config: AuthConfig, clock: Callable[[], float] = time.time) -> None:
        self.config = config
        self.clock = clock

    def verify(self, token: str) -> User:
        if not token:
            raise AuthError(401, "missing bearer token")
        self.config.require_configured()

        try:
            claims = jwt.decode(
                token,
                _derive_key(self.config.secret, "access"),
                algorithms=["HS256"],
                audience=AUDIENCE,
                issuer=ISSUER,
                # Expiry and issue time are checked below against the injectable clock.
                options={
                    "verify_exp": False,
                    "verify_iat": False,
                    "require": ["exp", "iat", "sub", "iss", "aud"],
                },
            )
        except jwt.PyJWTError as exc:
            raise AuthError(401, "invalid token") from exc

        now = self.clock()
        if claims["exp"] <= now:
            raise AuthError(401, "token expired")
        if claims["iat"] > now + 60:
            raise AuthError(401, "invalid token")
        if claims.get("typ") != "access":
            raise AuthError(401, "invalid token")
        user_id = str(claims.get("sub") or "")
        if not user_id:
            raise AuthError(401, "token has no subject")

        email = str(claims.get("email") or "").strip().lower()
        if email not in self.config.allowed_emails:
            # Logged without the address: the log is not a place to collect other people's emails.
            log.warning("Rejected a valid token for an account that is not on the allow-list")
            raise AuthError(403, "this Prova belongs to someone else")
        return User(id=user_id, email=email, expires_at=int(claims["exp"]))


# --- Google ----------------------------------------------------------------------------------


class SigningKeySource(Protocol):
    """``jwt.PyJWKClient`` satisfies this; tests pass a stub holding one generated key."""

    def get_signing_key_from_jwt(self, token: str) -> Any: ...


@dataclass(frozen=True)
class GoogleIdentity:
    sub: str
    email: str


class GoogleVerifier:
    """Checks a Google ID token: signature, issuer, audience, expiry, verified email, allow-list."""

    def __init__(
        self,
        config: AuthConfig,
        keys: SigningKeySource | None = None,
        clock: Callable[[], float] = time.time,
    ) -> None:
        self.config = config
        self._keys = keys
        self.clock = clock

    @property
    def keys(self) -> SigningKeySource:
        if self._keys is None:
            self._keys = jwt.PyJWKClient(
                GOOGLE_JWKS_URL, cache_keys=True, lifespan=3600, timeout=10
            )
        return self._keys

    def verify(self, id_token: str, nonce: str | None = None) -> GoogleIdentity:
        self.config.require_configured()
        try:
            header = jwt.get_unverified_header(id_token)
        except jwt.PyJWTError as exc:
            raise AuthError(401, "malformed Google token") from exc
        # Never "none", and never a symmetric algorithm: Google signs with RSA.
        if header.get("alg") != "RS256":
            raise AuthError(401, f"unsupported token algorithm {header.get('alg')!r}")

        try:
            key = self.keys.get_signing_key_from_jwt(id_token)
            claims = jwt.decode(
                id_token,
                key.key,
                algorithms=["RS256"],
                audience=list(self.config.google_client_ids),
                options={
                    "verify_exp": False,
                    "verify_iat": False,
                    "require": ["exp", "iss", "aud", "sub"],
                },
            )
        except jwt.PyJWKClientConnectionError as exc:
            raise AuthError(503, "could not reach Google to check the token") from exc
        except jwt.PyJWTError as exc:
            raise AuthError(401, "invalid Google token") from exc

        if claims["iss"] not in GOOGLE_ISSUERS:
            raise AuthError(401, "invalid Google token")
        if claims["exp"] <= self.clock():
            raise AuthError(401, "Google token expired")
        if claims.get("email_verified") not in (True, "true"):
            raise AuthError(401, "the Google account's email is not verified")
        self._check_nonce(claims.get("nonce"), nonce)

        email = str(claims.get("email") or "").strip().lower()
        sub = str(claims.get("sub") or "")
        if not sub or not email:
            raise AuthError(401, "invalid Google token")
        if email not in self.config.allowed_emails:
            log.warning("Rejected a Google account that is not on the allow-list")
            raise AuthError(403, "this Prova belongs to someone else")
        return GoogleIdentity(sub=sub, email=email)

    def _check_nonce(self, in_token: Any, supplied: str | None) -> None:
        """Ties one Google token to one sign-in attempt, so a copied token cannot be replayed."""
        if not in_token:
            if self.config.allow_no_nonce:
                return
            raise AuthError(401, "the Google token carries no nonce")
        if not supplied:
            raise AuthError(401, "a nonce is required for this Google token")
        expected = str(in_token).encode()
        # The Android app gives Google the SHA-256 of its nonce; a web page gives it as it is.
        raw_ok = hmac.compare_digest(expected, supplied.encode())
        hashed_ok = hmac.compare_digest(expected, _sha256_hex(supplied).encode())
        if not (raw_ok or hashed_ok):
            raise AuthError(401, "nonce mismatch")


# --- Login sessions and refresh tokens -------------------------------------------------------


@dataclass(frozen=True)
class TokenPair:
    access_token: str
    refresh_token: str
    user: User
    expires_in: int = ACCESS_TTL_SECONDS
    refresh_expires_in: int = REFRESH_TTL_SECONDS


def _as_datetime(value: Any) -> datetime:
    if isinstance(value, datetime):
        return value if value.tzinfo else value.replace(tzinfo=UTC)
    return datetime.fromisoformat(str(value).replace("Z", "+00:00"))


class AuthService:
    """Login, refresh and logout. The only code that writes ``login_sessions``.
    A refresh token is ``<session id>.<secret>.<tag>``. The tag is an HMAC of the first two parts,
    checked before anything else, so a made-up token is refused without a database read. The
    database keeps only a hash of the token, and the hash of the token it replaced: if an old one
    shows up again, someone copied it, and the whole session is revoked.
    """

    def __init__(
        self,
        config: AuthConfig,
        store: Any,
        google: GoogleVerifier | None = None,
        clock: Callable[[], float] = time.time,
        delay: Callable[[float], Awaitable[None]] = asyncio.sleep,
    ) -> None:
        self.config = config
        self.store = store
        self.google = google or GoogleVerifier(config, clock=clock)
        self.clock = clock
        self.delay = delay

    # --- Public API ---------------------------------------------------------

    async def login_with_google(
        self, id_token: str, nonce: str | None, client: str, device: str
    ) -> TokenPair:
        self.config.require_configured()
        try:
            # Fetching Google's keys is a blocking call the first time; keep it off the event loop.
            identity = await asyncio.to_thread(self.google.verify, id_token, nonce)
        except AuthError:
            await self.delay(FAILURE_DELAY_SECONDS)
            raise
        user = User(id=user_id_for(identity.sub), email=identity.email)

        now = self._now()
        await self.store.prune_login_sessions(user.id, now)
        session_id = str(uuid.uuid4())
        refresh_token = self._mint_refresh_token(session_id)
        await self.store.create_login_session(
            user.id,
            {
                "id": session_id,
                "email": user.email,
                "refresh_hash": _sha256_hex(refresh_token),
                "client": client,
                "device": device,
                "created_at": now.isoformat(),
                "last_used_at": now.isoformat(),
                "expires_at": (now + timedelta(seconds=REFRESH_TTL_SECONDS)).isoformat(),
            },
        )
        return self._pair(user, session_id, refresh_token, now)

    async def refresh(self, refresh_token: str) -> TokenPair:
        self.config.require_configured()
        session_id = self._session_id_of(refresh_token)
        if session_id is None:
            await self._refuse("invalid refresh token")

        row = await self.store.find_login_session(session_id)
        if row is None:
            await self._refuse("invalid refresh token")
        user_id = row["user_id"]

        presented = _sha256_hex(refresh_token)
        if not hmac.compare_digest(presented, str(row["refresh_hash"])):
            previous = row.get("prev_hash")
            if previous and hmac.compare_digest(presented, str(previous)):
                # An earlier token came back after it was replaced: it was copied. End the session.
                await self.store.delete_login_session(user_id, session_id)
                log.warning("A replaced refresh token was used again; the session was revoked")
                await self._refuse("refresh token reused; sign in again")
            await self._refuse("invalid refresh token")

        now = self._now()
        created = _as_datetime(row["created_at"])
        if (
            _as_datetime(row["expires_at"]) <= now
            or created + timedelta(seconds=SESSION_MAX_SECONDS) <= now
        ):
            await self.store.delete_login_session(user_id, session_id)
            await self._refuse("session expired; sign in again")

        email = str(row.get("email") or "").lower()
        if email not in self.config.allowed_emails:
            await self.store.delete_login_session(user_id, session_id)
            raise AuthError(403, "this Prova belongs to someone else")

        new_token = self._mint_refresh_token(session_id)
        renewed = min(
            now + timedelta(seconds=REFRESH_TTL_SECONDS),
            created + timedelta(seconds=SESSION_MAX_SECONDS),
        )
        await self.store.update_login_session(
            user_id,
            session_id,
            {
                "refresh_hash": _sha256_hex(new_token),
                "prev_hash": row["refresh_hash"],
                "last_used_at": now.isoformat(),
                "expires_at": renewed.isoformat(),
            },
        )
        return self._pair(User(id=user_id, email=email), session_id, new_token, now)

    async def logout(self, refresh_token: str) -> None:
        """Ends the session. Says nothing about whether the token was real."""
        session_id = self._session_id_of(refresh_token)
        if session_id is None:
            return
        row = await self.store.find_login_session(session_id)
        if row is None:
            return
        presented = _sha256_hex(refresh_token)
        known = (str(row["refresh_hash"]), str(row.get("prev_hash") or ""))
        if any(hmac.compare_digest(presented, candidate) for candidate in known):
            await self.store.delete_login_session(row["user_id"], session_id)

    # --- Internals ----------------------------------------------------------

    def _now(self) -> datetime:
        return datetime.fromtimestamp(self.clock(), UTC)

    async def _refuse(self, message: str) -> NoReturn:
        await self.delay(FAILURE_DELAY_SECONDS)
        raise AuthError(401, message)

    def _tag(self, body: str) -> str:
        key = _derive_key(self.config.secret, "refresh")
        return _b64(hmac.new(key, body.encode(), hashlib.sha256).digest())

    def _mint_refresh_token(self, session_id: str) -> str:
        body = f"{session_id}.{secrets.token_urlsafe(32)}"
        return f"{body}.{self._tag(body)}"

    def _session_id_of(self, refresh_token: str) -> str | None:
        """The session id, but only for a token this brain minted (its tag checks out)."""
        parts = refresh_token.split(".")
        if len(parts) != 3:
            return None
        body = f"{parts[0]}.{parts[1]}"
        if not hmac.compare_digest(parts[2].encode(), self._tag(body).encode()):
            return None
        return parts[0]

    def _pair(self, user: User, session_id: str, refresh_token: str, now: datetime) -> TokenPair:
        access = issue_access_token(self.config, user, session_id=session_id, now=now.timestamp())
        return TokenPair(access_token=access, refresh_token=refresh_token, user=user)


# --- Helpers ---------------------------------------------------------------------------------


def bearer_token(authorization: str | None) -> str:
    if not authorization:
        return ""
    scheme, _, token = authorization.partition(" ")
    return token.strip() if scheme.lower() == "bearer" else ""


def check_cron_secret(config: AuthConfig, supplied: str | None) -> None:
    if not config.cron_secret:
        raise AuthError(500, "CRON_SECRET is not configured")
    if not supplied or not hmac.compare_digest(supplied.encode(), config.cron_secret.encode()):
        raise AuthError(401, "bad cron secret")

	"""Who is calling. Master prompt 2 and 8: every request is verified, every time.

	The brain owns login (Karnak's call). Google proves who you are once; the brain then hands out
	its own credentials, so any front end (Android, web, a script) uses the same small API and
	Supabase is only ever the database:

	* ``POST /auth/google``: a Google ID token in, an access token (JWT, 3 hours) and a **refresh
	token** (30 days, rotated on every use) out.
	* Every other request carries the access token. Checking it needs no database, so strangers and
	expired tokens cost nothing.
	* ``user_id`` comes from the token's ``sub`` and from nowhere else (CLAUDE.md). No route reads a
	user id from a request body.
	* The email must be on ``ALLOWED_EMAIL``, at login and again on every request.
	* Nothing is ever accepted when ``AUTH_SECRET``, ``GOOGLE_CLIENT_IDS`` or ``ALLOWED_EMAIL`` is
	missing.
	* The two cron routes use a shared-secret header instead, compared in constant time.
	"""

	import asyncio
	import base64
	import hashlib
	import hmac
	import logging
	import secrets
	import time
	import uuid
	from collections.abc import Awaitable, Callable
	from dataclasses import dataclass
	from datetime import UTC, datetime, timedelta
	from typing import Any, NoReturn, Protocol

	import jwt

	log = logging.getLogger(__name__)

	ACCESS_TTL_SECONDS = 3 * 60 * 60
	REFRESH_TTL_SECONDS = 30 * 24 * 60 * 60
	#: A session can be renewed by refreshing, but never lives longer than this in total.
	SESSION_MAX_SECONDS = 90 * 24 * 60 * 60
	#: How long a failed attempt waits before it is answered. Slows guessing; costs nothing.
	FAILURE_DELAY_SECONDS = 1.0
	MIN_SECRET_LENGTH = 32

	ISSUER = "prova-brain"
	AUDIENCE = "prova"
	GOOGLE_JWKS_URL = "https://www.googleapis.com/oauth2/v3/certs"
	GOOGLE_ISSUERS = ("https://accounts.google.com", "accounts.google.com")
	CRON_HEADER = "x-prova-cron-secret"


	class AuthError(Exception):
	def __init__(self, status: int, message: str) -> None:
	super().__init__(message)
	self.status = status


	@dataclass(frozen=True)
	class User:
	id: str
	email: str = ""
	#: Unix time the access token stops working. Set by the verifier, unset for the dev user.
	expires_at: int \| None = None


	def user_id_for(google_sub: str) -> str:
	"""A stable UUID per Google account, so the same person always lands on the same rows."""
	return str(uuid.uuid5(uuid.NAMESPACE_URL, f"prova:google:{google_sub}"))


	def _flag(value: str) -> bool:
	return value.strip().lower() in {"1", "true", "yes", "on"}


	def _csv(value: str, *, lower: bool = False) -> tuple[str, ...]:
	parts = (part.strip() for part in value.split(","))
	return tuple(part.lower() if lower else part for part in parts if part)


	@dataclass(frozen=True)
	class AuthConfig:
	#: Signs the access and refresh tokens. 32+ random characters, a Space secret.
	secret: str = ""
	#: The Google OAuth client ids whose ID tokens are accepted (the Web client id; one per front
	#: end if a web app gets its own).
	google_client_ids: tuple[str, ...] = ()
	#: Comma-separated in the environment. Master prompt 8 allows exactly one; a list costs nothing.
	allowed_emails: tuple[str, ...] = ()
	#: Accept a Google token that carries no nonce. A switch for testing with Postman, where no
	#: app can bind a nonce; leave it off in normal use.
	allow_no_nonce: bool = False
	cron_secret: str = ""

	@classmethod
	def from_env(cls, env: dict[str, str]) -> "AuthConfig":
	return cls(
	secret=env.get("AUTH_SECRET", "").strip(),
	google_client_ids=_csv(env.get("GOOGLE_CLIENT_IDS", "")),
	allowed_emails=_csv(env.get("ALLOWED_EMAIL", ""), lower=True),
	allow_no_nonce=_flag(env.get("AUTH_ALLOW_NO_NONCE", "")),
	cron_secret=env.get("CRON_SECRET", ""),
	)

	def problems(self) -> list[str]:
	"""What is missing. Empty means login works."""
	found = []
	if len(self.secret) < MIN_SECRET_LENGTH:
	found.append(f"AUTH_SECRET (at least {MIN_SECRET_LENGTH} characters)")
	if not self.google_client_ids:
	found.append("GOOGLE_CLIENT_IDS")
	if not self.allowed_emails:
	found.append("ALLOWED_EMAIL")
	return found

	@property
	def configured(self) -> bool:
	return not self.problems()

	def require_configured(self) -> None:
	# Fail closed: an unconfigured login must never mean "everyone".
	if problems := self.problems():
	raise AuthError(503, "login is not configured: " + ", ".join(problems))


	def _derive_key(secret: str, purpose: str) -> bytes:
	"""One key per purpose, so an access-token key can never validate a refresh token."""
	return hmac.new(secret.encode(), f"prova/{purpose}/v1".encode(), hashlib.sha256).digest()


	def _sha256_hex(value: str) -> str:
	return hashlib.sha256(value.encode()).hexdigest()


	def _b64(data: bytes) -> str:
	return base64.urlsafe_b64encode(data).decode().rstrip("=")


	# --- Access tokens ---------------------------------------------------------------------------


	def issue_access_token(config: AuthConfig, user: User, *, session_id: str, now: float) -> str:
	config.require_configured()
	claims = {
	"iss": ISSUER,
	"aud": AUDIENCE,
	"sub": user.id,
	"email": user.email,
	"sid": session_id,
	"typ": "access",
	"iat": int(now),
	"exp": int(now) + ACCESS_TTL_SECONDS,
	"jti": secrets.token_hex(8),
	}
	return jwt.encode(claims, _derive_key(config.secret, "access"), algorithm="HS256")


	class Verifier:
	"""Checks the access token on every request. Pure computation: no database, no network."""

	def __init__(self, config: AuthConfig, clock: Callable[[], float] = time.time) -> None:
	self.config = config
	self.clock = clock

	def verify(self, token: str) -> User:
	if not token:
	raise AuthError(401, "missing bearer token")
	self.config.require_configured()

	try:
	claims = jwt.decode(
	token,
	_derive_key(self.config.secret, "access"),
	algorithms=["HS256"],
	audience=AUDIENCE,
	issuer=ISSUER,
	# Expiry and issue time are checked below against the injectable clock.
	options={
	"verify_exp": False,
	"verify_iat": False,
	"require": ["exp", "iat", "sub", "iss", "aud"],
	},
	)
	except jwt.PyJWTError as exc:
	raise AuthError(401, "invalid token") from exc

	now = self.clock()
	if claims["exp"] <= now:
	raise AuthError(401, "token expired")
	if claims["iat"] > now + 60:
	raise AuthError(401, "invalid token")
	if claims.get("typ") != "access":
	raise AuthError(401, "invalid token")
	user_id = str(claims.get("sub") or "")
	if not user_id:
	raise AuthError(401, "token has no subject")

	email = str(claims.get("email") or "").strip().lower()
	if email not in self.config.allowed_emails:
	# Logged without the address: the log is not a place to collect other people's emails.
	log.warning("Rejected a valid token for an account that is not on the allow-list")
	raise AuthError(403, "this Prova belongs to someone else")
	return User(id=user_id, email=email, expires_at=int(claims["exp"]))


	# --- Google ----------------------------------------------------------------------------------


	class SigningKeySource(Protocol):
	"""``jwt.PyJWKClient`` satisfies this; tests pass a stub holding one generated key."""

	def get_signing_key_from_jwt(self, token: str) -> Any: ...


	@dataclass(frozen=True)
	class GoogleIdentity:
	sub: str
	email: str


	class GoogleVerifier:
	"""Checks a Google ID token: signature, issuer, audience, expiry, verified email, allow-list."""

	def __init__(
	self,
	config: AuthConfig,
	keys: SigningKeySource \| None = None,
	clock: Callable[[], float] = time.time,
	) -> None:
	self.config = config
	self._keys = keys
	self.clock = clock

	@property
	def keys(self) -> SigningKeySource:
	if self._keys is None:
	self._keys = jwt.PyJWKClient(
	GOOGLE_JWKS_URL, cache_keys=True, lifespan=3600, timeout=10
	)
	return self._keys

	def verify(self, id_token: str, nonce: str \| None = None) -> GoogleIdentity:
	self.config.require_configured()
	try:
	header = jwt.get_unverified_header(id_token)
	except jwt.PyJWTError as exc:
	raise AuthError(401, "malformed Google token") from exc
	# Never "none", and never a symmetric algorithm: Google signs with RSA.
	if header.get("alg") != "RS256":
	raise AuthError(401, f"unsupported token algorithm {header.get('alg')!r}")

	try:
	key = self.keys.get_signing_key_from_jwt(id_token)
	claims = jwt.decode(
	id_token,
	key.key,
	algorithms=["RS256"],
	audience=list(self.config.google_client_ids),
	options={
	"verify_exp": False,
	"verify_iat": False,
	"require": ["exp", "iss", "aud", "sub"],
	},
	)
	except jwt.PyJWKClientConnectionError as exc:
	raise AuthError(503, "could not reach Google to check the token") from exc
	except jwt.PyJWTError as exc:
	raise AuthError(401, "invalid Google token") from exc

	if claims["iss"] not in GOOGLE_ISSUERS:
	raise AuthError(401, "invalid Google token")
	if claims["exp"] <= self.clock():
	raise AuthError(401, "Google token expired")
	if claims.get("email_verified") not in (True, "true"):
	raise AuthError(401, "the Google account's email is not verified")
	self._check_nonce(claims.get("nonce"), nonce)

	email = str(claims.get("email") or "").strip().lower()
	sub = str(claims.get("sub") or "")
	if not sub or not email:
	raise AuthError(401, "invalid Google token")
	if email not in self.config.allowed_emails:
	log.warning("Rejected a Google account that is not on the allow-list")
	raise AuthError(403, "this Prova belongs to someone else")
	return GoogleIdentity(sub=sub, email=email)

	def _check_nonce(self, in_token: Any, supplied: str \| None) -> None:
	"""Ties one Google token to one sign-in attempt, so a copied token cannot be replayed."""
	if not in_token:
	if self.config.allow_no_nonce:
	return
	raise AuthError(401, "the Google token carries no nonce")
	if not supplied:
	raise AuthError(401, "a nonce is required for this Google token")
	expected = str(in_token).encode()
	# The Android app gives Google the SHA-256 of its nonce; a web page gives it as it is.
	raw_ok = hmac.compare_digest(expected, supplied.encode())
	hashed_ok = hmac.compare_digest(expected, _sha256_hex(supplied).encode())
	if not (raw_ok or hashed_ok):
	raise AuthError(401, "nonce mismatch")


	# --- Login sessions and refresh tokens -------------------------------------------------------


	@dataclass(frozen=True)
	class TokenPair:
	access_token: str
	refresh_token: str
	user: User
	expires_in: int = ACCESS_TTL_SECONDS
	refresh_expires_in: int = REFRESH_TTL_SECONDS


	def _as_datetime(value: Any) -> datetime:
	if isinstance(value, datetime):
	return value if value.tzinfo else value.replace(tzinfo=UTC)
	return datetime.fromisoformat(str(value).replace("Z", "+00:00"))


	class AuthService:
	"""Login, refresh and logout. The only code that writes ``login_sessions``.

	A refresh token is ``<session id>.<secret>.<tag>``. The tag is an HMAC of the first two parts,
	checked before anything else, so a made-up token is refused without a database read. The
	database keeps only a hash of the token, and the hash of the token it replaced: if an old one
	shows up again, someone copied it, and the whole session is revoked.
	"""

	def __init__(
	self,
	config: AuthConfig,
	store: Any,
	google: GoogleVerifier \| None = None,
	clock: Callable[[], float] = time.time,
	delay: Callable[[float], Awaitable[None]] = asyncio.sleep,
	) -> None:
	self.config = config
	self.store = store
	self.google = google or GoogleVerifier(config, clock=clock)
	self.clock = clock
	self.delay = delay

	# --- Public API ---------------------------------------------------------

	async def login_with_google(
	self, id_token: str, nonce: str \| None, client: str, device: str
	) -> TokenPair:
	self.config.require_configured()
	try:
	# Fetching Google's keys is a blocking call the first time; keep it off the event loop.
	identity = await asyncio.to_thread(self.google.verify, id_token, nonce)
	except AuthError:
	await self.delay(FAILURE_DELAY_SECONDS)
	raise
	user = User(id=user_id_for(identity.sub), email=identity.email)

	now = self._now()
	await self.store.prune_login_sessions(user.id, now)
	session_id = str(uuid.uuid4())
	refresh_token = self._mint_refresh_token(session_id)
	await self.store.create_login_session(
	user.id,
	{
	"id": session_id,
	"email": user.email,
	"refresh_hash": _sha256_hex(refresh_token),
	"client": client,
	"device": device,
	"created_at": now.isoformat(),
	"last_used_at": now.isoformat(),
	"expires_at": (now + timedelta(seconds=REFRESH_TTL_SECONDS)).isoformat(),
	},
	)
	return self._pair(user, session_id, refresh_token, now)

	async def refresh(self, refresh_token: str) -> TokenPair:
	self.config.require_configured()
	session_id = self._session_id_of(refresh_token)
	if session_id is None:
	await self._refuse("invalid refresh token")

	row = await self.store.find_login_session(session_id)
	if row is None:
	await self._refuse("invalid refresh token")
	user_id = row["user_id"]

	presented = _sha256_hex(refresh_token)
	if not hmac.compare_digest(presented, str(row["refresh_hash"])):
	previous = row.get("prev_hash")
	if previous and hmac.compare_digest(presented, str(previous)):
	# An earlier token came back after it was replaced: it was copied. End the session.
	await self.store.delete_login_session(user_id, session_id)
	log.warning("A replaced refresh token was used again; the session was revoked")
	await self._refuse("refresh token reused; sign in again")
	await self._refuse("invalid refresh token")

	now = self._now()
	created = _as_datetime(row["created_at"])
	if (
	_as_datetime(row["expires_at"]) <= now
	or created + timedelta(seconds=SESSION_MAX_SECONDS) <= now
	):
	await self.store.delete_login_session(user_id, session_id)
	await self._refuse("session expired; sign in again")

	email = str(row.get("email") or "").lower()
	if email not in self.config.allowed_emails:
	await self.store.delete_login_session(user_id, session_id)
	raise AuthError(403, "this Prova belongs to someone else")

	new_token = self._mint_refresh_token(session_id)
	renewed = min(
	now + timedelta(seconds=REFRESH_TTL_SECONDS),
	created + timedelta(seconds=SESSION_MAX_SECONDS),
	)
	await self.store.update_login_session(
	user_id,
	session_id,
	{
	"refresh_hash": _sha256_hex(new_token),
	"prev_hash": row["refresh_hash"],
	"last_used_at": now.isoformat(),
	"expires_at": renewed.isoformat(),
	},
	)
	return self._pair(User(id=user_id, email=email), session_id, new_token, now)

	async def logout(self, refresh_token: str) -> None:
	"""Ends the session. Says nothing about whether the token was real."""
	session_id = self._session_id_of(refresh_token)
	if session_id is None:
	return
	row = await self.store.find_login_session(session_id)
	if row is None:
	return
	presented = _sha256_hex(refresh_token)
	known = (str(row["refresh_hash"]), str(row.get("prev_hash") or ""))
	if any(hmac.compare_digest(presented, candidate) for candidate in known):
	await self.store.delete_login_session(row["user_id"], session_id)

	# --- Internals ----------------------------------------------------------

	def _now(self) -> datetime:
	return datetime.fromtimestamp(self.clock(), UTC)

	async def _refuse(self, message: str) -> NoReturn:
	await self.delay(FAILURE_DELAY_SECONDS)
	raise AuthError(401, message)

	def _tag(self, body: str) -> str:
	key = _derive_key(self.config.secret, "refresh")
	return _b64(hmac.new(key, body.encode(), hashlib.sha256).digest())

	def _mint_refresh_token(self, session_id: str) -> str:
	body = f"{session_id}.{secrets.token_urlsafe(32)}"
	return f"{body}.{self._tag(body)}"

	def _session_id_of(self, refresh_token: str) -> str \| None:
	"""The session id, but only for a token this brain minted (its tag checks out)."""
	parts = refresh_token.split(".")
	if len(parts) != 3:
	return None
	body = f"{parts[0]}.{parts[1]}"
	if not hmac.compare_digest(parts[2].encode(), self._tag(body).encode()):
	return None
	return parts[0]

	def _pair(self, user: User, session_id: str, refresh_token: str, now: datetime) -> TokenPair:
	access = issue_access_token(self.config, user, session_id=session_id, now=now.timestamp())
	return TokenPair(access_token=access, refresh_token=refresh_token, user=user)


	# --- Helpers ---------------------------------------------------------------------------------


	def bearer_token(authorization: str \| None) -> str:
	if not authorization:
	return ""
	scheme, _, token = authorization.partition(" ")
	return token.strip() if scheme.lower() == "bearer" else ""


	def check_cron_secret(config: AuthConfig, supplied: str \| None) -> None:
	if not config.cron_secret:
	raise AuthError(500, "CRON_SECRET is not configured")
	if not supplied or not hmac.compare_digest(supplied.encode(), config.cron_secret.encode()):
	raise AuthError(401, "bad cron secret")