diff --git a/backend/api/oidc/urls.py b/backend/api/oidc/urls.py index 05635849..dcb60f68 100644 --- a/backend/api/oidc/urls.py +++ b/backend/api/oidc/urls.py @@ -6,6 +6,7 @@ from .views import ( CorporateIntrospectTokenView, CorporateRevokeTokenView, CorporateTokenView, + hiro_employee_directory, ) app_name = "oauth2_provider" @@ -32,4 +33,5 @@ urlpatterns = [ ), path(".well-known/jwks.json", views.JwksInfoView.as_view(), name="jwks-info"), path("userinfo/", views.UserInfoView.as_view(), name="user-info"), + path("hiro-directory/", hiro_employee_directory, name="hiro-directory"), ] diff --git a/backend/api/oidc/views.py b/backend/api/oidc/views.py index 36fbd7b7..6fd06569 100644 --- a/backend/api/oidc/views.py +++ b/backend/api/oidc/views.py @@ -1,11 +1,23 @@ +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.http import HttpResponseNotFound +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): @@ -64,3 +76,60 @@ class CorporateIntrospectTokenView( 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 diff --git a/backend/eusrr_backend/settings.py b/backend/eusrr_backend/settings.py index 2feba9f4..23274f6b 100644 --- a/backend/eusrr_backend/settings.py +++ b/backend/eusrr_backend/settings.py @@ -509,6 +509,9 @@ OIDC_LEGACY_NO_PKCE_CLIENT_IDS = _split_env_list( os.getenv("OIDC_LEGACY_NO_PKCE_CLIENT_IDS", "") ) +# Dedicated machine credential; an empty value leaves this integration disabled. +HIRO_DIRECTORY_SECRET = os.getenv("HIRO_DIRECTORY_SECRET", "").strip() + OAUTH2_PROVIDER = { "OIDC_ENABLED": OIDC_PROVIDER_ENABLED, "OIDC_ISS_ENDPOINT": OIDC_ISSUER_URL, diff --git a/backend/tests/api/auth/test_oidc_provider.py b/backend/tests/api/auth/test_oidc_provider.py index 304f729a..7a041cb3 100644 --- a/backend/tests/api/auth/test_oidc_provider.py +++ b/backend/tests/api/auth/test_oidc_provider.py @@ -8,14 +8,18 @@ from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import rsa from django.contrib.auth import get_user_model from django.contrib.auth.hashers import check_password +from django.contrib.auth.models import Group from django.core.management import call_command from django.test import Client from django.urls import reverse +from django.utils import timezone from oauth2_provider.models import get_application_model from oauth2_provider.settings import oauth2_settings from rest_framework.test import APIClient from api.oidc.validators import _employee_groups, _preferred_username +from employees.constants import ACTION_DISMISSED +from employees.models import EmployeeAction pytestmark = pytest.mark.django_db User = get_user_model() @@ -242,6 +246,41 @@ def test_inactive_employee_cannot_use_an_existing_access_token( assert response.status_code == 401 +def test_hiro_directory_requires_dedicated_secret_and_reports_dismissals( + client, employee, settings +): + url = reverse("oauth2_provider:hiro-directory") + settings.HIRO_DIRECTORY_SECRET = "" + assert client.get(url).status_code == 404 + settings.HIRO_DIRECTORY_SECRET = "s" * 40 + assert client.get(url).status_code == 403 + assert client.get(url, HTTP_AUTHORIZATION="Bearer wrong").status_code == 403 + group = Group.objects.create(name="hiro_admins") + employee.groups.add(group) + response = client.get( + url, HTTP_AUTHORIZATION="Bearer " + settings.HIRO_DIRECTORY_SECRET + ) + assert response.status_code == 200 + data = response.json() + assert data["issuer"] == settings.OIDC_ISSUER_URL + assert data["complete"] is True + assert data["count"] == len(data["employees"]) + item = next(row for row in data["employees"] if row["sub"] == str(employee.pk)) + assert item["active"] is True + assert "hiro_admins" in item["groups"] + assert "email" not in item + EmployeeAction.objects.create( + employee=employee, action=ACTION_DISMISSED, date=timezone.now() + ) + response = client.get( + url, HTTP_AUTHORIZATION="Bearer " + settings.HIRO_DIRECTORY_SECRET + ) + item = next( + row for row in response.json()["employees"] if row["sub"] == str(employee.pk) + ) + assert item == {"sub": str(employee.pk), "active": False} + + def test_browser_session_bridge_accepts_only_own_authorization_endpoint(employee): api = APIClient() api.force_authenticate(employee)