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:  # noqa: PLR0911
    """Return the most specific API-key authentication failure code for ``token``."""
    if token == API_KEY_HMAC_SCHEME:
        return await self._classify_signed_failure_code()
    if token is None:
        return ErrorCode.API_KEY_INVALID
    parsed = parse_api_key(token, expected_prefix_env=self.prefix_env, prefix=self.prefix)
    if parsed is None:
        return ErrorCode.API_KEY_INVALID
    api_key = await self.api_key_store.get_by_key_id(parsed.key_id, include_inactive=True)
    if api_key is None or api_key.prefix_env != parsed.prefix_env:
        return ErrorCode.API_KEY_INVALID
    if getattr(api_key, "signing_required", False):
        return ErrorCode.API_KEY_SIGNATURE_INVALID
    if api_key.revoked_at is not None:
        return ErrorCode.API_KEY_REVOKED
    if _expires_in_past(api_key.expires_at):
        return ErrorCode.API_KEY_EXPIRED
    if not api_key_secret_matches(
        stored_digest=api_key.hashed_secret,
        api_key_hash_secret=self._api_key_hash_secret,
        secret=parsed.secret,
    ):
        return ErrorCode.API_KEY_INVALID
    return ErrorCode.API_KEY_INVALID

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

@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.
    """
    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_with_context(  # noqa: PLR0911
    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.
    """
    if token is None:
        return None
    if token == API_KEY_HMAC_SCHEME:
        return await self._read_signed_request(user_manager)

    parsed = parse_api_key(token, expected_prefix_env=self.prefix_env, prefix=self.prefix)
    if parsed is None:
        return None

    api_key = await self.api_key_store.get_by_key_id(parsed.key_id, include_inactive=True)
    if api_key is None or not self._api_key_can_be_verified(api_key, prefix_env=parsed.prefix_env):
        return None
    if getattr(api_key, "signing_required", False):
        return None
    if not api_key_secret_matches(
        stored_digest=api_key.hashed_secret,
        api_key_hash_secret=self._api_key_hash_secret,
        secret=parsed.secret,
    ):
        return None

    user = await user_manager.get(cast("ID", api_key.user_id))
    if user is None:
        return None
    return ApiKeyAuthenticationResult(
        user=user,
        context=ApiKeyContext(
            key_id=api_key.key_id,
            scopes=tuple(api_key.scopes),
            prefix_env=api_key.prefix_env,
            scope_subset_check=self.scope_subset_check,
            scope_authority=self.scope_authority,
        ),
    )

@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.
    """
    del user
    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,
    )

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),
    )
    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."""
    del user
    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 has_recent_totp_verification(self, user: UP, session_id: str) -> bool:
    """Return whether a DB-backed refresh session has a live TOTP step-up marker."""
    row = await self._load_refresh_session_row(user, session_id)
    if row is None:
        return False
    metadata = row.client_metadata or {}
    try:
        expires_at = float(metadata.get(_CLIENT_METADATA_TOTP_STEPUP_EXPIRES_AT_KEY, ""))
    except ValueError:
        return False
    if expires_at > datetime.now(tz=UTC).timestamp():
        return True
    metadata = dict(metadata)
    metadata.pop(_CLIENT_METADATA_TOTP_STEPUP_EXPIRES_AT_KEY, None)
    row.client_metadata = metadata or None
    await self.session.commit()
    return False

async def identify_refresh_session(self, user: UP, refresh_token: str) -> str | None:
    """Return the public refresh-session id for ``refresh_token`` when it belongs to ``user``.

    Returns:
        Public refresh-session id, or ``None`` when the token is missing, expired, or not owned by ``user``.
    """
    persisted_token = cast(
        "_RefreshTokenRow | None",
        await self._resolve_token(
            self._repository(self._refresh_token_repository_type),
            refresh_token,
            load=[],
        ),
    )
    if persisted_token is None or persisted_token.user_id != user.id:
        return None
    if self._is_token_expired(persisted_token.created_at, self.refresh_max_age):
        await self._delete_refresh_token_row(persisted_token)
        await self.session.commit()
        return None
    return persisted_token.session_id

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._repository(self._refresh_token_repository_type).delete_where(user_id=user.id, auto_commit=False)
    await self.session.commit()

async def issue_totp_stepup(self, user: UP, session_id: str, *, ttl_seconds: int) -> None:
    """Store a short-lived TOTP step-up marker on a DB-backed refresh session."""
    row = await self._load_refresh_session_row(user, session_id)
    if row is None:
        return
    metadata = dict(row.client_metadata or {})
    if ttl_seconds <= 0:
        metadata.pop(_CLIENT_METADATA_TOTP_STEPUP_EXPIRES_AT_KEY, None)
    else:
        expires_at = datetime.now(tz=UTC).timestamp() + ttl_seconds
        metadata[_CLIENT_METADATA_TOTP_STEPUP_EXPIRES_AT_KEY] = str(expires_at)
    row.client_metadata = metadata or None
    await self.session.commit()

