Redis contrib¶

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)

def __post_init__(self) -> None:
    """Snapshot group-specific rate-limit tiers into a read-only mapping."""
    self.group_rate_limit_tiers = MappingProxyType(dict(self.group_rate_limit_tiers))

def build_rate_limit_config(
    self,
    *,
    options: RedisAuthRateLimitConfigOptions | None = None,
) -> AuthRateLimitConfig:
    """Build ``AuthRateLimitConfig`` from the preset's shared Redis client.

    Args:
        options: Optional shared builder options. When omitted, every
            supported auth slot uses the preset's default tier.

    Returns:
        The auth rate-limit config built from the preset's shared client and
        tier settings.
    """
    resolved_options = options or RedisAuthRateLimitConfigOptions()
    derived_group_backends: dict[AuthRateLimitEndpointGroup, RateLimiterBackend] = {
        group: self._build_rate_limit_backend(tier) for group, tier in self.group_rate_limit_tiers.items()
    }
    if resolved_options.group_backends is not None:
        derived_group_backends.update(resolved_options.group_backends)
    shared_backend: RateLimiterBackend = self._build_rate_limit_backend(self.rate_limit_tier)
    resolved_group_backends = derived_group_backends or None
    default_shared_options = SharedRateLimitConfigOptions()

    shared_options = SharedRateLimitConfigOptions(
        enabled=resolved_options.enabled,
        disabled=resolved_options.disabled,
        group_backends=resolved_group_backends,
        endpoint_overrides=resolved_options.endpoint_overrides,
        trusted_proxy=resolved_options.trusted_proxy,
        identity_fields=(
            resolved_options.identity_fields
            if resolved_options.identity_fields is not None
            else default_shared_options.identity_fields
        ),
        trusted_headers=(
            resolved_options.trusted_headers
            if resolved_options.trusted_headers is not None
            else default_shared_options.trusted_headers
        ),
        trusted_proxy_hops=resolved_options.trusted_proxy_hops,
    )
    return AuthRateLimitConfig.from_shared_backend(shared_backend, options=shared_options)

def build_totp_enrollment_store(self, *, key_prefix: str | None = None) -> RedisTotpEnrollmentStore:
    """Build ``RedisTotpEnrollmentStore`` from the preset's shared Redis client.

    Args:
        key_prefix: Optional per-call Redis key prefix override. When
            omitted, the preset uses ``totp_enrollment_key_prefix`` and
            finally falls back to the current ``RedisTotpEnrollmentStore``
            default.

    Returns:
        Redis-backed pending-enrollment store sharing the preset's client.
    """
    resolved_key_prefix = self.totp_enrollment_key_prefix if key_prefix is None else key_prefix
    if resolved_key_prefix is None:
        resolved_key_prefix = DEFAULT_TOTP_ENROLLMENT_KEY_PREFIX
    return RedisTotpEnrollmentStore(redis=self.redis, key_prefix=resolved_key_prefix)

def build_totp_pending_jti_store(self, *, key_prefix: str | None = None) -> RedisJWTDenylistStore:
    """Build ``RedisJWTDenylistStore`` from the preset's shared Redis client.

    Args:
        key_prefix: Optional per-call Redis key prefix override. When
            omitted, the preset uses ``totp_pending_jti_key_prefix`` and
            finally falls back to the current ``RedisJWTDenylistStore``
            default.

    Returns:
        Redis-backed pending-token JTI denylist sharing the preset's
        client.
    """
    resolved_key_prefix = self.totp_pending_jti_key_prefix if key_prefix is None else key_prefix
    if resolved_key_prefix is None:
        return RedisJWTDenylistStore(redis=self.redis)
    return RedisJWTDenylistStore(redis=self.redis, key_prefix=resolved_key_prefix)

def build_totp_used_tokens_store(self, *, key_prefix: str | None = None) -> RedisUsedTotpCodeStore:
    """Build ``RedisUsedTotpCodeStore`` from the preset's shared Redis client.

    Args:
        key_prefix: Optional per-call Redis key prefix override. When
            omitted, the preset uses ``totp_used_tokens_key_prefix`` and
            finally falls back to the current ``RedisUsedTotpCodeStore``
            default.

    Returns:
        Redis-backed TOTP replay store sharing the preset's client.
    """
    resolved_key_prefix = self.totp_used_tokens_key_prefix if key_prefix is None else key_prefix
    if resolved_key_prefix is None:
        resolved_key_prefix = DEFAULT_TOTP_USED_KEY_PREFIX
    return RedisUsedTotpCodeStore(redis=self.redis, key_prefix=resolved_key_prefix)

@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