Strategies¶

async def mark_used(self, *, key_id: str, nonce: str, ttl_seconds: int) -> ApiKeyNonceStoreResult:
    """Record ``nonce`` for ``key_id`` if it was not already seen."""

async def classify_failure_code(self, token: str | None) -> ErrorCode:
    """Return the most specific API-key authentication failure code for ``token``."""
    return api_key_failure_reason_to_error_code(await self.classify_failure_reason(token))

async def classify_failure_reason(self, token: str | None) -> ApiKeyFailureReason:
    """Return the most specific API-key authentication failure reason for ``token``."""
    if token == API_KEY_HMAC_SCHEME:
        return await self._classify_signed_failure_reason()
    return await self._classify_bearer_failure_reason(token)

@override
async def destroy_token(self, token: str, user: UP) -> None:
    """Do nothing because API-key revocation is handled by API-key management flows."""

@override
async def read_token(self, token: str | None, user_manager: object) -> UP | None:
    """Resolve a user from an API-key token.

    Returns:
        Resolved user, or ``None`` when verification fails.
    """
    # StrategyProtocol accepts object for pluggability; this concrete strategy requires the user-manager contract.
    result = await self.read_token_with_context(
        token,
        user_manager=cast("UserManagerProtocol[UP, ID]", user_manager),
    )
    return None if result is None else result.user

async def read_token_attempt(
    self,
    token: str | None,
    user_manager: UserManagerProtocol[UP, ID],
) -> ApiKeyAuthenticationAttempt[UP]:
    """Resolve an API-key request and preserve the failed reason when rejected.

    Returns:
        Successful authentication result, or ``None`` plus a typed failure reason.
    """
    if token is None:
        return _api_key_failure(ApiKeyFailureReason.INVALID)
    if token == API_KEY_HMAC_SCHEME:
        return await self._read_signed_request(user_manager)
    return await self._read_bearer_api_key(token, user_manager)

async def read_token_with_context(
    self,
    token: str | None,
    user_manager: UserManagerProtocol[UP, ID],
) -> ApiKeyAuthenticationResult[UP] | None:
    """Resolve a user and API-key context from a canonical API-key token.

    Returns:
        Resolved user and API-key context, or ``None`` when verification fails.
    """
    return (await self.read_token_attempt(token, user_manager)).result

@override
async def write_token(self, user: UP) -> str:
    """Reject login-token issuance because API keys are manager-issued credentials.

    Raises:
        TokenError: Always, because API keys are not login-flow tokens.
    """
    msg = "ApiKeyStrategy does not issue login tokens."
    raise TokenError(msg)

async def read_token_with_context(
    self,
    token: str | None,
    user_manager: UserManagerProtocol[UP, ID],
) -> AuthT | None:
    """Resolve a user plus strategy-specific authentication context."""

def __post_init__(self) -> None:
    """Validate the supplied token-model classes eagerly."""
    _validate_token_model_contract(
        self.access_token_model,
        field_name="access_token_model",
        required_attributes=_REQUIRED_ACCESS_TOKEN_MODEL_ATTRIBUTES,
    )
    _validate_token_model_contract(
        self.refresh_token_model,
        field_name="refresh_token_model",
        required_attributes=_REQUIRED_REFRESH_TOKEN_MODEL_ATTRIBUTES,
    )
    _validate_token_model_contract(
        self.consumed_refresh_token_digest_model,
        field_name="consumed_refresh_token_digest_model",
        required_attributes=_REQUIRED_CONSUMED_DIGEST_MODEL_ATTRIBUTES,
    )

