98 lines
3 KiB
Python
98 lines
3 KiB
Python
"""Bearer-token auth for machine clients.
|
|
|
|
Three carriers are recognised:
|
|
|
|
1. ``Authorization: Bearer <token>`` header.
|
|
2. ``?api_token=<token>`` query string.
|
|
3. A token the visitor entered in a browser form, stashed in the session. This
|
|
lets token-gated HTML pages remember the credential without a login system.
|
|
|
|
API controllers use :func:`request_has_valid_token` (carriers 1 and 2 only);
|
|
browser pages use :func:`resolve_token`, which also accepts the session.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from django.http import HttpRequest
|
|
|
|
from .models import ApiToken, TokenKind
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_HEADER = "Authorization"
|
|
_QUERY_PARAM = "api_token"
|
|
_BEARER_PREFIX = "bearer "
|
|
|
|
SESSION_TOKEN_KEY = "api_token"
|
|
|
|
|
|
def validate_raw_token(raw: str | None) -> ApiToken | None:
|
|
raw = (raw or "").strip()
|
|
if not raw:
|
|
return None
|
|
token = ApiToken.objects.filter(token=raw, is_active=True).first()
|
|
if token is None:
|
|
logger.info("api: unknown or revoked token presented")
|
|
return None
|
|
try:
|
|
token.mark_used()
|
|
except Exception: # noqa: BLE001
|
|
logger.warning("api: could not update last_used_at for token %s", token.name)
|
|
return token
|
|
|
|
|
|
def extract_token(request: HttpRequest) -> ApiToken | None:
|
|
return validate_raw_token(_read_token_value(request))
|
|
|
|
|
|
def request_has_valid_token(request: HttpRequest) -> bool:
|
|
return extract_token(request) is not None
|
|
|
|
|
|
def request_can_ingest(request: HttpRequest) -> bool:
|
|
"""True only for a valid INGEST token (push endpoints)."""
|
|
token = extract_token(request)
|
|
return token is not None and token.kind == TokenKind.INGEST
|
|
|
|
|
|
def store_session_token(request: HttpRequest, token: ApiToken) -> None:
|
|
request.session[SESSION_TOKEN_KEY] = token.token
|
|
|
|
|
|
def clear_session_token(request: HttpRequest) -> None:
|
|
request.session.pop(SESSION_TOKEN_KEY, None)
|
|
|
|
|
|
def token_from_session(request: HttpRequest) -> ApiToken | None:
|
|
if not hasattr(request, "session"):
|
|
return None
|
|
return validate_raw_token(request.session.get(SESSION_TOKEN_KEY))
|
|
|
|
|
|
def resolve_token(request: HttpRequest, *, allow_session: bool = True) -> ApiToken | None:
|
|
token = extract_token(request)
|
|
if token is not None:
|
|
return token
|
|
if allow_session:
|
|
return token_from_session(request)
|
|
return None
|
|
|
|
|
|
def request_is_authenticated(request: HttpRequest, *, allow_session: bool = True) -> bool:
|
|
return resolve_token(request, allow_session=allow_session) is not None
|
|
|
|
|
|
def _read_token_value(request: HttpRequest) -> str | None:
|
|
header = request.headers.get(_HEADER, "").strip()
|
|
if header:
|
|
if header.lower().startswith(_BEARER_PREFIX):
|
|
value = header[len(_BEARER_PREFIX):].strip()
|
|
if value:
|
|
return value
|
|
# Accept the raw value too if it's the only thing in the header.
|
|
elif " " not in header:
|
|
return header
|
|
qs_value = request.GET.get(_QUERY_PARAM, "").strip()
|
|
return qs_value or None
|