Игорь Надеин
86fed54811
All checks were successful
Deploy production / deploy (push) Successful in 10m4s
95 lines
3.2 KiB
Python
95 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
from urllib.parse import urljoin
|
|
|
|
from django.conf import settings
|
|
from oauth2_provider.oauth2_validators import OAuth2Validator
|
|
|
|
|
|
def _preferred_username(user) -> str:
|
|
username = (getattr(user, "username", "") or "").strip()
|
|
return username or f"employee-{user.pk}"
|
|
|
|
|
|
def _is_active_employee(user) -> bool:
|
|
if not user or not getattr(user, "is_authenticated", False):
|
|
return False
|
|
if not getattr(user, "is_active", False):
|
|
return False
|
|
if not getattr(user, "email_verified", False):
|
|
return False
|
|
return bool(getattr(user, "is_actually_active", True))
|
|
|
|
|
|
def _employee_groups(user) -> list[str]:
|
|
groups = {"portal-users"}
|
|
groups.update(user.groups.values_list("name", flat=True))
|
|
if getattr(user, "is_staff", False):
|
|
groups.add("portal-staff")
|
|
if getattr(user, "is_superuser", False):
|
|
groups.add("portal-admins")
|
|
return sorted(groups)
|
|
|
|
|
|
def _picture_url(user) -> str | None:
|
|
avatar = getattr(user, "avatar", None)
|
|
if not avatar:
|
|
return None
|
|
try:
|
|
return urljoin(f"{settings.SITE_URL}/", avatar.url.lstrip("/"))
|
|
except (ValueError, AttributeError):
|
|
return None
|
|
|
|
|
|
class CorporateOAuth2Validator(OAuth2Validator):
|
|
"""Expose only verified corporate identity claims to OIDC clients."""
|
|
|
|
oidc_claim_scope = {
|
|
**OAuth2Validator.oidc_claim_scope,
|
|
"groups": "groups",
|
|
}
|
|
|
|
def _save_bearer_token(self, token, request, *args, **kwargs):
|
|
result = super()._save_bearer_token(token, request, *args, **kwargs)
|
|
from .backchannel import remember_login
|
|
|
|
remember_login(request)
|
|
return result
|
|
|
|
def validate_code(self, client_id, code, client, request, *args, **kwargs):
|
|
valid = super().validate_code(client_id, code, client, request, *args, **kwargs)
|
|
return bool(valid and _is_active_employee(getattr(request, "user", None)))
|
|
|
|
def validate_bearer_token(self, token, scopes, request):
|
|
valid = super().validate_bearer_token(token, scopes, request)
|
|
return bool(valid and _is_active_employee(getattr(request, "user", None)))
|
|
|
|
def validate_refresh_token(self, refresh_token, client, request, *args, **kwargs):
|
|
valid = super().validate_refresh_token(
|
|
refresh_token, client, request, *args, **kwargs
|
|
)
|
|
return bool(valid and _is_active_employee(getattr(request, "user", None)))
|
|
|
|
def get_additional_claims(self, request):
|
|
user = request.user
|
|
full_name = user.get_full_name().strip()
|
|
updated_at = getattr(user, "updated_at", None)
|
|
claims = {
|
|
"name": full_name or str(user),
|
|
"given_name": user.first_name,
|
|
"family_name": user.last_name,
|
|
"preferred_username": _preferred_username(user),
|
|
"nickname": _preferred_username(user),
|
|
"email": user.email,
|
|
"email_verified": bool(user.email_verified),
|
|
"locale": "ru-RU",
|
|
"zoneinfo": settings.TIME_ZONE,
|
|
"groups": _employee_groups(user),
|
|
}
|
|
picture = _picture_url(user)
|
|
if picture:
|
|
claims["picture"] = picture
|
|
if updated_at:
|
|
claims["updated_at"] = int(updated_at.timestamp())
|
|
return claims
|