async def cleanup_expired_tokens(self, session: AsyncSession) -> int:
    """Delete expired access and refresh tokens for the configured TTLs.

    Returns:
        Total number of deleted access-token and refresh-token rows.
    """
    now = datetime.now(tz=UTC)
    access_cutoff = now - self.max_age
    refresh_cutoff = now - self.refresh_max_age

    access_result = await session.execute(
        delete(self.access_token_model).where(self.access_token_model.created_at <= access_cutoff),
    )
    expired_refresh_session_ids = select(self.refresh_token_model.session_id).where(
        self.refresh_token_model.created_at <= refresh_cutoff,
    )
    await session.execute(
        delete(self.consumed_refresh_token_digest_model).where(
            self.consumed_refresh_token_digest_model.session_id.in_(expired_refresh_session_ids),
        ),
    )
    await session.execute(
        delete(self.consumed_refresh_token_digest_model).where(
            self.consumed_refresh_token_digest_model.consumed_at <= refresh_cutoff,
        ),
    )
    refresh_result = await session.execute(
        delete(self.refresh_token_model).where(self.refresh_token_model.created_at <= refresh_cutoff),
    )
    await session.commit()

    access_rowcount = getattr(access_result, "rowcount", 0) or 0
    refresh_rowcount = getattr(refresh_result, "rowcount", 0) or 0
    return access_rowcount + refresh_rowcount

@override
async def destroy_token(self, token: str, user: UP) -> None:
    """Delete a persisted token."""
    token_digest = self._token_digest(token)
    await self._repository(self._access_token_repository_type).delete_where(token=token_digest, auto_commit=False)
    await self.session.commit()

async def invalidate_all_tokens(self, user: UP) -> None:
    """Delete all persisted access and refresh tokens for the given user."""
    await self._repository(self._access_token_repository_type).delete_where(user_id=user.id, auto_commit=False)
    await self._delete_refresh_session_consumed_digests(
        select(self.refresh_token_model.session_id).where(self.refresh_token_model.user_id == user.id),
    )
    await self._repository(self._refresh_token_repository_type).delete_where(user_id=user.id, auto_commit=False)
    await self.session.commit()

@override
async def read_token(
    self,
    token: str | None,
    user_manager: UserManagerProtocol[UP, ID],
) -> UP | None:
    """Resolve a user from an opaque database token.

    Returns:
        Related user when the token exists and is not expired, otherwise ``None``.
    """
    if token is None:
        return None

    access_token = await self._resolve_access_token(token)
    if access_token is None or self._is_token_expired(access_token.created_at, self.max_age):
        return None

    return access_token.user

def with_session(self, session: AsyncSessionT) -> DatabaseTokenStrategy[UP, ID]:
    """Return a copy of the strategy bound to the provided async session."""
    return type(self)(
        session=session,
        token_hash_secret=self._token_hash_secret.decode(),
        token_models=self.token_models,
        max_age=self.max_age,
        refresh_max_age=self.refresh_max_age,
        token_bytes=self.token_bytes,
        unsafe_testing=self.unsafe_testing,
    )

@override
async def write_refresh_token(self, user: UP) -> str:
    """Persist and return a new opaque refresh token for the user.

    Returns:
        Newly created opaque refresh-token string.
    """
    token, token_digest = mint_opaque_token(token_bytes=self.token_bytes, token_hash_secret=self._token_hash_secret)
    refresh_token = self.refresh_token_model(
        token=token_digest,
        user_id=user.id,
        client_metadata=self._consume_refresh_token_request_metadata(),
    )
    await self._repository(self._refresh_token_repository_type).add(refresh_token, auto_refresh=True)
    return token

@override
async def write_token(self, user: UP) -> str:
    """Persist and return a new opaque token for the user.

    Returns:
        Newly created opaque token string.
    """
    token, token_digest = mint_opaque_token(token_bytes=self.token_bytes, token_hash_secret=self._token_hash_secret)
    access_token = self.access_token_model(token=token_digest, user_id=user.id)
    await self._repository(self._access_token_repository_type).add(access_token, auto_refresh=True)
    return token

