Игорь Надеин
8b75c6b5d0
All checks were successful
Deploy production / deploy (push) Successful in 10m4s
136 lines
4.3 KiB
Python
136 lines
4.3 KiB
Python
import hmac
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth import get_user_model
|
|
from django.contrib.auth.views import redirect_to_login
|
|
from django.db.models import Prefetch
|
|
from django.http import HttpResponseForbidden, HttpResponseNotFound, JsonResponse
|
|
from django.shortcuts import resolve_url
|
|
from django.utils import timezone
|
|
from django.utils.decorators import method_decorator
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
from django.views.decorators.http import require_GET
|
|
from oauth2_provider import views as oauth_views
|
|
|
|
from employees.constants import ACTION_DISMISSED
|
|
from employees.models import EmployeeAction
|
|
from employees.services.personnel_state import resolve_employee_personnel_state
|
|
|
|
from .validators import _preferred_username
|
|
|
|
|
|
class OidcProviderEnabledMixin:
|
|
def dispatch(self, request, *args, **kwargs):
|
|
if not settings.OIDC_PROVIDER_ENABLED:
|
|
return HttpResponseNotFound()
|
|
return super().dispatch(request, *args, **kwargs)
|
|
|
|
|
|
class CorporateAuthorizationView(
|
|
OidcProviderEnabledMixin,
|
|
oauth_views.AuthorizationView,
|
|
):
|
|
"""Send unauthenticated authorization requests through the portal login."""
|
|
|
|
def get_login_url(self):
|
|
return settings.OIDC_PORTAL_LOGIN_URL
|
|
|
|
def handle_no_permission(self):
|
|
# Keep the return target relative. This remains safe behind a reverse proxy
|
|
# even if Django receives an internal host name and the SPA is public.
|
|
return redirect_to_login(
|
|
self.request.get_full_path(),
|
|
resolve_url(self.get_login_url()),
|
|
self.get_redirect_field_name(),
|
|
)
|
|
|
|
def handle_prompt_login(self):
|
|
query = self.request.GET.copy()
|
|
query.pop("prompt", None)
|
|
return_path = self.request.path
|
|
if query:
|
|
return_path = f"{return_path}?{query.urlencode()}"
|
|
return redirect_to_login(
|
|
return_path,
|
|
resolve_url(self.get_login_url()),
|
|
self.get_redirect_field_name(),
|
|
)
|
|
|
|
|
|
@method_decorator(csrf_exempt, name="dispatch")
|
|
class CorporateTokenView(OidcProviderEnabledMixin, oauth_views.TokenView):
|
|
pass
|
|
|
|
|
|
@method_decorator(csrf_exempt, name="dispatch")
|
|
class CorporateRevokeTokenView(
|
|
OidcProviderEnabledMixin,
|
|
oauth_views.RevokeTokenView,
|
|
):
|
|
pass
|
|
|
|
|
|
@method_decorator(csrf_exempt, name="dispatch")
|
|
class CorporateIntrospectTokenView(
|
|
OidcProviderEnabledMixin,
|
|
oauth_views.IntrospectTokenView,
|
|
):
|
|
pass
|
|
|
|
|
|
@require_GET
|
|
def hiro_employee_directory(request):
|
|
"""Complete, minimal identity roster for the separately authenticated PBX."""
|
|
secret = settings.HIRO_DIRECTORY_SECRET
|
|
if not settings.OIDC_PROVIDER_ENABLED or len(secret) < 32:
|
|
return HttpResponseNotFound()
|
|
supplied = request.headers.get("Authorization", "")
|
|
if not hmac.compare_digest(
|
|
supplied.encode("utf-8"), f"Bearer {secret}".encode("utf-8")
|
|
):
|
|
return HttpResponseForbidden()
|
|
employees = []
|
|
users = (
|
|
get_user_model()
|
|
.objects.order_by("pk")
|
|
.prefetch_related(
|
|
"groups",
|
|
Prefetch(
|
|
"actions",
|
|
queryset=EmployeeAction.objects.select_related("source_request"),
|
|
),
|
|
)
|
|
)
|
|
today = timezone.localdate()
|
|
for user in users:
|
|
state = resolve_employee_personnel_state(
|
|
user, today, actions=user.actions.all()
|
|
)
|
|
active = bool(
|
|
user.is_active and user.email_verified and state.status != ACTION_DISMISSED
|
|
)
|
|
item = {"sub": str(user.pk), "active": active}
|
|
if active:
|
|
groups = {"portal-users", *(group.name for group in user.groups.all())}
|
|
if user.is_staff:
|
|
groups.add("portal-staff")
|
|
if user.is_superuser:
|
|
groups.add("portal-admins")
|
|
item.update(
|
|
preferred_username=_preferred_username(user),
|
|
name=user.get_full_name().strip() or str(user),
|
|
groups=sorted(groups),
|
|
)
|
|
employees.append(item)
|
|
response = JsonResponse(
|
|
{
|
|
"version": 1,
|
|
"issuer": settings.OIDC_ISSUER_URL,
|
|
"complete": True,
|
|
"count": len(employees),
|
|
"employees": employees,
|
|
}
|
|
)
|
|
response["Cache-Control"] = "no-store, private"
|
|
return response
|