TOTP¶

async def mark_used(self, user_id: Hashable, counter: int, ttl_seconds: float) -> bool:
    """Store a used code pair until its TTL elapses.

    Returns:
        ``True`` when the pair was stored, otherwise ``False`` for a replay.
    """
    async with self._lock:
        now = self._clock()
        self._prune(now)
        key = (user_id, counter)
        if key in self._entries:
            return False

        if len(self._entries) >= self.max_entries:
            self._prune(now)
            self._evict_oldest_until_below_cap()

        self._entries[key] = now + ttl_seconds
        return True

async def mark_used(self, user_id: Hashable, counter: int, ttl_seconds: float) -> bool:
    """Atomically record a used (user_id, counter) pair via SET key 1 NX PX ttl_ms.

    Returns:
        ``True`` when the pair was newly stored, ``False`` for a replay.
    """
    key = self._key(user_id, counter)
    ttl_ms = int(ttl_seconds * 1000)
    result = await self._redis.set(key, "1", nx=True, px=ttl_ms)
    return result is True

async def set(
    self,
    name: str,
    value: str,
    *,
    nx: bool = False,
    px: int | None = None,
) -> bool | None:
    """Set a key; with ``nx=True`` only set if not exists; ``px`` is TTL in milliseconds.

    Returns:
        ``True`` when the key was set, ``False`` when ``nx=True`` and key already existed.
    """

async def mark_used(self, user_id: Hashable, counter: int, ttl_seconds: float) -> bool:
    """Atomically record a `(user_id, counter)` pair when unused.

    Returns:
        ``True`` when the pair was newly stored, otherwise ``False``.
    """

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

async def read_totp_secret(self, secret: str | None) -> str | None:
    """Return a plain-text TOTP secret from storage."""

async def authenticate_pending_login(
    self,
    *,
    pending_token: str,
    code: str,
    validate_user: PendingUserValidator[UP] | None = None,
) -> UP:
    """Validate a pending-login token plus TOTP code and return the resolved user.

    Returns:
        The user resolved from the verified pending-login challenge.

    Raises:
        InvalidTotpCodeError: If the TOTP code is invalid or TOTP is not enabled.
    """
    pending_login = await self._resolve_pending_login(pending_token)
    if validate_user is not None:
        await validate_user(pending_login.user)
    secret = await self._user_manager.read_totp_secret(pending_login.user.totp_secret)
    if not secret or not await verify_totp_with_store(
        secret,
        code,
        user_id=pending_login.user.id,
        used_tokens_store=self._used_tokens_store,
        algorithm=self._totp_algorithm,
        require_replay_protection=self._require_replay_protection,
    ):
        raise InvalidTotpCodeError
    await self._deny_pending_login(pending_login)
    return pending_login.user

async def issue_pending_token(self, user: UP) -> str | None:
    """Return a pending-login JWT when the user has TOTP enabled."""
    if await self._user_manager.read_totp_secret(user.totp_secret) is None:
        return None
    issued_at = datetime.now(tz=UTC)
    payload = {
        "sub": str(user.id),
        "aud": TOTP_PENDING_AUDIENCE,
        "iat": issued_at,
        "nbf": issued_at,
        "exp": issued_at + self._totp_pending_lifetime,
        "jti": secrets.token_hex(16),
    }
    return jwt.encode(payload, self._totp_pending_secret, algorithm="HS256")