async def mark_used(self, *, key_id: str, nonce: str, ttl_seconds: int) -> ApiKeyNonceStoreResult:
    """Record a nonce until TTL expiry, rejecting replays fail-closed.

    Returns:
        Stored/replay outcome for the nonce insert.
    """
    async with self._lock:
        now = read_clock(self._clock)
        self._prune(now)
        key = (key_id, nonce)
        if key in self._entries:
            return ApiKeyNonceStoreResult(stored=False, rejected_as_replay=True)
        if len(self._entries) >= self.max_entries:
            return ApiKeyNonceStoreResult(stored=False, rejected_as_replay=False)
        self._entries[key] = now + ttl_seconds
        return ApiKeyNonceStoreResult(stored=True)

@override
async def destroy_token(self, token: str, user: UP) -> None:
    """Revoke the given token by adding its ``jti`` to the configured denylist.

    Tokens without a ``jti`` claim, or tokens that fail to decode, are ignored.

    Raises:
        TokenError: When the denylist refuses a new revocation (for example, the
            compatibility in-memory store is at ``max_entries`` with no reclaimable slots).
    """
    try:
        payload = decode_signed_jwt(
            token,
            config=JwtDecodeConfig(
                key=self.verify_key,
                algorithms=[self.algorithm],
                audience=JWT_ACCESS_TOKEN_AUDIENCE,
                options={"verify_exp": False},
                issuer=self.issuer,
            ),
        )
    except InvalidTokenError:
        return

    jti = payload.get("jti")
    exp = payload.get("exp")
    if not isinstance(jti, str):
        return
    ttl_seconds = denylist_ttl_seconds(exp)
    recorded = await self._denylist_store.deny(jti, ttl_seconds=ttl_seconds)
    if not recorded:
        msg = (
            "Could not record JWT revocation in the denylist (in-memory store at capacity). "
            "Use RedisJWTDenylistStore or increase max_entries."
        )
        raise TokenError(msg)

@override
async def read_token(
    self,
    token: str | None,
    user_manager: UserManagerProtocol[UP, ID],
) -> UP | None:
    """Decode a JWT token and load its user.

    Returns:
        The matching user, or ``None`` when the token is invalid.
    """
    result = await self._read_verified_user_and_payload(token, user_manager)
    return None if result is None else result[0]

async def read_token_with_context(
    self,
    token: str | None,
    user_manager: UserManagerProtocol[UP, ID],
) -> JWTAuthenticationResult[UP] | None:
    """Decode a JWT token and expose verified organization claim context.

    Returns:
        Resolved user plus optional organization context, or ``None`` when
        the token does not authenticate.
    """
    result = await self._read_verified_user_and_payload(token, user_manager)
    if result is None:
        return None
    user, payload = result

    return JWTAuthenticationResult(user=user, context=JWTContext(organization=_organization_from_payload(payload)))

@override
async def write_token(self, user: UP) -> str:
    """Generate an organization-free JWT token for the provided user.

    Returns:
        The encoded JWT token string.
    """
    return self._build_access_token(user)

async def write_token_for_organization(self, user: UP, organization: str) -> str:
    """Generate a JWT token bound to a verified active organization.

    Callers must verify that ``user`` is a member of ``organization`` before
    invoking this method. The strategy only normalizes and signs the
    organization claim; it performs no membership lookup itself.

    Returns:
        The encoded JWT token string carrying the normalized organization claim.
    """
    return self._build_access_token(user, organization=organization)

async def mark_used(self, *, key_id: str, nonce: str, ttl_seconds: int) -> ApiKeyNonceStoreResult:
    """Atomically record a nonce with ``SET NX PX``.

    Returns:
        Stored/replay outcome for the nonce insert.
    """
    result = await self._redis.set(self._key(key_id, nonce), "1", nx=True, px=max(ttl_seconds, 1) * 1000)
    if result is True:
        return ApiKeyNonceStoreResult(stored=True)
    return ApiKeyNonceStoreResult(stored=False, rejected_as_replay=True)