async def list_refresh_sessions(self, user: UP) -> list[RefreshSession]:
    """Return active refresh sessions belonging to ``user``.

    Expired refresh-token rows are deleted before active sessions are returned.

    Returns:
        Active refresh-session metadata ordered by creation time.
    """
    expired_count = await self._delete_expired_refresh_sessions_for_user(user)
    if expired_count:
        await self.session.commit()

    result = await self.session.execute(
        select(self.refresh_token_model)
        .where(self.refresh_token_model.user_id == user.id)
        .order_by(self.refresh_token_model.created_at),
    )
    rows = cast("list[_RefreshTokenRow]", result.scalars().all())
    return [self._refresh_session_from_row(row) for row in rows]

@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``.
    """
    del user_manager
    if token is None:
        return None

    access_token = cast(
        "Any",
        await self._resolve_token(
            self._repository(self._access_token_repository_type),
            token,
            load=[self.access_token_model.user],
        ),
    )
    if access_token is None or self._is_token_expired(access_token.created_at, self.max_age):
        return None

    return cast("UP", access_token.user)

async def revoke_other_refresh_sessions(self, user: UP, current_session_id: str | None) -> int:
    """Revoke active refresh sessions for ``user`` except ``current_session_id``.

    Returns:
        Number of active refresh sessions revoked.
    """
    expired_count = await self._delete_expired_refresh_sessions_for_user(user)
    conditions = [self.refresh_token_model.user_id == user.id]
    if current_session_id is not None:
        conditions.append(self.refresh_token_model.session_id != current_session_id)
    deleted_count = await self._execute_delete(delete(self.refresh_token_model).where(*conditions))
    if expired_count or deleted_count:
        await self.session.commit()
    return deleted_count

async def revoke_refresh_session(self, user: UP, session_id: str) -> bool:
    """Revoke one active refresh session for ``user`` by public session id.

    Returns:
        ``True`` when a matching active session was deleted, otherwise ``False``.
    """
    expired_count = await self._delete_expired_refresh_sessions_for_user(user)
    deleted_count = await self._execute_delete(
        delete(self.refresh_token_model).where(
            self.refresh_token_model.user_id == user.id,
            self.refresh_token_model.session_id == session_id,
        ),
    )
    if expired_count or deleted_count:
        await self.session.commit()
    return deleted_count > 0

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

    Returns:
        Tuple of the resolved user and rotated refresh token, or ``None`` when invalid.
    """
    del user_manager
    client_metadata = self._consume_refresh_token_request_metadata()
    persisted_token = await self._load_refresh_token_for_rotation(refresh_token)
    if persisted_token is None:
        return None
    if self._is_token_expired(persisted_token.created_at, self.refresh_max_age):
        await self._delete_refresh_token_row(persisted_token)
        return None

    user = cast("UP", persisted_token.user)
    await self._delete_refresh_token_row(persisted_token)
    rotated_refresh_token = await self._mint_replacement_refresh_token(
        user,
        persisted_token,
        client_metadata=client_metadata,
    )
    return user, rotated_refresh_token

def set_refresh_token_request_context(self, request: object) -> None:
    """Capture safe request metadata for the next refresh-token write or rotation."""
    self._refresh_token_request_metadata = self._extract_refresh_token_client_metadata(request)

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).
    """
    del user

    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 = 1
    if isinstance(exp, int):
        ttl_seconds = max(exp - int(time.time()), 1)
    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(  # noqa: PLR0911
    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.
    """
    if token is None:
        return None

    payload = self._decode_verified_access_token(token)
    if payload is None:
        logger.info("JWT decode failed (expired or invalid signature)")
        return None

    if await self._is_token_denied(payload):
        logger.info("JWT denied (revoked)")
        return None

    subject = payload.get("sub")
    if not isinstance(subject, str) or not subject:
        logger.info("JWT missing or invalid 'sub' claim")
        return None

    try:
        user_id = self.subject_decoder(subject) if self.subject_decoder is not None else subject
    except ValueError:
        # Security: avoid logging the subject itself to prevent user enumeration
        # via authentication-failure log analysis (OWASP / NIST SP 800-63B §5.2.2).
        logger.info("JWT subject could not be decoded")
        return None

    if user_id is None:
        return None

    user = await user_manager.get(cast("ID", user_id))
    if user is None:
        logger.info("JWT subject references non-existent user")
        return None

    if not self._validate_fingerprint(payload, user):
        logger.info("JWT fingerprint mismatch")
        return None

    return user

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

    Returns:
        The encoded JWT token string.
    """
    issued_at = datetime.now(tz=UTC)
    payload = {
        "sub": str(user.id),
        "aud": JWT_ACCESS_TOKEN_AUDIENCE,
        "iat": issued_at,
        "nbf": issued_at,
        "exp": issued_at + self.lifetime,
        "jti": secrets.token_hex(16),
    }
    if self.issuer is not None:
        payload["iss"] = self.issuer

    fingerprint = self.session_fingerprint_getter(user)
    if fingerprint is not None:
        payload[self.session_fingerprint_claim] = fingerprint
    return jwt.encode(payload, self.secret, algorithm=self.algorithm, headers=jwt_encode_headers())

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)