124 lines
No EOL
4.3 KiB
Python
124 lines
No EOL
4.3 KiB
Python
|
|
from django.contrib.auth import get_user_model
|
|
from django.utils import timezone
|
|
from django.http import HttpRequest
|
|
|
|
from rest_framework.authentication import SessionAuthentication, get_authorization_header
|
|
|
|
from .models import PersonalAccessToken
|
|
|
|
|
|
# === AUTH HYBRIDE (JWT court OU PAT opaque) ===
|
|
# - JWT: via djangorestframework-simplejwt (RS256)
|
|
# - PAT: tokens opaques en DB (lookup + expiration + revoke)
|
|
|
|
User = get_user_model()
|
|
|
|
# Import JWTAuthentication uniquement si simplejwt est installé/configuré
|
|
try:
|
|
from rest_framework_simplejwt.authentication import JWTAuthentication
|
|
|
|
_jwt_auth = JWTAuthentication()
|
|
except Exception: # pragma: no cover
|
|
_jwt_auth = None
|
|
|
|
|
|
def collections_for(user) -> list[str]:
|
|
"""
|
|
Mapping des permissions Django -> collections accessibles via token.
|
|
Inclut les collections pygeoapi et les endpoints de reporting.
|
|
"""
|
|
collections = ["technical_galleries"]
|
|
|
|
# Accès reporting : superuser ou can_access_view('reporting')
|
|
try:
|
|
from common.models import UserConfig
|
|
user_config = UserConfig.objects.get(user=user)
|
|
if user_config.can_access_view("reporting"):
|
|
collections.append("reporting")
|
|
except Exception:
|
|
pass
|
|
|
|
if user.is_superuser:
|
|
if "reporting" not in collections:
|
|
collections.append("reporting")
|
|
|
|
return collections
|
|
|
|
|
|
class HybridTokenAuthentication(SessionAuthentication):
|
|
"""
|
|
Authentifie les requêtes API de 3 façons:
|
|
1) Session (cookie) — via SessionAuthentication (hérité)
|
|
2) Authorization: Bearer <JWT | PAT>
|
|
3) ?token=<JWT | PAT>
|
|
|
|
Remplit request.collections (list[str]) quand un JWT/PAT est utilisé.
|
|
"""
|
|
|
|
def authenticate(self, request: HttpRequest):
|
|
# 1) D'abord, si un header Bearer / query ?token est présent, tenter JWT/PAT
|
|
raw_token: str | None = None
|
|
auth = get_authorization_header(request).decode("utf-8")
|
|
if auth.startswith("Bearer "):
|
|
raw_token = auth.split(" ", 1)[1]
|
|
else:
|
|
raw_token = request.GET.get("token")
|
|
|
|
# Si pas trouvé dans header/query, tenter depuis le chemin: /api/geo/pat_xxx/...
|
|
if not raw_token:
|
|
path = request.path or ""
|
|
# On cherche un segment 'pat_' immédiatement après /api/geo/
|
|
# Exemple: /api/geo/pat_abCDe.../collections/technical_galleries/items
|
|
try:
|
|
after_geo = path.split("/geo/", 1)[1] # peut lever IndexError
|
|
first_seg = after_geo.split("/", 1)[0]
|
|
if first_seg.startswith("pat_"):
|
|
raw_token = first_seg
|
|
except Exception:
|
|
pass
|
|
|
|
if raw_token:
|
|
# Essai JWT
|
|
if _jwt_auth is not None:
|
|
try:
|
|
validated = _jwt_auth.get_validated_token(raw_token)
|
|
user = _jwt_auth.get_user(validated)
|
|
request.collections = validated.get("collections", [])
|
|
return (user, validated)
|
|
except Exception:
|
|
pass
|
|
|
|
# Essai PAT
|
|
pat = PersonalAccessToken.lookup(raw_token)
|
|
if pat is None:
|
|
# Échec explicite si un token a été fourni mais est invalide
|
|
return None
|
|
request.collections = pat.collections
|
|
return (pat.user, None)
|
|
|
|
# 2) Sinon, retomber sur la session (auth interactive)
|
|
result = super().authenticate(request)
|
|
if result is not None:
|
|
# Pour les vues geo, on peut projeter les collections depuis la session
|
|
user, x = result
|
|
request.collections = collections_for(user)
|
|
return result
|
|
|
|
|
|
# === JWT issue utilitaire ===
|
|
# (JWT court, claims minimales, signé RS256 via SimpleJWT)
|
|
try:
|
|
from rest_framework_simplejwt.tokens import AccessToken
|
|
|
|
def mint_jwt(user, lifetime_minutes: int = 20) -> str:
|
|
t = AccessToken()
|
|
t["sub"] = str(user.pk)
|
|
t["aud"] = "pygeoapi"
|
|
t["collections"] = collections_for(user)
|
|
t.set_exp(from_time=timezone.now(), lifetime=timezone.timedelta(minutes=int(lifetime_minutes)))
|
|
return str(t)
|
|
|
|
except Exception: # pragma: no cover
|
|
def mint_jwt(user, lifetime_minutes: int = 20) -> str: # fallback clair
|
|
raise RuntimeError("SimpleJWT n'est pas installé/configuré") |