@override
async def destroy_token(self, token: str, user: UP) -> None:
    """Delete a persisted Redis token."""
    token_key = self._key(token)
    user_id = str(user.id)
    index_key = self._user_index_key(user_id)
    await self.redis.delete(token_key)
    await self.redis.srem(index_key, token_key)

async def has_recent_totp_verification(self, user: UP, session_id: str) -> bool:
    """Return whether a Redis-backed session has a live TOTP step-up marker."""
    return await self.redis.get(self._totp_stepup_key(str(user.id), session_id)) is not None

async def invalidate_all_tokens(self, user: UP) -> None:
    """Delete all Redis-backed tokens associated with the given user.

    This bumps a per-user invalidation epoch before deleting indexed token
    and step-up marker keys, so out-of-index tokens are rejected on their
    next read without requiring a keyspace scan.
    """
    user_id = str(user.id)
    await self.redis.eval(
        _REDIS_INVALIDATE_USER_TOKENS_SCRIPT,
        3,
        self._user_epoch_key(user_id),
        self._user_index_key(user_id),
        self._totp_stepup_index_key(user_id),
    )

async def issue_totp_stepup(self, user: UP, session_id: str, *, ttl_seconds: int) -> None:
    """Store a short-lived TOTP step-up marker for a Redis-backed session."""
    user_id = str(user.id)
    key = self._totp_stepup_key(user_id, session_id)
    index_key = self._totp_stepup_index_key(user_id)
    if ttl_seconds <= 0:
        await self.redis.delete(key)
        await self.redis.srem(index_key, key)
        return
    await self.redis.set(key, "1", ex=ttl_seconds)
    await self.redis.sadd(index_key, key)
    await self.redis.expire(index_key, ttl_seconds)

@override
async def read_token(
    self,
    token: str | None,
    user_manager: UserManagerProtocol[UP, ID],
) -> UP | None:
    """Resolve a user from a Redis-backed token.

    Returns:
        The resolved user when the token exists and decodes successfully,
        otherwise ``None``.
    """
    if token is None:
        return None

    stored_user_id = await self.redis.get(self._key(token))
    if stored_user_id is None:
        return None

    token_epoch, user_id_text = self._decode_token_payload(stored_user_id)
    if token_epoch != await self._current_user_epoch(user_id_text):
        return None

    try:
        user_id = self.subject_decoder(user_id_text) if self.subject_decoder is not None else user_id_text
    except (TypeError, ValueError):
        return None

    return await user_manager.get(cast("ID", user_id))

@override
async def write_token(self, user: UP) -> str:
    """Persist a new opaque token in Redis and return it.

    Returns:
        Newly created opaque token string.
    """
    token, token_key = self._mint_token_key()
    user_id = str(user.id)
    epoch = await self._current_user_epoch(user_id)
    await self.redis.set(
        token_key,
        self._encode_token_payload(epoch=epoch, user_id=user_id),
        ex=self._ttl_seconds,
    )
    index_key = self._user_index_key(user_id)
    await self.redis.sadd(index_key, token_key)
    await self.redis.expire(index_key, self._ttl_seconds)
    return token

async def rotate_refresh_token(
    self,
    refresh_token: str,
    user_manager: UserManagerProtocol[UP, ID],
) -> tuple[UP, str] | None:
    """Consume a refresh token and return the user plus a rotated replacement."""

async def write_refresh_token(self, user: UP) -> str:
    """Issue a refresh token for the provided user."""

@abstractmethod
async def destroy_token(self, token: str, user: UP) -> None:
    """Invalidate a token for the provided user."""

@abstractmethod
async def read_token(self, token: str | None, user_manager: UserManagerProtocol[UP, ID]) -> UP | None:
    """Resolve a user from a token."""

@abstractmethod
async def write_token(self, user: UP) -> str:
    """Issue a token for the provided user."""

async def get(self, user_id: ID) -> UP | None:
    """Return the user for the given identifier."""