Strategies¶

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")
    _validate_token_model_contract(self.refresh_token_model, field_name="refresh_token_model")

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().delete_where(token=token_digest, auto_commit=False)
    if self.accept_legacy_plaintext_tokens:
        await self._repository().delete_where(token=token, 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().delete_where(user_id=user.id, auto_commit=False)
    await self._refresh_repository().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``.
    """
    del user_manager
    if token is None:
        return None

    access_token = cast(
        "Any",
        await self._resolve_token(
            self._repository(),
            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)

@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
    persisted_token = cast(
        "Any",
        await self._resolve_token(
            self._refresh_repository(),
            refresh_token,
            load=[self.refresh_token_model.user],
        ),
    )
    if persisted_token is None:
        return None
    if self._is_token_expired(persisted_token.created_at, self.refresh_max_age):
        await self._refresh_repository().delete_where(token=persisted_token.token, auto_commit=False)
        return None

    user = cast("UP", persisted_token.user)
    await self._refresh_repository().delete_where(token=persisted_token.token, auto_commit=False)
    rotated_refresh_token = secrets.token_urlsafe(self.token_bytes)
    rotated_model = self.refresh_token_model(token=self._token_digest(rotated_refresh_token), user_id=user.id)
    await self._refresh_repository().add(rotated_model, auto_commit=False, auto_refresh=True)
    return user, rotated_refresh_token

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,
        accept_legacy_plaintext_tokens=self.accept_legacy_plaintext_tokens,
    )

@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 = secrets.token_urlsafe(self.token_bytes)
    refresh_token = self.refresh_token_model(token=self._token_digest(token), user_id=user.id)
    await self._refresh_repository().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 = secrets.token_urlsafe(self.token_bytes)
    access_token = self.access_token_model(token=self._token_digest(token), user_id=user.id)
    await self._repository().add(access_token, auto_refresh=True)
    return token

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

    The denylist is maintained in memory on the strategy instance. Tokens
    without a ``jti`` claim, or tokens that fail to decode, are ignored.
    """
    del user

    try:
        payload = jwt.decode(
            token,
            self.verify_key,
            algorithms=[self.algorithm],
            audience=JWT_ACCESS_TOKEN_AUDIENCE,
            options={
                "verify_exp": False,
                "verify_iss": False,
            },
        )
    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)
    await self._denylist_store.deny(jti, ttl_seconds=ttl_seconds)

@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:
        logger.info("JWT subject could not be decoded: %s", subject)
        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: %s", subject)
        return None

    if not self._validate_fingerprint(payload, user):
        logger.info("JWT fingerprint mismatch for user: %s", subject)
        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)

@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 invalidate_all_tokens(self, user: UP) -> None:
    """Delete all Redis-backed tokens associated with the given user.

    This uses a per-user index to delete only the keys associated with the
    user, avoiding keyspace scans under the global prefix.

    Backward compatibility:
        If the per-user index is missing, we fall back to a best-effort scan
        over keys matching the configured prefix and remove those whose
        stored subject matches ``user.id``.
    """
    user_id = str(user.id)
    if await self._invalidate_via_index(user_id):
        return
    await self._invalidate_via_scan(user_id)

@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

    user_id_text = self._decode_user_id(stored_user_id)

    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 = secrets.token_urlsafe(self.token_bytes)
    token_key = self._key(token)
    user_id = str(user.id)
    await self.redis.setex(token_key, self._ttl_seconds, user_id)
    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."""