226 lines
8.6 KiB
Python
226 lines
8.6 KiB
Python
import hashlib
|
||
from dataclasses import dataclass
|
||
from datetime import UTC, datetime
|
||
from pathlib import Path
|
||
|
||
import jwt
|
||
from cryptography.hazmat.primitives import serialization
|
||
from cryptography.hazmat.primitives.asymmetric import ec
|
||
from django.conf import settings
|
||
from django.contrib.auth import get_user_model
|
||
from django.core.cache import cache
|
||
from django.core.exceptions import ValidationError as DjangoValidationError
|
||
from django.core.validators import validate_email
|
||
from django.db import transaction
|
||
from django.db.models import Q
|
||
from django.utils import timezone
|
||
from jwt import InvalidTokenError
|
||
|
||
from .models import SsoIdentity
|
||
|
||
User = get_user_model()
|
||
|
||
|
||
class SsoAuthenticationError(Exception):
|
||
"""Безопасная ошибка проверки или связывания SSO-пользователя."""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SsoClaims:
|
||
issuer: str
|
||
subject: str
|
||
username: str
|
||
email: str
|
||
first_name: str
|
||
middle_name: str
|
||
last_name: str
|
||
jti: str
|
||
expires_at: int
|
||
|
||
|
||
def _load_public_key():
|
||
key_path = Path(settings.SSO_PUBLIC_KEY_PATH)
|
||
try:
|
||
key_data = key_path.read_bytes()
|
||
public_key = serialization.load_pem_public_key(key_data)
|
||
except (OSError, ValueError, TypeError) as exc:
|
||
raise SsoAuthenticationError("Публичный ключ SSO недоступен.") from exc
|
||
|
||
if not isinstance(public_key, ec.EllipticCurvePublicKey) or not isinstance(
|
||
public_key.curve, ec.SECP256R1
|
||
):
|
||
raise SsoAuthenticationError("Публичный ключ SSO должен использовать P-256.")
|
||
return public_key
|
||
|
||
|
||
def _decode_sso_payload(encoded_token: str) -> dict:
|
||
try:
|
||
header = jwt.get_unverified_header(encoded_token)
|
||
if header.get("alg") != "ES256":
|
||
raise SsoAuthenticationError("Недопустимый алгоритм SSO-токена.")
|
||
|
||
return jwt.decode(
|
||
encoded_token,
|
||
_load_public_key(),
|
||
algorithms=["ES256"],
|
||
issuer=settings.SSO_ISSUER,
|
||
leeway=settings.SSO_CLOCK_SKEW_SECONDS,
|
||
options={
|
||
"require": [
|
||
"iss",
|
||
"exp",
|
||
"iat",
|
||
"jti",
|
||
"token_type",
|
||
"sso_target",
|
||
"user_id",
|
||
"username",
|
||
"email",
|
||
]
|
||
},
|
||
)
|
||
except SsoAuthenticationError:
|
||
raise
|
||
except (InvalidTokenError, ValueError, TypeError) as exc:
|
||
raise SsoAuthenticationError("SSO-токен не прошёл проверку.") from exc
|
||
|
||
|
||
def _validate_token_lifetime(payload: dict) -> tuple[int, int]:
|
||
issued_at = payload.get("iat")
|
||
expires_at = payload.get("exp")
|
||
if (
|
||
not isinstance(issued_at, int)
|
||
or isinstance(issued_at, bool)
|
||
or not isinstance(expires_at, int)
|
||
or isinstance(expires_at, bool)
|
||
):
|
||
raise SsoAuthenticationError("Некорректное время жизни SSO-токена.")
|
||
if expires_at <= issued_at or (
|
||
expires_at - issued_at > settings.SSO_TOKEN_MAX_AGE_SECONDS
|
||
):
|
||
raise SsoAuthenticationError("Недопустимое время жизни SSO-токена.")
|
||
return issued_at, expires_at
|
||
|
||
|
||
def verify_sso_token(encoded_token: str) -> SsoClaims:
|
||
"""Проверяет ES256 SSO JWT и возвращает только нужные локальному сервису claims."""
|
||
payload = _decode_sso_payload(encoded_token)
|
||
|
||
if payload.get("token_type") != "sso":
|
||
raise SsoAuthenticationError("Недопустимый тип SSO-токена.")
|
||
if payload.get("sso_target") != settings.SSO_SERVICE_CODE:
|
||
raise SsoAuthenticationError("SSO-токен предназначен другому сервису.")
|
||
|
||
_, expires_at = _validate_token_lifetime(payload)
|
||
|
||
username = payload.get("username")
|
||
email = payload.get("email")
|
||
jti = payload.get("jti")
|
||
subject = payload.get("user_id")
|
||
if not isinstance(username, str) or not username.strip():
|
||
raise SsoAuthenticationError("В SSO-токене отсутствует username.")
|
||
if not isinstance(email, str) or not email.strip():
|
||
raise SsoAuthenticationError("В SSO-токене отсутствует email.")
|
||
if not isinstance(jti, str) or not jti.strip():
|
||
raise SsoAuthenticationError("В SSO-токене отсутствует jti.")
|
||
if isinstance(subject, bool) or not isinstance(subject, int | str):
|
||
raise SsoAuthenticationError("В SSO-токене отсутствует user_id.")
|
||
if not str(subject).strip():
|
||
raise SsoAuthenticationError("В SSO-токене отсутствует user_id.")
|
||
|
||
try:
|
||
validate_email(email)
|
||
User._meta.get_field("username").run_validators(username)
|
||
except DjangoValidationError as exc:
|
||
raise SsoAuthenticationError(
|
||
"Некорректные данные пользователя в SSO-токене."
|
||
) from exc
|
||
|
||
return SsoClaims(
|
||
issuer=str(payload["iss"]),
|
||
subject=str(subject),
|
||
username=username,
|
||
email=email,
|
||
first_name=str(payload.get("first_name") or ""),
|
||
middle_name=str(payload.get("surname") or ""),
|
||
last_name=str(payload.get("last_name") or ""),
|
||
jti=jti,
|
||
expires_at=expires_at,
|
||
)
|
||
|
||
|
||
def reserve_sso_token(claims: SsoClaims) -> None:
|
||
"""Атомарно помечает одноразовый SSO-токен использованным."""
|
||
digest = hashlib.sha256(f"{claims.issuer}\0{claims.jti}".encode()).hexdigest()
|
||
cache_key = f"sso:jti:{digest}"
|
||
now = int(datetime.now(tz=UTC).timestamp())
|
||
timeout = max(1, claims.expires_at - now + settings.SSO_CLOCK_SKEW_SECONDS)
|
||
if not cache.add(cache_key, "used", timeout=timeout):
|
||
raise SsoAuthenticationError("SSO-токен уже использован.")
|
||
|
||
|
||
def _ensure_user_claims_available(claims: SsoClaims, *, exclude_user_id=None) -> None:
|
||
conflicts = User.objects.filter(
|
||
Q(username=claims.username) | Q(email__iexact=claims.email)
|
||
)
|
||
if exclude_user_id is not None:
|
||
conflicts = conflicts.exclude(id=exclude_user_id)
|
||
if conflicts.exists():
|
||
raise SsoAuthenticationError("Данные SSO-пользователя уже заняты.")
|
||
|
||
|
||
@transaction.atomic
|
||
def provision_sso_user(claims: SsoClaims):
|
||
"""Создаёт или обновляет локального пользователя по неизменяемой SSO identity."""
|
||
identity = (
|
||
SsoIdentity.objects.select_for_update()
|
||
.select_related("user", "user__profile")
|
||
.filter(issuer=claims.issuer, subject=claims.subject)
|
||
.first()
|
||
)
|
||
|
||
if identity is not None:
|
||
user = identity.user
|
||
if not user.is_active:
|
||
raise SsoAuthenticationError("Локальная учётная запись отключена.")
|
||
_ensure_user_claims_available(claims, exclude_user_id=user.id)
|
||
else:
|
||
username_user = User.objects.filter(username=claims.username).first()
|
||
email_users = list(User.objects.filter(email__iexact=claims.email)[:2])
|
||
if username_user is not None or email_users:
|
||
if (
|
||
username_user is None
|
||
or len(email_users) != 1
|
||
or email_users[0].id != username_user.id
|
||
):
|
||
raise SsoAuthenticationError(
|
||
"Неоднозначное совпадение локальной учётной записи."
|
||
)
|
||
user = username_user
|
||
if not user.is_active:
|
||
raise SsoAuthenticationError("Локальная учётная запись отключена.")
|
||
else:
|
||
user = User(username=claims.username, email=claims.email, is_verified=True)
|
||
user.set_unusable_password()
|
||
user.save()
|
||
|
||
identity = SsoIdentity.objects.create(
|
||
user=user,
|
||
issuer=claims.issuer,
|
||
subject=claims.subject,
|
||
last_authenticated_at=timezone.now(),
|
||
)
|
||
|
||
user.username = claims.username
|
||
user.email = claims.email
|
||
user.is_verified = True
|
||
user.save(update_fields=["username", "email", "is_verified", "updated_at"])
|
||
profile = user.profile
|
||
profile.first_name = claims.first_name
|
||
profile.mid_name = claims.middle_name
|
||
profile.last_name = claims.last_name
|
||
profile.save(update_fields=["first_name", "mid_name", "last_name", "updated_at"])
|
||
identity.last_authenticated_at = timezone.now()
|
||
identity.save(update_fields=["last_authenticated_at", "updated_at"])
|
||
return user
|