diff --git a/.env.example b/.env.example index f82a55a2e7..519f518dd0 100644 --- a/.env.example +++ b/.env.example @@ -274,6 +274,16 @@ DATA_UPLOAD_MAX_MEMORY_SIZE=2621440 # (typically representing the main web interface or API client). OAUTH_CLIENT_APP_NAME=local_web_client +# OIDC Provider. Generate the signing key with: +# python manage.py generate_oidc_rsa_key --output oidc_key.pem +OIDC_RSA_KEY_FILE= +# OIDC_RSA_PRIVATE_KEY= +# Comma-separated paths to rotated-out keys; still published in JWKS during +# the rotation overlap window so RPs can verify tokens signed under them. +# OIDC_RSA_INACTIVE_KEY_FILES= +# Required when a key is configured. Becomes the 'iss' claim in every ID token. +OIDC_ISS_ENDPOINT=http://localhost:8000 + # Geographic restriction # **IMPORTANT NOTE: # Geographic restrictions are only applied to projects where the `georestricted` flag is `True`. diff --git a/physionet-django/console/navbar.py b/physionet-django/console/navbar.py index 453e9466e3..5f9944758c 100644 --- a/physionet-django/console/navbar.py +++ b/physionet-django/console/navbar.py @@ -205,6 +205,8 @@ def get_menu_items(self, request): NavLink(_('Administrators'), 'users', view_args=['admin']), ]), + NavLink(_('Partners'), 'partner_list', 'id-card'), + NavLink(_('Featured Content'), 'featured_content', 'star'), NavSubmenu(_('Guidelines'), 'guidelines', 'book', [ diff --git a/physionet-django/console/templates/console/partners/detail.html b/physionet-django/console/templates/console/partners/detail.html new file mode 100644 index 0000000000..3ef491b7d0 --- /dev/null +++ b/physionet-django/console/templates/console/partners/detail.html @@ -0,0 +1,96 @@ +{% extends "console/base_console.html" %} + +{% block title %}Partner: {{ partner.organization_name }}{% endblock %} + +{% block content %} +
+
+

{{ partner.organization_name }}

+ Back to list +
+ + {% if one_time_secret %} +
+ Client secret (shown once) +

Copy this now. It will not be shown again. Store it securely.

+ 
{{ one_time_secret }}
+ Hint: select the value above and copy to clipboard. +
+ {% endif %} + +
+
Organisation
+
+
+
Name
{{ partner.organization_name }}
+
Contact name
{{ partner.contact_name|default:"-" }}
+
Contact email
{{ partner.contact_email|default:"-" }}
+
Agreement signed
{{ partner.agreement_signed_date|default:"-" }}
+
Requires PKCE
{{ partner.requires_pkce|yesno:"Yes,No" }}
+
+ Edit organisation +
+
+ +
+
Credentials
+
+

Client ID: {{ partner.application.client_id }}

+
+ {% csrf_token %} + +
+
+
+ +
+
Allowed scopes
+
+ {% if partner.allowed_scopes %} +
    + {% for scope in partner.allowed_scopes %}
  • {{ scope }}
    • {% endfor %} +
+ {% else %} +

No scope allow-list set (legacy: all configured scopes allowed).

+ {% endif %} + Edit scopes +
+
+ +
+
Redirect URIs
+
+

Login (redirect_uris):

+ 
{{ partner.application.redirect_uris }}
+

Post-logout:

+ 
{{ partner.post_logout_redirect_uris|default:"(none)" }}
+ Edit redirect URIs +
+
+ +
+
Lifecycle
+
+

Status: {{ partner.get_status_display }}

+ {% if partner.status_reason %}

Reason: {{ partner.status_reason }}

{% endif %} + {% if partner.status_changed_at %}

Changed at: {{ partner.status_changed_at }}

{% endif %} + + {% if partner.status == Status.ACTIVE %} + Suspend + Revoke + {% elif partner.status == Status.SUSPENDED %} +
+ {% csrf_token %} + +
+ Revoke + {% elif partner.status == Status.REVOKED %} +

Revoked partners cannot be reactivated.

+ {% endif %} +
+
+
+{% endblock %} diff --git a/physionet-django/console/templates/console/partners/edit.html b/physionet-django/console/templates/console/partners/edit.html new file mode 100644 index 0000000000..8bd289259f --- /dev/null +++ b/physionet-django/console/templates/console/partners/edit.html @@ -0,0 +1,31 @@ +{% extends "console/base_console.html" %} + +{% block title %}Edit Partner: {{ partner.organization_name }}{% endblock %} + +{% block content %} +
+

Edit organisation: {{ partner.organization_name }}

+ +
+ {% csrf_token %} + {{ form.non_field_errors }} + {% for field in form %} +
+ {% if field.field.widget.input_type == 'checkbox' %} +
+ {{ field }} + +
+ {% else %} + + {{ field }} + {% endif %} + {% if field.help_text %}{{ field.help_text }}{% endif %} + {{ field.errors }} +
+ {% endfor %} + + Cancel +
+
+{% endblock %} diff --git a/physionet-django/console/templates/console/partners/list.html b/physionet-django/console/templates/console/partners/list.html new file mode 100644 index 0000000000..f74012f3b0 --- /dev/null +++ b/physionet-django/console/templates/console/partners/list.html @@ -0,0 +1,48 @@ +{% extends "console/base_console.html" %} + +{% block title %}OAuth Partners{% endblock %} + +{% block content %} +
+
+

OAuth Partners

+ New Partner +
+ +
+ + + +
+ + + + + + + + + + + + + {% for partner in partners %} + + + + + + + + {% empty %} + + {% endfor %} + +
OrganisationClient IDContactStatus
{{ partner.organization_name }}{{ partner.application.client_id }}{{ partner.contact_email|default:"-" }}{{ partner.get_status_display }}View
No partners found.
+
+{% endblock %} diff --git a/physionet-django/console/templates/console/partners/new.html b/physionet-django/console/templates/console/partners/new.html new file mode 100644 index 0000000000..3630a7ad1c --- /dev/null +++ b/physionet-django/console/templates/console/partners/new.html @@ -0,0 +1,31 @@ +{% extends "console/base_console.html" %} + +{% block title %}New OAuth Partner{% endblock %} + +{% block content %} +
+

New OAuth Partner

+ +
+ {% csrf_token %} + {{ form.non_field_errors }} + {% for field in form %} +
+ {% if field.field.widget.input_type == 'checkbox' %} +
+ {{ field }} + +
+ {% else %} + + {{ field }} + {% endif %} + {% if field.help_text %}{{ field.help_text }}{% endif %} + {{ field.errors }} +
+ {% endfor %} + + Cancel +
+
+{% endblock %} diff --git a/physionet-django/console/templates/console/partners/redirect_uris_form.html b/physionet-django/console/templates/console/partners/redirect_uris_form.html new file mode 100644 index 0000000000..47e36db31d --- /dev/null +++ b/physionet-django/console/templates/console/partners/redirect_uris_form.html @@ -0,0 +1,24 @@ +{% extends "console/base_console.html" %} + +{% block title %}Edit Redirect URIs: {{ partner.organization_name }}{% endblock %} + +{% block content %} +
+

Redirect URIs: {{ partner.organization_name }}

+ +
+ {% csrf_token %} + {{ form.non_field_errors }} + {% for field in form %} +
+ + {{ field }} + {% if field.help_text %}{{ field.help_text }}{% endif %} + {{ field.errors }} +
+ {% endfor %} + + Cancel +
+
+{% endblock %} diff --git a/physionet-django/console/templates/console/partners/scopes_form.html b/physionet-django/console/templates/console/partners/scopes_form.html new file mode 100644 index 0000000000..1ac5a5dd0e --- /dev/null +++ b/physionet-django/console/templates/console/partners/scopes_form.html @@ -0,0 +1,23 @@ +{% extends "console/base_console.html" %} + +{% block title %}Edit Scopes: {{ partner.organization_name }}{% endblock %} + +{% block content %} +
+

Allowed scopes: {{ partner.organization_name }}

+ +
+ {% csrf_token %} + {{ form.non_field_errors }} + {% for field in form %} +
+ + {{ field }} + {{ field.errors }} +
+ {% endfor %} + + Cancel +
+
+{% endblock %} diff --git a/physionet-django/console/templates/console/partners/suspend_form.html b/physionet-django/console/templates/console/partners/suspend_form.html new file mode 100644 index 0000000000..faba8b0580 --- /dev/null +++ b/physionet-django/console/templates/console/partners/suspend_form.html @@ -0,0 +1,35 @@ +{% extends "console/base_console.html" %} + +{% block title %}{{ action|title }} Partner: {{ partner.organization_name }}{% endblock %} + +{% block content %} +
+

{{ action|title }} partner: {{ partner.organization_name }}

+ + {% if action == "revoke" %} +
+ Warning: Revocation is terminal. The partner cannot be reactivated. +
+ {% else %} +
+ Suspending the partner blocks new authorisation requests. You may also revoke active access tokens. +
+ {% endif %} + +
+ {% csrf_token %} + {{ form.non_field_errors }} + {% for field in form %} +
+ + {{ field }} + {{ field.errors }} +
+ {% endfor %} + + Cancel +
+
+{% endblock %} diff --git a/physionet-django/console/test_views.py b/physionet-django/console/test_views.py index 10b94764cb..dd94556834 100644 --- a/physionet-django/console/test_views.py +++ b/physionet-django/console/test_views.py @@ -1362,6 +1362,231 @@ def test_editor_home_excludes_on_hold(self): self.assertIn(project, response.context['on_hold_projects']) +class TestPartnersConsole(TestMixin): + """ + Test the /console/partners/ admin workflow: list, create, detail, + edit, scope and redirect-uri editing, secret rotation, and the + suspend/reactivate/revoke lifecycle transitions. + """ + + ADMIN_USER = 'admin' + ADMIN_PASSWORD = 'Tester11!' + NON_ADMIN_USER = 'aewj' + NON_ADMIN_PASSWORD = 'Tester11!' + + def _make_partner(self, organization_name="Acme", status=None, + allowed_scopes=None, redirect_uris="http://example.org"): + """Create a fresh Application + Partner pair without conflicting on OneToOne.""" + from oauth2_provider.models import get_application_model + from oauth.models import Partner + Application = get_application_model() + admin = User.objects.get(username=self.ADMIN_USER) + application = Application.objects.create( + name=organization_name[:255], + redirect_uris=redirect_uris, + user=admin, + client_type=Application.CLIENT_CONFIDENTIAL, + authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE, + client_secret="cleartext-test-secret-1234567890", + ) + partner = Partner.objects.create( + application=application, + organization_name=organization_name, + contact_email="contact@example.org", + allowed_scopes=( + allowed_scopes if allowed_scopes is not None else ["openid", "profile", "email"] + ), + created_by=admin, + ) + if status is not None: + partner.status = status + partner.save() + return partner + + def test_partner_list_requires_permission(self): + """A non-admin user is denied access to the partner list.""" + self.client.login(username=self.NON_ADMIN_USER, password=self.NON_ADMIN_PASSWORD) + response = self.client.get(reverse('partner_list')) + self.assertIn(response.status_code, (302, 403)) + + def test_partner_list_renders_for_admin(self): + """An admin user can render the partner list page.""" + self.client.login(username=self.ADMIN_USER, password=self.ADMIN_PASSWORD) + response = self.client.get(reverse('partner_list')) + self.assertEqual(response.status_code, 200) + self.assertTemplateUsed(response, 'console/partners/list.html') + + def test_create_partner_creates_application_and_partner(self): + """Submitting the new-partner form creates an Application and a Partner.""" + from oauth.models import Partner + self.client.login(username=self.ADMIN_USER, password=self.ADMIN_PASSWORD) + response = self.client.post(reverse('partner_new'), data={ + 'organization_name': 'Acme Corp', + 'contact_name': 'Alice', + 'contact_email': 'alice@acme.test', + 'agreement_signed_date': '2025-01-01', + 'redirect_uris': 'https://acme.test/callback', + 'post_logout_redirect_uris': '', + 'allowed_scopes': ['openid', 'profile', 'email'], + }) + self.assertEqual(response.status_code, 302) + self.assertIn('show_secret=1', response.url) + partner = Partner.objects.get(organization_name='Acme Corp') + self.assertEqual(partner.contact_name, 'Alice') + self.assertEqual(partner.application.redirect_uris, 'https://acme.test/callback') + self.assertEqual(partner.allowed_scopes, ['openid', 'profile', 'email']) + + def test_create_partner_rejects_invalid_redirect_uri(self): + """A non-http(s) redirect URI causes the form to re-render with an error.""" + self.client.login(username=self.ADMIN_USER, password=self.ADMIN_PASSWORD) + response = self.client.post(reverse('partner_new'), data={ + 'organization_name': 'Bad Corp', + 'contact_name': '', + 'contact_email': '', + 'agreement_signed_date': '', + 'redirect_uris': 'javascript:alert(1)', + 'post_logout_redirect_uris': '', + 'allowed_scopes': ['openid'], + }) + self.assertEqual(response.status_code, 200) + self.assertContains(response, 'must be http://') + + def test_secret_shown_once_after_creation(self): + """The one-time secret is shown on detail with show_secret=1, then cleared.""" + self.client.login(username=self.ADMIN_USER, password=self.ADMIN_PASSWORD) + self.client.post(reverse('partner_new'), data={ + 'organization_name': 'OneTime Corp', + 'contact_name': '', + 'contact_email': '', + 'agreement_signed_date': '', + 'redirect_uris': 'https://onetime.test/callback', + 'post_logout_redirect_uris': '', + 'allowed_scopes': ['openid'], + }) + from oauth.models import Partner + partner = Partner.objects.get(organization_name='OneTime Corp') + url = reverse('partner_detail', args=[partner.pk]) + '?show_secret=1' + first = self.client.get(url) + self.assertEqual(first.status_code, 200) + self.assertContains(first, 'shown once') + # Second visit no longer renders the secret block. + second = self.client.get(url) + self.assertNotContains(second, 'shown once') + + def test_detail_view_renders_partner_data(self): + """Detail view shows org name and client_id.""" + partner = self._make_partner(organization_name='Detail Co') + self.client.login(username=self.ADMIN_USER, password=self.ADMIN_PASSWORD) + response = self.client.get(reverse('partner_detail', args=[partner.pk])) + self.assertEqual(response.status_code, 200) + self.assertContains(response, 'Detail Co') + self.assertContains(response, partner.application.client_id) + + def test_edit_view_updates_org_fields(self): + """Posting the edit form updates organisation fields.""" + partner = self._make_partner(organization_name='Old Name') + self.client.login(username=self.ADMIN_USER, password=self.ADMIN_PASSWORD) + response = self.client.post(reverse('partner_edit', args=[partner.pk]), data={ + 'organization_name': 'New Name', + 'contact_name': 'Bob', + 'contact_email': 'bob@new.test', + 'agreement_signed_date': '2025-02-02', + }) + self.assertEqual(response.status_code, 302) + partner.refresh_from_db() + self.assertEqual(partner.organization_name, 'New Name') + self.assertEqual(str(partner.agreement_signed_date), '2025-02-02') + + def test_scopes_view_rejects_unknown_scope(self): + """Posting an unknown scope re-renders the form and leaves data unchanged.""" + partner = self._make_partner(allowed_scopes=['openid', 'profile']) + self.client.login(username=self.ADMIN_USER, password=self.ADMIN_PASSWORD) + response = self.client.post(reverse('partner_scopes', args=[partner.pk]), data={ + 'allowed_scopes': ['openid', 'not-a-real-scope'], + }) + self.assertEqual(response.status_code, 200) + partner.refresh_from_db() + self.assertEqual(partner.allowed_scopes, ['openid', 'profile']) + + def test_rotate_secret_invalidates_active_tokens(self): + """Rotating the client secret deletes active access tokens.""" + from datetime import timedelta + from django.utils import timezone as dj_timezone + from oauth2_provider.models import get_access_token_model + AccessToken = get_access_token_model() + + partner = self._make_partner(organization_name='Rotate Co') + admin = User.objects.get(username=self.ADMIN_USER) + AccessToken.objects.create( + user=admin, + scope='openid', + expires=dj_timezone.now() + timedelta(seconds=300), + token='to-be-revoked', + application=partner.application, + ) + self.assertEqual(AccessToken.objects.filter(application=partner.application).count(), 1) + + self.client.login(username=self.ADMIN_USER, password=self.ADMIN_PASSWORD) + response = self.client.post(reverse('partner_rotate_secret', args=[partner.pk])) + self.assertEqual(response.status_code, 302) + self.assertEqual(AccessToken.objects.filter(application=partner.application).count(), 0) + + def test_suspend_records_reason_and_status(self): + """Suspend transitions the partner and records a reason and timestamp.""" + from oauth.models import Partner + partner = self._make_partner(organization_name='Suspend Co') + self.client.login(username=self.ADMIN_USER, password=self.ADMIN_PASSWORD) + response = self.client.post(reverse('partner_suspend', args=[partner.pk]), data={ + 'status_reason': 'misuse', + }) + self.assertEqual(response.status_code, 302) + partner.refresh_from_db() + self.assertEqual(partner.status, Partner.Status.SUSPENDED) + self.assertEqual(partner.status_reason, 'misuse') + self.assertIsNotNone(partner.status_changed_at) + + def test_reactivate_only_works_on_suspended(self): + """Reactivate is rejected for revoked partners but works for suspended ones.""" + from oauth.models import Partner + partner = self._make_partner(status=Partner.Status.REVOKED) + self.client.login(username=self.ADMIN_USER, password=self.ADMIN_PASSWORD) + response = self.client.post(reverse('partner_reactivate', args=[partner.pk])) + self.assertEqual(response.status_code, 400) + + partner2 = self._make_partner(organization_name='Reactivate Co', + status=Partner.Status.SUSPENDED) + response = self.client.post(reverse('partner_reactivate', args=[partner2.pk])) + self.assertEqual(response.status_code, 302) + partner2.refresh_from_db() + self.assertEqual(partner2.status, Partner.Status.ACTIVE) + + def test_revoke_is_terminal(self): + """After revoke, a subsequent reactivate attempt is rejected.""" + from oauth.models import Partner + partner = self._make_partner(organization_name='Revoke Co') + self.client.login(username=self.ADMIN_USER, password=self.ADMIN_PASSWORD) + response = self.client.post(reverse('partner_revoke', args=[partner.pk]), data={ + 'status_reason': 'breach', + }) + self.assertEqual(response.status_code, 302) + partner.refresh_from_db() + self.assertEqual(partner.status, Partner.Status.REVOKED) + + response = self.client.post(reverse('partner_reactivate', args=[partner.pk])) + self.assertEqual(response.status_code, 400) + + def test_redirect_uris_view_validates_format(self): + """Posting an invalid URI to the redirect-uris view shows a validation error.""" + partner = self._make_partner(organization_name='URI Co') + self.client.login(username=self.ADMIN_USER, password=self.ADMIN_PASSWORD) + response = self.client.post(reverse('partner_redirect_uris', args=[partner.pk]), data={ + 'redirect_uris': 'javascript:alert(1)', + 'post_logout_redirect_uris': '', + }) + self.assertEqual(response.status_code, 200) + self.assertContains(response, 'must be http://') + + class TestDUALogs(TestMixin): """Test DUA Logs console views.""" diff --git a/physionet-django/console/urls.py b/physionet-django/console/urls.py index 02f153ef59..9a224110ed 100644 --- a/physionet-django/console/urls.py +++ b/physionet-django/console/urls.py @@ -187,6 +187,18 @@ path('courses//archive/', training_views.archive_course, name='archive_course_version'), + # OAuth Partners + path('partners/', views.partner_list, name='partner_list'), + path('partners/new/', views.partner_new, name='partner_new'), + path('partners//', views.partner_detail, name='partner_detail'), + path('partners//edit/', views.partner_edit, name='partner_edit'), + path('partners//scopes/', views.partner_scopes, name='partner_scopes'), + path('partners//redirect-uris/', views.partner_redirect_uris, name='partner_redirect_uris'), + path('partners//rotate-secret/', views.partner_rotate_secret, name='partner_rotate_secret'), + path('partners//suspend/', views.partner_suspend, name='partner_suspend'), + path('partners//reactivate/', views.partner_reactivate, name='partner_reactivate'), + path('partners//revoke/', views.partner_revoke, name='partner_revoke'), + # Federated Sites path('federated-sites/', views.federated_sites, name='federated_sites'), path('federated-sites/add/', views.federated_site_add, name='federated_site_add'), @@ -313,6 +325,16 @@ 'event_agreement_delete': {'_skip_': True}, 'event_agreement_new_version': {'_skip_': True}, + # OAuth Partners (no demo data; covered by TestPartnersConsole) + 'partner_detail': {'_skip_': True}, + 'partner_edit': {'_skip_': True}, + 'partner_scopes': {'_skip_': True}, + 'partner_redirect_uris': {'_skip_': True}, + 'partner_rotate_secret': {'_skip_': True}, + 'partner_suspend': {'_skip_': True}, + 'partner_reactivate': {'_skip_': True}, + 'partner_revoke': {'_skip_': True}, + # DUA Logs: pk must be a credentialed project 'dua_logs_detail': { # id of a PublishedProject with access_policy=CREDENTIALED diff --git a/physionet-django/console/views.py b/physionet-django/console/views.py index 641055b332..3541e42345 100644 --- a/physionet-django/console/views.py +++ b/physionet-django/console/views.py @@ -22,7 +22,16 @@ from django.db.models.functions import Cast, TruncDate from django.forms import Select, Textarea, modelformset_factory from django.forms.models import model_to_dict -from django.http import Http404, HttpResponse, JsonResponse, HttpResponseRedirect, StreamingHttpResponse +from django.db import transaction +from django.http import ( + Http404, + HttpResponse, + HttpResponseBadRequest, + HttpResponseRedirect, + JsonResponse, + StreamingHttpResponse, +) +from django.views.decorators.http import require_POST from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse from django.utils import timezone @@ -3912,6 +3921,241 @@ def event_agreement_delete(request, pk): return redirect("event_agreement_list") +# ------------------------- OAuth Partners ------------------------- # + + +@console_permission_required('oauth.change_partner') +def partner_list(request): + from oauth.models import Partner + partners = Partner.objects.select_related("application").order_by("organization_name") + status_filter = request.GET.get("status") + if status_filter: + partners = partners.filter(status=status_filter) + return render(request, "console/partners/list.html", { + "partners": partners, + "status_filter": status_filter, + "Status": Partner.Status, + }) + + +@console_permission_required('oauth.change_partner') +def partner_new(request): + from oauth.forms import PartnerCreateForm + from oauth.models import Partner + from oauth2_provider.models import get_application_model + Application = get_application_model() + + if request.method == "POST": + form = PartnerCreateForm(request.POST) + if form.is_valid(): + # Opt the partner in to RS256 ID-token signing if they need OIDC. + # Without this, /oauth/token/ raises ImproperlyConfigured for any + # request including the openid scope. + allowed_scopes = form.cleaned_data["allowed_scopes"] + algorithm = ( + Application.RS256_ALGORITHM + if "openid" in allowed_scopes + else Application.NO_ALGORITHM + ) + with transaction.atomic(): + application = Application( + name=form.cleaned_data["organization_name"][:255], + redirect_uris=form.cleaned_data["redirect_uris"], + user=request.user, + client_type=Application.CLIENT_CONFIDENTIAL, + authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE, + algorithm=algorithm, + ) + # Captured BEFORE save: DOT may hash the secret on save in newer versions. + cleartext_secret = application.client_secret + application.save() + partner = Partner.objects.create( + application=application, + organization_name=form.cleaned_data["organization_name"], + contact_name=form.cleaned_data["contact_name"], + contact_email=form.cleaned_data["contact_email"], + agreement_signed_date=form.cleaned_data["agreement_signed_date"], + post_logout_redirect_uris=form.cleaned_data["post_logout_redirect_uris"], + allowed_scopes=allowed_scopes, + requires_pkce=form.cleaned_data.get("requires_pkce", True), + created_by=request.user, + ) + request.session["_partner_one_time_secret"] = cleartext_secret + return redirect(reverse("partner_detail", args=[partner.pk]) + "?show_secret=1") + else: + form = PartnerCreateForm() + return render(request, "console/partners/new.html", {"form": form}) + + +@console_permission_required('oauth.change_partner') +def partner_detail(request, pk): + from oauth.models import Partner + partner = get_object_or_404(Partner, pk=pk) + one_time_secret = None + if request.GET.get("show_secret") == "1": + one_time_secret = request.session.pop("_partner_one_time_secret", None) + return render(request, "console/partners/detail.html", { + "partner": partner, + "one_time_secret": one_time_secret, + "Status": Partner.Status, + }) + + +@console_permission_required('oauth.change_partner') +def partner_edit(request, pk): + from oauth.forms import PartnerEditForm + from oauth.models import Partner + partner = get_object_or_404(Partner, pk=pk) + if request.method == "POST": + form = PartnerEditForm(request.POST, instance=partner) + if form.is_valid(): + form.save() + return redirect("partner_detail", pk=partner.pk) + else: + form = PartnerEditForm(instance=partner) + return render(request, "console/partners/edit.html", + {"form": form, "partner": partner}) + + +@console_permission_required('oauth.change_partner') +def partner_scopes(request, pk): + from oauth.forms import PartnerScopesForm + from oauth.models import Partner + from oauth2_provider.models import get_application_model + Application = get_application_model() + + partner = get_object_or_404(Partner, pk=pk) + if request.method == "POST": + form = PartnerScopesForm(request.POST, instance=partner) + if form.is_valid(): + form.save() + # Keep the Application's signing algorithm in sync with whether + # the partner now needs OIDC; otherwise /oauth/token/ will either + # fail (no algorithm) or sign tokens nobody asked for. + needs_oidc = "openid" in (partner.allowed_scopes or []) + target_alg = ( + Application.RS256_ALGORITHM if needs_oidc else Application.NO_ALGORITHM + ) + if partner.application.algorithm != target_alg: + partner.application.algorithm = target_alg + partner.application.save(update_fields=["algorithm"]) + return redirect("partner_detail", pk=partner.pk) + else: + form = PartnerScopesForm(instance=partner) + return render(request, "console/partners/scopes_form.html", + {"form": form, "partner": partner}) + + +@console_permission_required('oauth.change_partner') +def partner_redirect_uris(request, pk): + from oauth.forms import PartnerRedirectURIsForm + from oauth.models import Partner + partner = get_object_or_404(Partner, pk=pk) + if request.method == "POST": + form = PartnerRedirectURIsForm(request.POST) + if form.is_valid(): + partner.application.redirect_uris = form.cleaned_data["redirect_uris"] + partner.application.save() + partner.post_logout_redirect_uris = form.cleaned_data["post_logout_redirect_uris"] + partner.save() + return redirect("partner_detail", pk=partner.pk) + else: + form = PartnerRedirectURIsForm(initial={ + "redirect_uris": partner.application.redirect_uris, + "post_logout_redirect_uris": partner.post_logout_redirect_uris, + }) + return render(request, "console/partners/redirect_uris_form.html", + {"form": form, "partner": partner}) + + +@console_permission_required('oauth.change_partner') +@require_POST +def partner_rotate_secret(request, pk): + from oauth.models import Partner + from oauth2_provider.generators import generate_client_secret + from oauth2_provider.models import get_access_token_model, get_refresh_token_model + AccessToken = get_access_token_model() + RefreshToken = get_refresh_token_model() + + partner = get_object_or_404(Partner, pk=pk) + new_secret = generate_client_secret() + partner.application.client_secret = new_secret + partner.application.save() + AccessToken.objects.filter(application=partner.application).delete() + RefreshToken.objects.filter(application=partner.application).delete() + request.session["_partner_one_time_secret"] = new_secret + return redirect(reverse("partner_detail", args=[partner.pk]) + "?show_secret=1") + + +@console_permission_required('oauth.change_partner') +def partner_suspend(request, pk): + from oauth.forms import PartnerSuspendForm + from oauth.models import Partner + partner = get_object_or_404(Partner, pk=pk) + if partner.status == Partner.Status.REVOKED: + return HttpResponseBadRequest("Cannot suspend a revoked partner.") + if request.method == "POST": + form = PartnerSuspendForm(request.POST) + if form.is_valid(): + partner.status = Partner.Status.SUSPENDED + partner.status_reason = form.cleaned_data["status_reason"] + partner.status_changed_at = timezone.now() + partner.save() + if form.cleaned_data["revoke_active_tokens"]: + from oauth2_provider.models import get_access_token_model, get_refresh_token_model + get_access_token_model().objects.filter( + application=partner.application, + ).delete() + get_refresh_token_model().objects.filter( + application=partner.application, + ).delete() + return redirect("partner_detail", pk=partner.pk) + else: + form = PartnerSuspendForm() + return render(request, "console/partners/suspend_form.html", + {"form": form, "partner": partner, "action": "suspend"}) + + +@console_permission_required('oauth.change_partner') +@require_POST +def partner_reactivate(request, pk): + from oauth.models import Partner + partner = get_object_or_404(Partner, pk=pk) + if partner.status != Partner.Status.SUSPENDED: + return HttpResponseBadRequest("Only suspended partners can be reactivated.") + partner.status = Partner.Status.ACTIVE + partner.status_reason = "" + partner.status_changed_at = timezone.now() + partner.save() + return redirect("partner_detail", pk=partner.pk) + + +@console_permission_required('oauth.change_partner') +def partner_revoke(request, pk): + from oauth.forms import PartnerSuspendForm + from oauth.models import Partner + partner = get_object_or_404(Partner, pk=pk) + if request.method == "POST": + form = PartnerSuspendForm(request.POST) + if form.is_valid(): + partner.status = Partner.Status.REVOKED + partner.status_reason = form.cleaned_data["status_reason"] + partner.status_changed_at = timezone.now() + partner.save() + from oauth2_provider.models import get_access_token_model, get_refresh_token_model + get_access_token_model().objects.filter( + application=partner.application, + ).delete() + get_refresh_token_model().objects.filter( + application=partner.application, + ).delete() + return redirect("partner_detail", pk=partner.pk) + else: + form = PartnerSuspendForm() + return render(request, "console/partners/suspend_form.html", + {"form": form, "partner": partner, "action": "revoke"}) + + # ------------------------- Federated Sites Views ------------------------- # diff --git a/physionet-django/oauth/admin.py b/physionet-django/oauth/admin.py index 8c38f3f3da..bb1069e16d 100644 --- a/physionet-django/oauth/admin.py +++ b/physionet-django/oauth/admin.py @@ -1,3 +1,14 @@ from django.contrib import admin -# Register your models here. +from oauth.models import Partner + + +@admin.register(Partner) +class PartnerAdmin(admin.ModelAdmin): + list_display = ('organization_name', 'application', 'status', 'agreement_signed_date', 'created_at') + list_filter = ('status',) + search_fields = ('organization_name', 'contact_email', 'application__client_id') + raw_id_fields = ('application', 'created_by') + + +# DOT auto-registers Application; we don't re-register here. diff --git a/physionet-django/oauth/forms.py b/physionet-django/oauth/forms.py new file mode 100644 index 0000000000..1e56b03c15 --- /dev/null +++ b/physionet-django/oauth/forms.py @@ -0,0 +1,126 @@ +from django import forms +from django.conf import settings + +from oauth.models import Partner + + +class PartnerCreateForm(forms.Form): + organization_name = forms.CharField(max_length=200) + contact_name = forms.CharField(max_length=200, required=False) + contact_email = forms.EmailField(required=False) + agreement_signed_date = forms.DateField(required=False, widget=forms.DateInput(attrs={"type": "date"})) + redirect_uris = forms.CharField( + widget=forms.Textarea(attrs={"rows": 3}), + help_text="One URI per line.", + ) + post_logout_redirect_uris = forms.CharField( + widget=forms.Textarea(attrs={"rows": 3}), + required=False, + help_text="One URI per line. Optional.", + ) + allowed_scopes = forms.MultipleChoiceField( + widget=forms.CheckboxSelectMultiple, + choices=[], + required=False, + ) + requires_pkce = forms.BooleanField( + required=False, + initial=True, + label="Require PKCE", + help_text=( + "Require PKCE (code_challenge) on /authorize. Disable only when " + "the partner cannot send a code_challenge (e.g. an upstream " + "federator acting as the OAuth client)." + ), + ) + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.fields["allowed_scopes"].choices = [ + (k, f"{k} - {v}") for k, v in settings.OAUTH2_PROVIDER["SCOPES"].items() + ] + if not self.is_bound: + self.initial["allowed_scopes"] = ["openid", "profile", "email"] + + def clean_redirect_uris(self): + raw = self.cleaned_data["redirect_uris"] + uris = [line.strip() for line in raw.splitlines() if line.strip()] + if not uris: + raise forms.ValidationError("At least one redirect URI is required.") + for uri in uris: + if not (uri.startswith("http://") or uri.startswith("https://")): + raise forms.ValidationError(f"{uri!r} must be http:// or https://") + return " ".join(uris) + + def clean_post_logout_redirect_uris(self): + raw = self.cleaned_data["post_logout_redirect_uris"] + uris = [line.strip() for line in raw.splitlines() if line.strip()] + for uri in uris: + if not (uri.startswith("http://") or uri.startswith("https://")): + raise forms.ValidationError(f"{uri!r} must be http:// or https://") + return " ".join(uris) + + +class PartnerEditForm(forms.ModelForm): + class Meta: + model = Partner + fields = ( + "organization_name", "contact_name", "contact_email", + "agreement_signed_date", "requires_pkce", + ) + widgets = { + "agreement_signed_date": forms.DateInput(attrs={"type": "date"}), + } + help_texts = { + "requires_pkce": ( + "Require PKCE (code_challenge) on /authorize. Disable only when " + "the partner cannot send a code_challenge (e.g. an upstream " + "federator acting as the OAuth client)." + ), + } + + +class PartnerScopesForm(forms.ModelForm): + allowed_scopes = forms.MultipleChoiceField( + widget=forms.CheckboxSelectMultiple, + choices=[], + required=False, + ) + + class Meta: + model = Partner + fields = ("allowed_scopes",) + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.fields["allowed_scopes"].choices = [ + (k, f"{k} - {v}") for k, v in settings.OAUTH2_PROVIDER["SCOPES"].items() + ] + + +class PartnerRedirectURIsForm(forms.Form): + redirect_uris = forms.CharField(widget=forms.Textarea(attrs={"rows": 3})) + post_logout_redirect_uris = forms.CharField(widget=forms.Textarea(attrs={"rows": 3}), required=False) + + def clean_redirect_uris(self): + raw = self.cleaned_data["redirect_uris"] + uris = [line.strip() for line in raw.splitlines() if line.strip()] + if not uris: + raise forms.ValidationError("At least one redirect URI is required.") + for uri in uris: + if not (uri.startswith("http://") or uri.startswith("https://")): + raise forms.ValidationError(f"{uri!r} must be http:// or https://") + return " ".join(uris) + + def clean_post_logout_redirect_uris(self): + raw = self.cleaned_data["post_logout_redirect_uris"] + uris = [line.strip() for line in raw.splitlines() if line.strip()] + for uri in uris: + if not (uri.startswith("http://") or uri.startswith("https://")): + raise forms.ValidationError(f"{uri!r} must be http:// or https://") + return " ".join(uris) + + +class PartnerSuspendForm(forms.Form): + status_reason = forms.CharField(widget=forms.Textarea(attrs={"rows": 3})) + revoke_active_tokens = forms.BooleanField(required=False, initial=False) diff --git a/physionet-django/oauth/management/__init__.py b/physionet-django/oauth/management/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/physionet-django/oauth/management/commands/__init__.py b/physionet-django/oauth/management/commands/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/physionet-django/oauth/management/commands/generate_oidc_rsa_key.py b/physionet-django/oauth/management/commands/generate_oidc_rsa_key.py new file mode 100644 index 0000000000..e5a22be957 --- /dev/null +++ b/physionet-django/oauth/management/commands/generate_oidc_rsa_key.py @@ -0,0 +1,54 @@ +import os +import sys + +from django.core.management.base import BaseCommand +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.hazmat.primitives import serialization + + +class Command(BaseCommand): + help = 'Generate an RSA private key for signing OIDC ID tokens' + + def add_arguments(self, parser): + parser.add_argument( + '--bits', + type=int, + default=2048, + help='Key size in bits (minimum 2048, default 2048)', + ) + parser.add_argument( + '--output', + type=str, + default=None, + help='Write key to file (with 0600 permissions) instead of stdout', + ) + + def handle(self, *args, **options): + bits = options['bits'] + if bits < 2048: + self.stderr.write(self.style.ERROR('Key size must be at least 2048 bits')) + return + + private_key = rsa.generate_private_key( + public_exponent=65537, + key_size=bits, + ) + pem = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ).decode() + + output_file = options['output'] + if output_file: + fd = os.open(output_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, 'w') as f: + f.write(pem) + self.stderr.write(self.style.SUCCESS(f'RSA key written to {output_file}')) + else: + if sys.stdout.isatty(): + self.stderr.write(self.style.WARNING( + 'WARNING: Writing private key to terminal. ' + 'Use --output or redirect to a file instead.' + )) + self.stdout.write(pem) diff --git a/physionet-django/oauth/migrations/0001_initial.py b/physionet-django/oauth/migrations/0001_initial.py new file mode 100644 index 0000000000..4b0fcee28a --- /dev/null +++ b/physionet-django/oauth/migrations/0001_initial.py @@ -0,0 +1,39 @@ +# Generated by Django 4.2.28 on 2026-05-04 13:45 + +from django.conf import settings +from django.db import migrations, models +import django.db.models.deletion + + +class Migration(migrations.Migration): + + initial = True + + dependencies = [ + migrations.swappable_dependency(settings.OAUTH2_PROVIDER_APPLICATION_MODEL), + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ] + + operations = [ + migrations.CreateModel( + name='Partner', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('organization_name', models.CharField(max_length=200)), + ('contact_name', models.CharField(blank=True, max_length=200)), + ('contact_email', models.EmailField(blank=True, max_length=254)), + ('agreement_signed_date', models.DateField(blank=True, null=True)), + ('allowed_scopes', models.JSONField(blank=True, default=list)), + ('post_logout_redirect_uris', models.TextField(blank=True)), + ('status', models.CharField(choices=[('active', 'Active'), ('suspended', 'Suspended'), ('revoked', 'Revoked')], default='active', max_length=20)), # noqa: E501 + ('status_reason', models.TextField(blank=True)), + ('status_changed_at', models.DateTimeField(blank=True, null=True)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('application', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='partner', to=settings.OAUTH2_PROVIDER_APPLICATION_MODEL)), # noqa: E501 + ('created_by', models.ForeignKey(null=True, on_delete=django.db.models.deletion.PROTECT, related_name='+', to=settings.AUTH_USER_MODEL)), # noqa: E501 + ], + options={ + 'ordering': ['organization_name'], + }, + ), + ] diff --git a/physionet-django/oauth/migrations/0002_partner_requires_pkce.py b/physionet-django/oauth/migrations/0002_partner_requires_pkce.py new file mode 100644 index 0000000000..405416473d --- /dev/null +++ b/physionet-django/oauth/migrations/0002_partner_requires_pkce.py @@ -0,0 +1,18 @@ +# Generated by Django 4.2.30 on 2026-05-11 12:41 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('oauth', '0001_initial'), + ] + + operations = [ + migrations.AddField( + model_name='partner', + name='requires_pkce', + field=models.BooleanField(default=True), + ), + ] diff --git a/physionet-django/oauth/migrations/0003_backfill_legacy_partners.py b/physionet-django/oauth/migrations/0003_backfill_legacy_partners.py new file mode 100644 index 0000000000..45fa022683 --- /dev/null +++ b/physionet-django/oauth/migrations/0003_backfill_legacy_partners.py @@ -0,0 +1,46 @@ +from django.conf import settings +from django.db import migrations + + +def backfill_legacy_partners(apps, schema_editor): + """For every existing Application, create a 'Legacy: ' Partner.""" + Application = apps.get_model(settings.OAUTH2_PROVIDER_APPLICATION_MODEL) + Partner = apps.get_model('oauth', 'Partner') + for app in Application.objects.all(): + Partner.objects.get_or_create( + application=app, + defaults={ + 'organization_name': f'Legacy: {app.client_id}', + 'allowed_scopes': [], + 'status': 'active', + 'created_by': None, + }, + ) + + +def remove_legacy_partners(apps, schema_editor): + """Reverse: delete only legacy rows; keep admin-created Partners.""" + Partner = apps.get_model('oauth', 'Partner') + Partner.objects.filter(organization_name__startswith='Legacy: ').delete() + + +class Migration(migrations.Migration): + """Backfill Partner rows for Applications that predate this app. + + Marked as a "late" migration (MIGRATE_AFTER_INSTALL = True) so the + upgrade flow applies the schema (0001_initial) ahead of the new + server code but defers this data step until after the codebase is + swapped. Without that deferral, the backfilled Partner rows are + orphaned when the old codebase deletes an Application (Django + handles cascade in Python and the old code can't see Partner). + """ + + MIGRATE_AFTER_INSTALL = True + + dependencies = [ + ('oauth', '0002_partner_requires_pkce'), + ] + + operations = [ + migrations.RunPython(backfill_legacy_partners, remove_legacy_partners), + ] diff --git a/physionet-django/oauth/models.py b/physionet-django/oauth/models.py index 71a8362390..8a1ba6b1e2 100644 --- a/physionet-django/oauth/models.py +++ b/physionet-django/oauth/models.py @@ -1,3 +1,61 @@ +from django.conf import settings from django.db import models +from oauth2_provider.settings import oauth2_settings -# Create your models here. + +class Partner(models.Model): + class Status(models.TextChoices): + ACTIVE = 'active', 'Active' + SUSPENDED = 'suspended', 'Suspended' + REVOKED = 'revoked', 'Revoked' + + application = models.OneToOneField( + oauth2_settings.APPLICATION_MODEL, + on_delete=models.CASCADE, + related_name='partner', + ) + organization_name = models.CharField(max_length=200) + contact_name = models.CharField(max_length=200, blank=True) + contact_email = models.EmailField(blank=True) + agreement_signed_date = models.DateField(null=True, blank=True) + + # Subset of OAUTH2_PROVIDER['SCOPES'] keys this partner is allowed to + # request. Empty list = wildcard (all configured scopes), preserved for + # legacy Applications that predate this model. + allowed_scopes = models.JSONField(default=list, blank=True) + + # Whitespace-separated list of URIs the partner may pass as + # post_logout_redirect_uri to /oauth/end-session/. Stored on Partner + # because DOT 2.2.0's swappable Application has no equivalent field. + post_logout_redirect_uris = models.TextField(blank=True) + + # PKCE is required by default (OAuth 2.1 / RFC 9700). Some upstream + # federators act as the OAuth client without sending a code_challenge; + # opt those partners out here rather than disabling PKCE globally. + requires_pkce = models.BooleanField(default=True) + + status = models.CharField( + max_length=20, + choices=Status.choices, + default=Status.ACTIVE, + ) + status_reason = models.TextField(blank=True) + status_changed_at = models.DateTimeField(null=True, blank=True) + + created_at = models.DateTimeField(auto_now_add=True) + created_by = models.ForeignKey( + settings.AUTH_USER_MODEL, + on_delete=models.PROTECT, + related_name='+', + null=True, + ) + + class Meta: + ordering = ['organization_name'] + + def __str__(self): + return f'{self.organization_name} ({self.application.client_id})' + + @property + def is_legacy(self): + return self.organization_name.startswith('Legacy: ') diff --git a/physionet-django/oauth/templates/oauth/logged_out.html b/physionet-django/oauth/templates/oauth/logged_out.html new file mode 100644 index 0000000000..34db3ec176 --- /dev/null +++ b/physionet-django/oauth/templates/oauth/logged_out.html @@ -0,0 +1,8 @@ +{% extends "base.html" %} +{% block title %}Logged out{% endblock %} +{% block content %} +
+

You have been logged out

+

Your PhysioNet session has ended.

+
+{% endblock %} diff --git a/physionet-django/oauth/tests.py b/physionet-django/oauth/tests.py index fe48247b89..47d9df7a0f 100644 --- a/physionet-django/oauth/tests.py +++ b/physionet-django/oauth/tests.py @@ -1,17 +1,31 @@ import base64 -import random import hashlib -from datetime import timedelta +import os +import random import re -from django.test import TestCase +import tempfile +from datetime import timedelta +from io import StringIO +from unittest import mock +from urllib.parse import parse_qs, urlparse + +import jwt +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from django.conf import settings +from django.core.exceptions import ImproperlyConfigured +from django.core.management import call_command +from django.test import TestCase, override_settings +from django.urls import reverse from django.utils import timezone -from user.models import User +from django.utils.crypto import get_random_string from oauth2_provider.models import get_access_token_model, get_application_model -from django.urls import reverse -from urllib.parse import parse_qs, urlparse from oauth2_provider.settings import oauth2_settings -from django.utils.crypto import get_random_string + +from oauth.models import Partner from oauth.views import SCOPES_MAPPING +from physionet.settings.base import load_oidc_provider_config +from user.models import User Application = get_application_model() @@ -161,26 +175,24 @@ def test_basic_auth(self): """ self.client.login(username="oauth_test_user", password="123456") - # Disabled PKCE removes the need for a code_verifier - # Checkout details on PKCE : https://oauth.net/2/pkce/ - self.oauth2_settings.PKCE_REQUIRED = False + non_pkce_settings = {**settings.OAUTH2_PROVIDER, "PKCE_REQUIRED": False} + with override_settings(OAUTH2_PROVIDER=non_pkce_settings): + authorization_code = self.get_auth() - authorization_code = self.get_auth() - - token_request_data = { - "grant_type": "authorization_code", - "code": authorization_code, - "redirect_uri": "http://example.org", - } - auth_headers = self.get_basic_auth_header( - self.application.client_id, CLEARTEXT_SECRET - ) + token_request_data = { + "grant_type": "authorization_code", + "code": authorization_code, + "redirect_uri": "http://example.org", + } + auth_headers = self.get_basic_auth_header( + self.application.client_id, CLEARTEXT_SECRET + ) - response = self.client.post( - reverse("oauth2_provider:token"), data=token_request_data, **auth_headers - ) - self.assertEqual(response.status_code, 200) - token = response.json()["access_token"] + response = self.client.post( + reverse("oauth2_provider:token"), data=token_request_data, **auth_headers + ) + self.assertEqual(response.status_code, 200) + token = response.json()["access_token"] # Testing the Token Acquired through the above request self.client.logout() @@ -270,3 +282,1036 @@ def test_scope_fields_are_accessible(self): self.assertIsInstance(result, dict, f"Scope '{scope}' did not return a dictionary") for field, value in result.items(): self.assertIsNotNone(field, f"Field name in scope '{scope}' is None") + + +class OIDCBaseTest(BaseTest): + """ + Base test class that enables the OIDC provider with a freshly generated + RSA signing key, scoped to the lifetime of the class. + """ + + @classmethod + def setUpClass(cls): + super().setUpClass() + + # Generate a temporary RSA key for signing test ID tokens + cls._test_rsa_key = rsa.generate_private_key( + public_exponent=65537, key_size=2048, + ) + cls.test_rsa_key_pem = cls._test_rsa_key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ).decode() + cls.test_rsa_public_key = cls._test_rsa_key.public_key() + + # Enable OIDC via override_settings so the change is rolled back even + # if a test fails mid-run, instead of leaking into sibling tests. + # OAUTH2_SERVER_CLASS is set explicitly because DOT only swaps to the + # OIDC server class via a conditional fallback in oauth2_settings, + # which is brittle across class-scoped override cycles. + # ALWAYS_RELOAD_OAUTHLIB_CORE bypasses DOT's class-level cache of the + # built server instance (OAuthLibMixin._oauthlib_core), which is + # populated on the first /authorize or /token call in the process. If + # an earlier test hit those endpoints while OIDC was disabled, the + # cached server is the non-OIDC one and would silently drop id_token + # from the token response here even though OIDC is now enabled. + oidc_overrides = { + **settings.OAUTH2_PROVIDER, + "OIDC_ENABLED": True, + "OIDC_RSA_PRIVATE_KEY": cls.test_rsa_key_pem, + "OIDC_ISS_ENDPOINT": "http://testserver", + "OAUTH2_SERVER_CLASS": "oauthlib.openid.Server", + "ALWAYS_RELOAD_OAUTHLIB_CORE": True, + } + cls._settings_override = override_settings(OAUTH2_PROVIDER=oidc_overrides) + cls._settings_override.enable() + cls.addClassCleanup(cls._settings_override.disable) + + def setUp(self): + super().setUp() + # OIDC ID-token signing requires the Application to opt in to RS256; + # the BaseTest default leaves algorithm blank so the OAuth2-only flow works. + self.application.algorithm = Application.RS256_ALGORITHM + self.application.save() + + +class TestOIDCDiscovery(OIDCBaseTest): + """ + Test the OIDC discovery document is served at the spec-mandated root + location and from the oauth/ namespace. + """ + + def test_root_discovery_endpoint(self): + """The discovery document is served at /.well-known/openid-configuration.""" + response = self.client.get("/.well-known/openid-configuration") + self.assertEqual(response.status_code, 200) + data = response.json() + self.assertEqual(data["issuer"], "http://testserver") + self.assertIn("authorization_endpoint", data) + self.assertIn("token_endpoint", data) + self.assertIn("userinfo_endpoint", data) + self.assertIn("jwks_uri", data) + self.assertIn("response_types_supported", data) + self.assertIn("id_token_signing_alg_values_supported", data) + + def test_oauth_discovery_endpoint(self): + """The discovery document is also reachable under the oauth/ namespace.""" + response = self.client.get("/oauth/.well-known/openid-configuration") + self.assertEqual(response.status_code, 200) + data = response.json() + self.assertEqual(data["issuer"], "http://testserver") + + def test_advertises_introspection_endpoint(self): + """The discovery document publishes the introspection_endpoint.""" + response = self.client.get("/.well-known/openid-configuration") + self.assertEqual(response.status_code, 200) + self.assertEqual( + response.json()["introspection_endpoint"], + "http://testserver/oauth/introspect/", + ) + + def test_advertises_end_session_endpoint(self): + """The discovery document publishes the end_session_endpoint.""" + response = self.client.get("/.well-known/openid-configuration") + self.assertEqual(response.status_code, 200) + self.assertEqual( + response.json()["end_session_endpoint"], + "http://testserver/oauth/end-session/", + ) + + +class TestJWKSEndpoint(OIDCBaseTest): + """ + Test the JWKS endpoint publishes the active signing key in the format + relying parties need to verify ID token signatures. + """ + + def test_jwks_returns_valid_keyset(self): + """JWKS publishes an RSA key with the required JWK fields.""" + response = self.client.get("/oauth/jwks/") + self.assertEqual(response.status_code, 200) + data = response.json() + self.assertIn("keys", data) + self.assertGreater(len(data["keys"]), 0) + + key = data["keys"][0] + self.assertEqual(key["kty"], "RSA") + self.assertIn("n", key) + self.assertIn("e", key) + self.assertIn("kid", key) + + +class TestOIDCTokenFlow(OIDCBaseTest): + """ + Test the OpenID Connect authorization code flow returns a signed ID + token alongside the access token, with the expected standard claims. + """ + + def test_openid_scope_returns_id_token(self): + """An auth code flow with the openid scope returns a verifiable id_token.""" + self.client.login(username="oauth_test_user", password="123456") + + # Use a fresh Application created with RS256 from the start. Mutating + # self.application.algorithm post-creation is unreliable when an earlier + # test class has touched DOT's internal application/grant-type state. + oidc_secret = "1234567890abcdefghijklmnopqrstuvwxyz" + oidc_app = Application.objects.create( + name="OIDC Token Test App", + redirect_uris="http://example.org", + user=self.dev_user, + client_type=Application.CLIENT_CONFIDENTIAL, + authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE, + algorithm=Application.RS256_ALGORITHM, + client_secret=oidc_secret, + ) + + # PKCE_REQUIRED defaults True; disable it for this confidential-client flow. + non_pkce = {**settings.OAUTH2_PROVIDER, "PKCE_REQUIRED": False} + + with override_settings(OAUTH2_PROVIDER=non_pkce): + # Request authorization with the standard OIDC scopes + authcode_data = { + "client_id": oidc_app.client_id, + "state": "random_state_string", + "scope": "openid profile email", + "redirect_uri": "http://example.org", + "response_type": "code", + "allow": True, + } + + response = self.client.post( + reverse("oauth2_provider:authorize"), data=authcode_data + ) + query_dict = parse_qs(urlparse(response["Location"]).query) + authorization_code = query_dict["code"].pop() + + # Exchange the code for tokens + token_request_data = { + "grant_type": "authorization_code", + "code": authorization_code, + "redirect_uri": "http://example.org", + } + auth_headers = self.get_basic_auth_header( + oidc_app.client_id, oidc_secret, + ) + + response = self.client.post( + reverse("oauth2_provider:token"), data=token_request_data, **auth_headers + ) + self.assertEqual(response.status_code, 200) + token_data = response.json() + self.assertIn("id_token", token_data) + self.assertIn("access_token", token_data) + + # Verify the ID token signature and standard claims + id_token = jwt.decode( + token_data["id_token"], + self.test_rsa_public_key, + algorithms=["RS256"], + audience=oidc_app.client_id, + ) + self.assertEqual(id_token["iss"], "http://testserver") + self.assertEqual(id_token["sub"], str(self.test_user.public_user_uuid)) + self.assertEqual(id_token["aud"], oidc_app.client_id) + self.assertIn("exp", id_token) + self.assertIn("iat", id_token) + + +class TestOIDCUserInfo(OIDCBaseTest): + """ + Test the OIDC UserInfo endpoint returns standard claims and is correctly + advertised in the discovery document. + """ + + def test_oidc_userinfo_endpoint(self): + """A token with openid scope returns standard OIDC claims from UserInfo.""" + self.access_token.scope = "openid profile email" + self.access_token.save() + + auth = self._create_authorization_header(self.access_token.token) + response = self.client.get("/oauth/oidc/userinfo", HTTP_AUTHORIZATION=auth) + self.assertEqual(response.status_code, 200) + data = response.json() + + self.assertEqual(data["sub"], str(self.test_user.public_user_uuid)) + self.assertIn("name", data) + self.assertIn("email", data) + + def test_oidc_userinfo_without_token(self): + """The UserInfo endpoint rejects unauthenticated requests.""" + response = self.client.get("/oauth/oidc/userinfo") + self.assertIn(response.status_code, [401, 403]) + + def test_oidc_discovery_advertises_correct_userinfo(self): + """The discovery document points clients at the OIDC userinfo path.""" + response = self.client.get("/.well-known/openid-configuration") + self.assertEqual(response.status_code, 200) + self.assertEqual( + response.json()["userinfo_endpoint"], + "http://testserver/oauth/oidc/userinfo", + ) + + +class TestLegacyUserInfoUnchanged(OIDCBaseTest): + """ + Test that adding the OIDC provider has not broken the pre-existing + /oauth/userinfo resource endpoint, which legacy clients continue to use. + """ + + def test_legacy_userinfo_still_works(self): + """Legacy /oauth/userinfo still returns the original (non-OIDC) format.""" + auth = self._create_authorization_header(self.access_token.token) + response = self.client.get("/oauth/userinfo", HTTP_AUTHORIZATION=auth) + self.assertEqual(response.status_code, 200) + data = response.json() + + # Legacy format uses "username" and "full_name", not OIDC claim names + self.assertIn("username", data) + self.assertIn("full_name", data) + self.assertNotIn("sub", data) + self.assertNotIn("preferred_username", data) + + def test_legacy_userinfo_with_trailing_slash_works(self): + """Both /oauth/userinfo and /oauth/userinfo/ resolve to the legacy view.""" + # Regression: /oauth/userinfo/ used to route to OIDCUserInfoView and + # reject legacy tokens with 401. + auth = self._create_authorization_header(self.access_token.token) + response = self.client.get("/oauth/userinfo/", HTTP_AUTHORIZATION=auth) + self.assertEqual(response.status_code, 200) + data = response.json() + self.assertIn("username", data) + self.assertNotIn("sub", data) + + +class TestOIDCDiscoveryRestrictsResponseTypes(OIDCBaseTest): + """ + Test that the discovery document advertises only the authorization-code + flow, since implicit and hybrid flows are deprecated by OAuth 2.1. + """ + + def test_response_types_supported_is_code_only(self): + """response_types_supported lists 'code' and nothing else.""" + response = self.client.get("/.well-known/openid-configuration") + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["response_types_supported"], ["code"]) + + +class TestOIDCKeyRotation(OIDCBaseTest): + """ + Test that rotated-out signing keys configured via OIDC_RSA_PRIVATE_KEYS_INACTIVE + are still published in JWKS so relying parties can verify tokens issued + under the previous key during the rotation overlap window. + """ + + def test_inactive_keys_are_published_in_jwks(self): + """JWKS publishes the active key plus every key listed in OIDC_RSA_PRIVATE_KEYS_INACTIVE.""" + # Generate an additional RSA key to act as a rotated-out signer + old_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + old_key_pem = old_key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ).decode() + + rotation_settings = { + **settings.OAUTH2_PROVIDER, + "OIDC_ENABLED": True, + "OIDC_RSA_PRIVATE_KEY": self.test_rsa_key_pem, + "OIDC_RSA_PRIVATE_KEYS_INACTIVE": [old_key_pem], + "OIDC_ISS_ENDPOINT": "http://testserver", + } + with override_settings(OAUTH2_PROVIDER=rotation_settings): + response = self.client.get("/oauth/jwks/") + + self.assertEqual(response.status_code, 200) + keys = response.json()["keys"] + self.assertEqual(len(keys), 2) + # Each published key carries a distinct kid + self.assertEqual(len({k["kid"] for k in keys}), 2) + + +class TestOIDCClaimScopes(OIDCBaseTest): + """ + Test that each OIDC scope contributes its expected claims to the UserInfo + response, covering the per-scope branches in CustomOAuth2Validator._build_claims. + """ + + def setUp(self): + super().setUp() + # Populate every profile field a claim might draw from + self.test_user.profile.website = "https://example.com/oauth_user" + self.test_user.profile.save() + + def _userinfo(self, scope): + """Hit the OIDC userinfo endpoint with a given scope and return the JSON.""" + self.access_token.scope = scope + self.access_token.save() + auth = self._create_authorization_header(self.access_token.token) + response = self.client.get("/oauth/oidc/userinfo", HTTP_AUTHORIZATION=auth) + self.assertEqual(response.status_code, 200) + return response.json() + + def test_profile_scope_returns_website(self): + """The profile scope includes the optional website claim when set.""" + data = self._userinfo("openid profile") + self.assertEqual(data["website"], "https://example.com/oauth_user") + self.assertEqual(data["preferred_username"], self.test_user.username) + + def test_email_scope_returns_email_verified(self): + """The email scope returns the email_verified claim alongside email.""" + data = self._userinfo("openid email") + self.assertIn("email", data) + self.assertIn("email_verified", data) + self.assertIsInstance(data["email_verified"], bool) + + def test_institution_scope_returns_affiliation(self): + """The institution:read scope returns the user's affiliation.""" + data = self._userinfo("openid institution:read") + self.assertEqual(data["affiliation"], "MIT") + + def test_credentialing_scope_returns_credentialing_status(self): + """The credentialing:read scope returns the user's credentialing status.""" + data = self._userinfo("openid credentialing:read") + self.assertIn("is_credentialed", data) + self.assertIsInstance(data["is_credentialed"], bool) + + def test_public_id_scope_returns_uuid(self): + """The public_id:read scope returns the persistent public UUID.""" + data = self._userinfo("openid public_id:read") + self.assertEqual(data["public_user_uuid"], str(self.test_user.public_user_uuid)) + + def test_no_scope_returns_only_sub(self): + """A token with no claim-bearing scopes still gets a sub claim.""" + data = self._userinfo("openid") + self.assertEqual(data["sub"], str(self.test_user.public_user_uuid)) + self.assertNotIn("email", data) + self.assertNotIn("affiliation", data) + self.assertNotIn("name", data) + + +class TestOIDCProviderConfigValidation(TestCase): + """ + Test that load_oidc_provider_config raises loudly on misconfiguration so + the OIDC provider cannot silently fall back to a degraded state. + """ + + def _env(self, **values): + """Return a get_env(name, default) callable backed by a static dict.""" + return lambda name, default='': values.get(name, default) + + def test_missing_key_file_raises(self): + """A non-existent OIDC_RSA_KEY_FILE path raises ImproperlyConfigured.""" + with self.assertRaisesMessage(ImproperlyConfigured, 'no such file exists'): + load_oidc_provider_config(self._env( + OIDC_RSA_KEY_FILE='/nonexistent/oidc-key.pem', + OIDC_ISS_ENDPOINT='https://example.org/oauth', + )) + + def test_missing_inactive_key_file_raises(self): + """A non-existent path in OIDC_RSA_INACTIVE_KEY_FILES raises ImproperlyConfigured.""" + with self.assertRaisesMessage(ImproperlyConfigured, 'OIDC_RSA_INACTIVE_KEY_FILES'): + load_oidc_provider_config(self._env( + OIDC_RSA_INACTIVE_KEY_FILES='/nonexistent/old-key.pem', + OIDC_ISS_ENDPOINT='https://example.org/oauth', + )) + + def test_missing_iss_endpoint_with_key_raises(self): + """A signing key without OIDC_ISS_ENDPOINT raises ImproperlyConfigured.""" + with self.assertRaisesMessage(ImproperlyConfigured, 'OIDC_ISS_ENDPOINT'): + load_oidc_provider_config(self._env( + OIDC_RSA_PRIVATE_KEY='-----BEGIN PRIVATE KEY-----\nfake\n-----END PRIVATE KEY-----', + OIDC_ISS_ENDPOINT='', + )) + + def test_no_key_returns_disabled_config(self): + """With no key configured, OIDC is disabled and no validation fires.""" + cfg = load_oidc_provider_config(self._env()) + self.assertFalse(cfg['OIDC_ENABLED']) + self.assertEqual(cfg['OIDC_RSA_PRIVATE_KEY'], '') + self.assertEqual(cfg['OIDC_RSA_PRIVATE_KEYS_INACTIVE'], []) + + def test_inline_key_with_iss_returns_enabled_config(self): + """An inline OIDC_RSA_PRIVATE_KEY with OIDC_ISS_ENDPOINT enables OIDC.""" + cfg = load_oidc_provider_config(self._env( + OIDC_RSA_PRIVATE_KEY='inline-pem', + OIDC_ISS_ENDPOINT='https://example.org/oauth', + )) + self.assertTrue(cfg['OIDC_ENABLED']) + self.assertEqual(cfg['OIDC_RSA_PRIVATE_KEY'], 'inline-pem') + self.assertEqual(cfg['OIDC_ISS_ENDPOINT'], 'https://example.org/oauth') + + def test_key_file_is_read_from_disk(self): + """A valid OIDC_RSA_KEY_FILE is read and returned as the private key.""" + with tempfile.NamedTemporaryFile('w', suffix='.pem', delete=False) as f: + f.write('contents-of-the-key-file') + key_path = f.name + try: + cfg = load_oidc_provider_config(self._env( + OIDC_RSA_KEY_FILE=key_path, + OIDC_ISS_ENDPOINT='https://example.org/oauth', + )) + self.assertEqual(cfg['OIDC_RSA_PRIVATE_KEY'], 'contents-of-the-key-file') + finally: + os.unlink(key_path) + + def test_inactive_keys_are_loaded_in_order(self): + """Each path in OIDC_RSA_INACTIVE_KEY_FILES is loaded into the inactive list.""" + with tempfile.NamedTemporaryFile('w', suffix='.pem', delete=False) as f1: + f1.write('old-key-1') + path1 = f1.name + with tempfile.NamedTemporaryFile('w', suffix='.pem', delete=False) as f2: + f2.write('old-key-2') + path2 = f2.name + try: + cfg = load_oidc_provider_config(self._env( + OIDC_RSA_PRIVATE_KEY='active', + OIDC_RSA_INACTIVE_KEY_FILES=f'{path1},{path2}', + OIDC_ISS_ENDPOINT='https://example.org/oauth', + )) + self.assertEqual(cfg['OIDC_RSA_PRIVATE_KEYS_INACTIVE'], ['old-key-1', 'old-key-2']) + finally: + os.unlink(path1) + os.unlink(path2) + + +class TestGenerateOIDCRSAKeyCommand(TestCase): + """ + Test the generate_oidc_rsa_key management command that operators run to + create a signing key for the OIDC provider. + """ + + def test_writes_pkcs8_pem_to_stdout_by_default(self): + """Without --output, the key is written to stdout in PKCS8 PEM format.""" + stdout = StringIO() + stderr = StringIO() + with mock.patch('sys.stdout.isatty', return_value=False): + call_command('generate_oidc_rsa_key', stdout=stdout, stderr=stderr) + pem = stdout.getvalue() + self.assertIn('-----BEGIN PRIVATE KEY-----', pem) + self.assertIn('-----END PRIVATE KEY-----', pem) + # PKCS8-formatted RSA keys load successfully via cryptography + serialization.load_pem_private_key(pem.encode(), password=None) + + def test_output_file_is_created_with_0600_permissions(self): + """--output writes the key with restrictive 0600 permissions.""" + with tempfile.TemporaryDirectory() as tmp: + target = os.path.join(tmp, 'oidc.pem') + call_command('generate_oidc_rsa_key', '--output', target) + self.assertTrue(os.path.isfile(target)) + mode = os.stat(target).st_mode & 0o777 + self.assertEqual(mode, 0o600) + with open(target) as f: + self.assertIn('-----BEGIN PRIVATE KEY-----', f.read()) + + def test_rejects_key_size_below_2048(self): + """--bits below 2048 is rejected with an error to stderr.""" + stdout = StringIO() + stderr = StringIO() + call_command('generate_oidc_rsa_key', '--bits', '1024', stdout=stdout, stderr=stderr) + self.assertIn('Key size must be at least 2048 bits', stderr.getvalue()) + self.assertEqual(stdout.getvalue(), '') + + +class TestCustomValidatorAdditionalClaims(OIDCBaseTest): + """ + Test the validator-level helpers on CustomOAuth2Validator that don't + have natural HTTP-level coverage. + """ + + def test_get_additional_claims_returns_empty_when_no_user(self): + """get_additional_claims returns {} when the request has no user.""" + from oauth.validators import CustomOAuth2Validator + validator = CustomOAuth2Validator() + request = mock.MagicMock() + request.user = None + self.assertEqual(validator.get_additional_claims(request), {}) + + +class TestTokenIntrospection(BaseTest): + """ + Test the RFC 7662 token introspection endpoint at /oauth/introspect/. + + Resource servers POST an opaque access token plus their client credentials + and learn whether the token is currently active and, if so, which scopes, + user, and client it was issued for. Inactive (expired, revoked, unknown) + tokens must surface as {"active": false} without leaking metadata, and + requests with bad client credentials must be rejected outright. + """ + + def _post_introspect(self, token, client_id, client_secret): + """Hit the introspection endpoint with HTTP Basic client auth.""" + auth_headers = self.get_basic_auth_header(client_id, client_secret) + return self.client.post( + reverse("oauth2_provider:introspect"), + data={"token": token}, + **auth_headers, + ) + + def test_active_token_returns_claims(self): + """An active token returns active=true with scope, client_id, username, exp.""" + response = self._post_introspect( + self.access_token.token, + self.application.client_id, + CLEARTEXT_SECRET, + ) + self.assertEqual(response.status_code, 200) + data = response.json() + self.assertTrue(data["active"]) + self.assertEqual(data["scope"], self.access_token.scope) + self.assertEqual(data["client_id"], self.application.client_id) + self.assertEqual(data["username"], self.test_user.username) + self.assertIn("exp", data) + + def test_expired_token_returns_inactive(self): + """An expired token introspects as active=false.""" + self.access_token.expires = timezone.now() - timedelta(hours=1) + self.access_token.save() + response = self._post_introspect( + self.access_token.token, + self.application.client_id, + CLEARTEXT_SECRET, + ) + self.assertEqual(response.status_code, 200) + self.assertFalse(response.json()["active"]) + + def test_revoked_token_returns_inactive(self): + """A deleted/revoked token introspects as active=false.""" + token_value = self.access_token.token + self.access_token.delete() + response = self._post_introspect( + token_value, + self.application.client_id, + CLEARTEXT_SECRET, + ) + self.assertEqual(response.status_code, 200) + self.assertFalse(response.json()["active"]) + + def test_unknown_token_returns_inactive(self): + """A token string that was never issued introspects as active=false.""" + response = self._post_introspect( + "not-a-real-token-string", + self.application.client_id, + CLEARTEXT_SECRET, + ) + self.assertEqual(response.status_code, 200) + self.assertFalse(response.json()["active"]) + + def test_bad_client_credentials_are_rejected(self): + """Wrong client secret is rejected, not silently downgraded.""" + # RFC 7662 calls for 401 + WWW-Authenticate, but DOT 2.2.0's + # ClientProtectedResourceMixin.dispatch returns 403. Accept either + # so we don't break on a DOT upgrade that brings this in line. + response = self._post_introspect( + self.access_token.token, + self.application.client_id, + "wrong-secret", + ) + self.assertIn(response.status_code, (401, 403)) + + +class TestApplicationsViewsRemoved(TestCase): + """ + Regression test: the django-oauth-toolkit-shipped /oauth/applications/* + views are not reachable. Operators manage Applications via the Console + workflow (Stage C) and Django admin only. + """ + + def test_applications_list_is_404(self): + """GET /oauth/applications/ returns 404.""" + response = self.client.get("/oauth/applications/") + self.assertEqual(response.status_code, 404) + + def test_applications_register_is_404(self): + """GET /oauth/applications/register/ returns 404.""" + response = self.client.get("/oauth/applications/register/") + self.assertEqual(response.status_code, 404) + + +class TestPartnerModel(TestCase): + """ + Test the Partner model that augments oauth2_provider.Application with + organization metadata, scope allow-listing, post-logout redirect URIs, + and a lifecycle status used by the Console partner-management workflow. + """ + + def setUp(self): + self.user = User.objects.create_user( + username="partner_owner", + email="partner_owner@example.com", + password=get_random_string(20), + ) + self.application = Application.objects.create( + name="Acme App", + redirect_uris="http://example.org", + user=self.user, + client_type=Application.CLIENT_CONFIDENTIAL, + authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE, + client_secret=CLEARTEXT_SECRET, + ) + + def test_str_includes_org_name_and_client_id(self): + """__str__ renders as 'organization_name (client_id)'.""" + partner = Partner.objects.create( + application=self.application, + organization_name="Acme", + created_by=self.user, + ) + self.assertEqual(str(partner), f"Acme ({self.application.client_id})") + + def test_status_defaults_to_active(self): + """A newly created Partner has status=ACTIVE without explicit value.""" + partner = Partner.objects.create( + application=self.application, + organization_name="Acme", + created_by=self.user, + ) + self.assertEqual(partner.status, Partner.Status.ACTIVE) + + def test_is_legacy_property(self): + """is_legacy is True iff organization_name starts with 'Legacy: '.""" + legacy = Partner.objects.create( + application=self.application, + organization_name=f"Legacy: {self.application.client_id}", + created_by=self.user, + ) + self.assertTrue(legacy.is_legacy) + + other_app = Application.objects.create( + name="Acme App 2", + redirect_uris="http://example.org", + user=self.user, + client_type=Application.CLIENT_CONFIDENTIAL, + authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE, + client_secret=CLEARTEXT_SECRET, + ) + normal = Partner.objects.create( + application=other_app, + organization_name="Acme", + created_by=self.user, + ) + self.assertFalse(normal.is_legacy) + + def test_cascade_delete_from_application(self): + """Deleting the linked Application cascades to delete the Partner.""" + partner = Partner.objects.create( + application=self.application, + organization_name="Acme", + created_by=self.user, + ) + partner_pk = partner.pk + self.application.delete() + self.assertFalse(Partner.objects.filter(pk=partner_pk).exists()) + + def test_allowed_scopes_round_trip(self): + """allowed_scopes survives a save/refresh_from_db cycle intact.""" + partner = Partner.objects.create( + application=self.application, + organization_name="Acme", + allowed_scopes=["openid", "profile", "email", "data:download"], + created_by=self.user, + ) + partner.refresh_from_db() + self.assertEqual( + partner.allowed_scopes, + ["openid", "profile", "email", "data:download"], + ) + + def test_post_logout_redirect_uris_default_blank(self): + """post_logout_redirect_uris defaults to an empty string.""" + partner = Partner.objects.create( + application=self.application, + organization_name="Acme", + created_by=self.user, + ) + self.assertEqual(partner.post_logout_redirect_uris, "") + + +class TestLegacyPartnerBackfill(TestCase): + """ + Test the data-migration that creates 'Legacy: ' Partner rows + for every Application present at deploy time. The migration runs as part + of test database setup, so we just inspect the resulting state. + """ + + def setUp(self): + super().setUp() + # Create a fresh Application *after* migration has run, then verify + # the backfill logic by re-running it on this row. + self.user = User.objects.create_user( + username="backfill_admin", + email="backfill@example.com", + password=get_random_string(20), + ) + self.application = Application.objects.create( + name="Pre-existing legacy app", + redirect_uris="https://legacy.example.com/cb", + user=self.user, + client_type=Application.CLIENT_CONFIDENTIAL, + authorization_grant_type=Application.GRANT_AUTHORIZATION_CODE, + ) + + def _backfill(self): + """Import and call the data migration's backfill function.""" + import importlib + from django.apps import apps as django_apps + migration = importlib.import_module("oauth.migrations.0003_backfill_legacy_partners") + migration.backfill_legacy_partners(django_apps, None) + + def test_backfill_creates_legacy_partner_for_application(self): + """Running the backfill creates a Legacy Partner for an unattached Application.""" + self._backfill() + partner = Partner.objects.get(application=self.application) + self.assertTrue(partner.is_legacy) + self.assertEqual(partner.organization_name, f"Legacy: {self.application.client_id}") + self.assertEqual(partner.status, Partner.Status.ACTIVE) + self.assertEqual(partner.allowed_scopes, []) + self.assertIsNone(partner.created_by) + + def test_backfill_is_idempotent(self): + """Running the backfill twice doesn't create duplicate Partners.""" + self._backfill() + self._backfill() + self.assertEqual( + Partner.objects.filter(application=self.application).count(), 1, + ) + + +class TestPartnerAdmin(TestCase): + """ + Smoke test that the Partner ModelAdmin is registered and gated behind + Django's staff-access requirement. + """ + + def setUp(self): + super().setUp() + # The project's create_superuser sets is_admin only, not is_superuser, + # so set is_superuser explicitly so /admin/ access is granted. + self.staff_user = User.objects.create_user( + username="staff_admin", + email="staff@example.com", + password=get_random_string(20), + is_admin=True, + ) + self.staff_user.is_superuser = True + self.staff_user.save() + + def test_partner_admin_is_registered_and_reachable(self): + """A staff user can access the Partner admin changelist.""" + self.client.force_login(self.staff_user) + response = self.client.get("/admin/oauth/partner/") + self.assertEqual(response.status_code, 200) + + def test_partner_admin_denied_to_anonymous(self): + """Anonymous users are redirected to the admin login page.""" + response = self.client.get("/admin/oauth/partner/") + self.assertEqual(response.status_code, 302) + self.assertIn("/admin/login/", response["Location"]) + + +class TestEndSessionView(OIDCBaseTest): + """ + Test the OIDC RP-initiated logout endpoint at /oauth/end-session/. + + Validates that id_token_hint is verified against active and rotated-out + signing keys, post_logout_redirect_uri is honoured only when registered + on the issuing Partner, and the user's Django session is cleared in all + cases (whether or not a redirect can be made). + """ + + END_SESSION_URL = "/oauth/end-session/" + REGISTERED_REDIRECT = "https://acme.example.com/logged-out" + + def setUp(self): + super().setUp() + self.partner_owner = User.objects.create_user( + username="end_session_owner", + email="end_session_owner@example.com", + password=get_random_string(20), + ) + self.partner = Partner.objects.create( + application=self.application, + organization_name="Acme", + post_logout_redirect_uris=self.REGISTERED_REDIRECT, + created_by=self.partner_owner, + ) + + def _mint_jwt(self, pem, claims=None): + """Sign a JWT with the given PEM key and default test claims.""" + payload = { + "aud": self.application.client_id, + "sub": str(self.test_user.public_user_uuid), + } + if claims: + payload.update(claims) + return jwt.encode(payload, pem, algorithm="RS256") + + def _login(self): + """Log the test user in and assert the session was created.""" + self.assertTrue( + self.client.login(username=self.test_user.username, password="123456") + ) + self.assertIn("_auth_user_id", self.client.session) + + def test_valid_hint_and_registered_redirect_logs_out_and_redirects(self): + """Valid hint + registered redirect logs the user out and 302s back.""" + token = self._mint_jwt(self.test_rsa_key_pem) + self._login() + response = self.client.get( + self.END_SESSION_URL, + data={ + "id_token_hint": token, + "post_logout_redirect_uri": self.REGISTERED_REDIRECT, + }, + ) + self.assertEqual(response.status_code, 302) + self.assertEqual(response["Location"], self.REGISTERED_REDIRECT) + self.assertNotIn("_auth_user_id", self.client.session) + + def test_unregistered_redirect_is_dropped(self): + """Unregistered post_logout_redirect_uri is rejected; session still cleared.""" + token = self._mint_jwt(self.test_rsa_key_pem) + self._login() + response = self.client.get( + self.END_SESSION_URL, + data={ + "id_token_hint": token, + "post_logout_redirect_uri": "https://evil.example.com/phish", + }, + ) + self.assertEqual(response.status_code, 200) + self.assertNotIn("_auth_user_id", self.client.session) + + def test_no_hint_logs_out_and_renders_fallback(self): + """With no params, the user is logged out and the fallback page renders.""" + self._login() + response = self.client.get(self.END_SESSION_URL) + self.assertEqual(response.status_code, 200) + self.assertNotIn("_auth_user_id", self.client.session) + + def test_forged_hint_is_rejected(self): + """A hint signed by an unknown key cannot drive a redirect.""" + forged_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + forged_pem = forged_key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ).decode() + token = self._mint_jwt(forged_pem) + self._login() + response = self.client.get( + self.END_SESSION_URL, + data={ + "id_token_hint": token, + "post_logout_redirect_uri": self.REGISTERED_REDIRECT, + }, + ) + self.assertEqual(response.status_code, 200) + self.assertNotIn("_auth_user_id", self.client.session) + + def test_inactive_key_signed_hint_is_accepted(self): + """A hint signed by a rotated-out (inactive) key still verifies.""" + old_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + old_pem = old_key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ).decode() + token = self._mint_jwt(old_pem) + rotation_settings = { + **settings.OAUTH2_PROVIDER, + "OIDC_RSA_PRIVATE_KEYS_INACTIVE": [old_pem], + } + with override_settings(OAUTH2_PROVIDER=rotation_settings): + self._login() + response = self.client.get( + self.END_SESSION_URL, + data={ + "id_token_hint": token, + "post_logout_redirect_uri": self.REGISTERED_REDIRECT, + }, + ) + self.assertEqual(response.status_code, 302) + self.assertEqual(response["Location"], self.REGISTERED_REDIRECT) + self.assertNotIn("_auth_user_id", self.client.session) + + def test_expired_hint_is_still_accepted_as_identity(self): + """An expired hint is still honoured: it identifies who to log out.""" + token = self._mint_jwt(self.test_rsa_key_pem, claims={"exp": 0}) + self._login() + response = self.client.get( + self.END_SESSION_URL, + data={ + "id_token_hint": token, + "post_logout_redirect_uri": self.REGISTERED_REDIRECT, + }, + ) + self.assertEqual(response.status_code, 302) + self.assertEqual(response["Location"], self.REGISTERED_REDIRECT) + self.assertNotIn("_auth_user_id", self.client.session) + + +class TestPartnerScopeAndStatusEnforcement(BaseTest): + """ + Test that CustomOAuth2Validator enforces per-partner scope allowlists + and active/suspended/revoked status at /authorize, /token, and + /introspect. + """ + + def setUp(self): + super().setUp() + from oauth.models import Partner + self.partner = Partner.objects.create( + application=self.application, + organization_name="Acme", + allowed_scopes=["openid", "profile", "email", "profile:read", "email:read"], + created_by=self.dev_user, + ) + + def _try_authorize(self, scope, pkce=False): + """Attempt the /authorize step and return the response.""" + self.client.login(username=self.test_user.username, password="123456") + data = { + "client_id": self.application.client_id, + "state": "x", + "scope": scope, + "redirect_uri": "http://example.org", + "response_type": "code", + "allow": True, + } + non_pkce = {**settings.OAUTH2_PROVIDER, "PKCE_REQUIRED": False} + with override_settings(OAUTH2_PROVIDER=non_pkce): + return self.client.post(reverse("oauth2_provider:authorize"), data=data) + + def test_in_allowlist_scope_yields_code(self): + """Requesting an in-allowlist scope returns an auth code.""" + response = self._try_authorize("openid profile") + self.assertEqual(response.status_code, 302) + self.assertIn("code=", response["Location"]) + + def test_out_of_allowlist_scope_is_invalid_scope(self): + """Requesting a scope not in allowed_scopes returns invalid_scope.""" + response = self._try_authorize("openid profile data:download") + self.assertEqual(response.status_code, 302) + self.assertIn("error=invalid_scope", response["Location"]) + + def test_wildcard_allowed_scopes_permits_any_global_scope(self): + """Empty allowed_scopes is treated as wildcard (legacy compat).""" + self.partner.allowed_scopes = [] + self.partner.save() + response = self._try_authorize("openid profile data:download") + self.assertEqual(response.status_code, 302) + self.assertIn("code=", response["Location"]) + + def test_suspended_partner_authorize_is_unauthorized_client(self): + """A suspended partner is rejected at /authorize without consent screen. + + validate_client_id failure causes oauthlib to render an error page + (HTTP 400) rather than redirect, since no redirect_uri can be trusted + when the client_id itself is invalid. + """ + from oauth.models import Partner + self.partner.status = Partner.Status.SUSPENDED + self.partner.save() + response = self._try_authorize("openid") + self.assertEqual(response.status_code, 400) + + def test_suspended_partner_token_exchange_returns_invalid_client(self): + """An auth code obtained while active cannot be exchanged after suspension.""" + from oauth.models import Partner + # First, get a code while active + self.partner.status = Partner.Status.ACTIVE + self.partner.save() + response = self._try_authorize("openid") + code = parse_qs(urlparse(response["Location"]).query)["code"][0] + + # Suspend the partner + self.partner.status = Partner.Status.SUSPENDED + self.partner.save() + + non_pkce = {**settings.OAUTH2_PROVIDER, "PKCE_REQUIRED": False} + with override_settings(OAUTH2_PROVIDER=non_pkce): + token_response = self.client.post( + reverse("oauth2_provider:token"), + data={ + "grant_type": "authorization_code", + "code": code, + "redirect_uri": "http://example.org", + }, + **self.get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET), + ) + self.assertEqual(token_response.status_code, 401) + self.assertEqual(token_response.json()["error"], "invalid_client") + + def test_suspended_partner_cannot_introspect(self): + """A suspended partner is rejected at /oauth/introspect/.""" + from oauth.models import Partner + self.partner.status = Partner.Status.SUSPENDED + self.partner.save() + response = self.client.post( + reverse("oauth2_provider:introspect"), + data={"token": self.access_token.token}, + **self.get_basic_auth_header(self.application.client_id, CLEARTEXT_SECRET), + ) + # DOT 2.2.0's ClientProtectedResourceMixin returns 403 on bad client auth + self.assertIn(response.status_code, (401, 403)) diff --git a/physionet-django/oauth/urls.py b/physionet-django/oauth/urls.py index 0d0d0cac11..ace9204173 100644 --- a/physionet-django/oauth/urls.py +++ b/physionet-django/oauth/urls.py @@ -1,55 +1,22 @@ from django.urls import path, include import oauth2_provider.views as oauth2_views -from django.conf import settings -from oauth.views import hello, UserInfoView +from oauth2_provider.views import JwksInfoView +from oauth2_provider.views import UserInfoView as OIDCUserInfoView +from oauth.views import hello, UserInfoView, PhysioNetDiscoveryView, EndSessionView -# OAuth2 provider endpoints +# OAuth2 provider endpoints. The OIDC userinfo and JWKS endpoints live here so +# DOT's discovery view (which does reverse("oauth2_provider:user-info") / +# reverse("oauth2_provider:jwks-info")) can resolve them inside the namespace. oauth2_endpoint_views = [ path("authorize/", oauth2_views.AuthorizationView.as_view(), name="authorize"), path("token/", oauth2_views.TokenView.as_view(), name="token"), path("revoke-token/", oauth2_views.RevokeTokenView.as_view(), name="revoke-token"), + path("introspect/", oauth2_views.IntrospectTokenView.as_view(), name="introspect"), + path("jwks/", JwksInfoView.as_view(), name="jwks-info"), + path("oidc/userinfo", OIDCUserInfoView.as_view(), name="user-info"), + path("end-session/", EndSessionView.as_view(), name="end-session"), ] -if settings.DEBUG: - # OAuth2 Application Management endpoints - oauth2_endpoint_views += [ - path("applications/", oauth2_views.ApplicationList.as_view(), name="list"), - path( - "applications//", - oauth2_views.ApplicationDetail.as_view(), - name="detail", - ), - path( - "applications//delete/", - oauth2_views.ApplicationDelete.as_view(), - name="delete", - ), - path( - "applications//update/", - oauth2_views.ApplicationUpdate.as_view(), - name="update", - ), - path( - "applications/register/", - oauth2_views.ApplicationRegistration.as_view(), - name="register", - ), - ] - - # OAuth2 Token Management endpoints - oauth2_endpoint_views += [ - path( - "authorized-tokens/", - oauth2_views.AuthorizedTokensListView.as_view(), - name="authorized-token-list", - ), - path( - "authorized-tokens//delete/", - oauth2_views.AuthorizedTokenDeleteView.as_view(), - name="authorized-token-delete", - ), - ] - urlpatterns = [ # OAuth 2 endpoints: path( @@ -58,6 +25,10 @@ (oauth2_endpoint_views, "oauth2_provider"), namespace="oauth2_provider" ), ), - path("hello", hello.as_view(), name="hello"), # an example resource endpoint + path("hello", hello.as_view(), name="hello"), + # Both slash forms route to the legacy view; OIDC userinfo lives under + # oidc/ to avoid silently rejecting legacy tokens sent to /oauth/userinfo/. path("userinfo", UserInfoView.as_view(), name="userinfo"), + path("userinfo/", UserInfoView.as_view()), + path(".well-known/openid-configuration", PhysioNetDiscoveryView.as_view(), name="oidc-connect-discovery-info"), ] diff --git a/physionet-django/oauth/validators.py b/physionet-django/oauth/validators.py new file mode 100644 index 0000000000..5cef187e7a --- /dev/null +++ b/physionet-django/oauth/validators.py @@ -0,0 +1,128 @@ +from oauth2_provider.models import get_application_model +from oauth2_provider.oauth2_validators import OAuth2Validator + +from oauth.models import Partner +from user.models import AssociatedEmail + +Application = get_application_model() + + +class CustomOAuth2Validator(OAuth2Validator): + """ + Provide PhysioNet-specific OIDC claims for ID tokens and the UserInfo + endpoint, keyed off the granted OAuth2 scopes. + """ + + def _get_sub(self, user): + """Return the stable subject identifier for an OIDC token.""" + return str(user.public_user_uuid) + + def _build_claims(self, user, scopes): + """Build the claim set for a user filtered by the granted scopes.""" + claims = {"sub": self._get_sub(user)} + + if "profile" in scopes or "profile:read" in scopes: + profile = getattr(user, 'profile', None) + if profile: + claims["name"] = user.get_full_name() + claims["given_name"] = profile.first_names + claims["family_name"] = profile.last_name + claims["preferred_username"] = user.username + if profile.website: + claims["website"] = profile.website + + if "email" in scopes or "email:read" in scopes: + try: + primary_email = user.get_primary_email() + except AssociatedEmail.DoesNotExist: + primary_email = None + if primary_email: + claims["email"] = primary_email.email + claims["email_verified"] = primary_email.is_verified + + if "institution:read" in scopes: + profile = getattr(user, 'profile', None) + if profile: + claims["affiliation"] = profile.affiliation + + if "credentialing:read" in scopes: + claims["is_credentialed"] = user.is_credentialed + + if "orcid:read" in scopes: + claims["orcid"] = user.get_orcid_id() + + if "public_id:read" in scopes: + claims["public_user_uuid"] = str(user.public_user_uuid) + + return claims + + def validate_scopes(self, client_id, scopes, client, request, *args, **kwargs): + """Enforce per-partner scope allowlist on top of DOT's global SCOPES check.""" + if not super().validate_scopes(client_id, scopes, client, request, *args, **kwargs): + return False + partner = getattr(client, 'partner', None) + if partner is None: + return True # Application predates Partner model; legacy compat + if not partner.allowed_scopes: + return True # empty list = wildcard, intentional for legacy + return set(scopes).issubset(set(partner.allowed_scopes)) + + def validate_client_id(self, client_id, request, *args, **kwargs): + """Reject /authorize for suspended/revoked partners before the consent screen.""" + if not super().validate_client_id(client_id, request, *args, **kwargs): + return False + partner = getattr(request.client, 'partner', None) + if partner and partner.status != Partner.Status.ACTIVE: + return False + return True + + def authenticate_client(self, request, *args, **kwargs): + """Block client authentication for suspended/revoked partners at /token, /introspect, /revoke-token.""" + if not super().authenticate_client(request, *args, **kwargs): + return False + partner = getattr(request.client, 'partner', None) + if partner and partner.status != Partner.Status.ACTIVE: + return False + return True + + def is_pkce_required(self, client_id, *args, **kwargs): + """Allow per-partner PKCE opt-out. + + OAuth 2.1 / RFC 9700 require PKCE; the global PKCE_REQUIRED setting + (default True) is honored first. When the global says "require", + an individual partner with Partner.requires_pkce=False can still + opt out — used when the upstream IdP can't send a code_challenge. + When the global says "don't require" (typically dev/test), nobody + requires it. + """ + global_required = super().is_pkce_required(client_id, *args, **kwargs) + if not global_required: + return False + try: + app = Application.objects.get(client_id=client_id) + except Application.DoesNotExist: + return True + partner = getattr(app, 'partner', None) + if partner is None: + return True + return partner.requires_pkce + + def get_additional_claims(self, request): + """Return the additional claims to embed in the ID token.""" + if not request.user: + return {} + scopes = set(getattr(request, 'scopes', []) or []) + return self._build_claims(request.user, scopes) + + def get_userinfo_claims(self, request): + """Return the claims served from the OIDC UserInfo endpoint. + + Bypasses DOT's default oidc_claim_scope filter so PhysioNet-specific + scopes (institution:read, credentialing:read, orcid:read, + public_id:read, profile:read, email:read) propagate through; DOT's + filter only knows about its built-in OIDC standard claims. + """ + if not (hasattr(request, 'user') and request.user): + return super().get_userinfo_claims(request) + scopes = set(getattr(request, 'scopes', []) or []) + return self._build_claims(request.user, scopes) diff --git a/physionet-django/oauth/views.py b/physionet-django/oauth/views.py index a8bb430f1f..92610498d7 100644 --- a/physionet-django/oauth/views.py +++ b/physionet-django/oauth/views.py @@ -1,6 +1,112 @@ +import json + from django.http import HttpResponse, JsonResponse +from django.urls import reverse from oauth2_provider.views.generic import ProtectedResourceView, ScopedProtectedResourceView from oauth2_provider.oauth2_backends import get_oauthlib_core +from oauth2_provider.settings import oauth2_settings +from oauth2_provider.views import ConnectDiscoveryInfoView + +import jwt +from django.contrib.auth import logout +from django.shortcuts import redirect, render +from django.views import View +from oauth2_provider.models import get_application_model +from cryptography.hazmat.primitives.serialization import load_pem_private_key + +Application = get_application_model() + + +def _signing_key_pems(): + """Return active + rotated-out OIDC signing keys (PEM strings).""" + keys = [] + if oauth2_settings.OIDC_RSA_PRIVATE_KEY: + keys.append(oauth2_settings.OIDC_RSA_PRIVATE_KEY) + keys.extend(getattr(oauth2_settings, 'OIDC_RSA_PRIVATE_KEYS_INACTIVE', []) or []) + return keys + + +def _decode_id_token(id_token_hint): + """Verify id_token_hint against any current signing key. + + exp is intentionally not enforced: a recently expired token is still a + valid identity hint for "log this user out". verify_aud is False because + PyJWT raises InvalidAudienceError otherwise when the JWT has an `aud` + claim and no expected audience is passed; we read the aud claim + downstream to look up the issuing Application, which then anchors the + post_logout_redirect_uri allowlist check. + """ + last_error = None + for pem in _signing_key_pems(): + try: + private_key = load_pem_private_key(pem.encode(), password=None) + return jwt.decode( + id_token_hint, + private_key.public_key(), + algorithms=["RS256"], + options={"verify_signature": True, "verify_exp": False, "verify_aud": False}, + ) + except jwt.PyJWTError as e: + last_error = e + raise last_error or jwt.InvalidTokenError("no signing keys configured") + + +class EndSessionView(View): + """OIDC RP-initiated logout (OpenID Connect Session Management 1.0).""" + + def get(self, request): + return self._handle(request) + + def post(self, request): + return self._handle(request) + + def _handle(self, request): + id_token_hint = request.GET.get("id_token_hint") or request.POST.get("id_token_hint") + post_logout = ( + request.GET.get("post_logout_redirect_uri") + or request.POST.get("post_logout_redirect_uri", "") + ) + + application = None + if id_token_hint: + try: + claims = _decode_id_token(id_token_hint) + application = Application.objects.filter(client_id=claims.get("aud")).first() + except jwt.PyJWTError: + application = None + + # post_logout_redirect_uri honoured only if the issuing partner registered it + if post_logout and application: + partner = getattr(application, "partner", None) + allowed = (partner.post_logout_redirect_uris.split() if partner else []) + if post_logout not in allowed: + post_logout = "" + else: + post_logout = "" + + logout(request) + if post_logout: + return redirect(post_logout) + return render(request, "oauth/logged_out.html") + + +class PhysioNetDiscoveryView(ConnectDiscoveryInfoView): + """ + Extends DOT's discovery view to publish endpoints DOT 2.2.0 omits from + the response (introspection_endpoint). + """ + + def get(self, request, *args, **kwargs): + response = super().get(request, *args, **kwargs) + if response.status_code != 200: + return response # OIDC disabled / OIDCOnlyMixin returned 404 + data = json.loads(response.content) + iss = oauth2_settings.OIDC_ISS_ENDPOINT.rstrip("/") + data["introspection_endpoint"] = iss + reverse("oauth2_provider:introspect") + data["end_session_endpoint"] = iss + reverse("oauth2_provider:end-session") + new_response = JsonResponse(data) + new_response["Access-Control-Allow-Origin"] = "*" + return new_response SCOPES_MAPPING = { diff --git a/physionet-django/physionet/settings/base.py b/physionet-django/physionet/settings/base.py index 5db15ea919..89eb977828 100644 --- a/physionet-django/physionet/settings/base.py +++ b/physionet-django/physionet/settings/base.py @@ -16,6 +16,7 @@ import sys from decouple import config, UndefinedValueError +from django.core.exceptions import ImproperlyConfigured # Build paths inside the project like this: os.path.join(BASE_DIR, ...) BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) @@ -814,7 +815,57 @@ class StorageTypes: # when programmatically generating access tokens (e.g., via the /settings/tokens). OAUTH_CLIENT_APP_NAME = config('OAUTH_CLIENT_APP_NAME', default='') -# OAUTH PROVIDER SCOPES + +def load_oidc_provider_config(get_env): + """ + Load and validate the OIDC provider settings from a get_env(name, default) + callable (e.g. python-decouple's `config`). Returns a dict suitable for + merging into OAUTH2_PROVIDER, or raises ImproperlyConfigured if the + operator has misconfigured the provider. + """ + key_file = get_env('OIDC_RSA_KEY_FILE', '') + if key_file: + if not os.path.isfile(key_file): + raise ImproperlyConfigured( + f"OIDC_RSA_KEY_FILE is set to {key_file!r} but no such file exists." + ) + with open(key_file) as f: + private_key = f.read() + else: + private_key = get_env('OIDC_RSA_PRIVATE_KEY', '') + + # Rotated-out keys are published in JWKS so RPs can verify tokens issued under + # the previous active key during the rotation overlap window. + inactive_files = [ + p.strip() for p in get_env('OIDC_RSA_INACTIVE_KEY_FILES', '').split(',') if p.strip() + ] + inactive_keys = [] + for path in inactive_files: + if not os.path.isfile(path): + raise ImproperlyConfigured( + f"OIDC_RSA_INACTIVE_KEY_FILES references {path!r} but no such file exists." + ) + with open(path) as f: + inactive_keys.append(f.read()) + + iss_endpoint = get_env('OIDC_ISS_ENDPOINT', '') + if private_key and not iss_endpoint: + raise ImproperlyConfigured( + "OIDC is enabled (RSA key configured) but OIDC_ISS_ENDPOINT is not set. " + "It must be the canonical public host with no path component " + "(e.g. https://physionet.org), since the discovery view appends the " + "/oauth/... path itself. Including the path produces double-prefixed " + "URLs in the discovery doc." + ) + + return { + "OIDC_ENABLED": bool(private_key), + "OIDC_RSA_PRIVATE_KEY": private_key, + "OIDC_RSA_PRIVATE_KEYS_INACTIVE": inactive_keys, + "OIDC_ISS_ENDPOINT": iss_endpoint, + } + + OAUTH2_PROVIDER = { "SCOPES": { "profile:read": "Read access to user's profile (username, full name)", @@ -830,9 +881,21 @@ class StorageTypes: "annotations:types:write": "Create/Update/Delete annotation types", "annotations:annotations:read": "Read access to annotations", "annotations:annotations:write": "Create/Update/Delete annotations", - } + "openid": "Sign you in to PhysioNet", + "profile": "Your basic profile (name, username)", + "email": "Your primary email address", + }, + **load_oidc_provider_config(config), + # Implicit and hybrid response types leak tokens via URL fragments; both + # are deprecated by OAuth 2.1 / RFC 9700 §2.1.2. + "OIDC_RESPONSE_TYPES_SUPPORTED": ["code"], + "OAUTH2_VALIDATOR_CLASS": "oauth.validators.CustomOAuth2Validator", } +# DOT's Application model is swappable; declaring the default explicitly lets +# our oauth.Partner FK to it resolve cleanly during migrations. +OAUTH2_PROVIDER_APPLICATION_MODEL = "oauth2_provider.Application" + # Path to GeoIP2 database directory GEOIP_PATH = config('GEOIP_PATH', default=None) diff --git a/physionet-django/physionet/urls.py b/physionet-django/physionet/urls.py index 5695d3ddb5..282997a67c 100644 --- a/physionet-django/physionet/urls.py +++ b/physionet-django/physionet/urls.py @@ -7,6 +7,7 @@ from django.contrib import admin from django.http import HttpResponse from django.urls import path +from oauth.views import PhysioNetDiscoveryView from physionet import views from physionet.settings.base import StorageTypes @@ -15,6 +16,8 @@ handler500 = 'physionet.views.error_500' urlpatterns = [ + # OIDC discovery (must be at root per OIDC spec) + path('.well-known/openid-configuration', PhysioNetDiscoveryView.as_view(), name='oidc-root-discovery'), # django admin app path('admin/', admin.site.urls), # management console app @@ -111,4 +114,9 @@ 'lightwave_server_compat': { '_skip_': lambda: (shutil.which('sandboxed-lightwave') is None), }, + # OIDC discovery is gated by OIDCOnlyMixin (returns 404 unless an + # OIDC RSA key is configured). Skip when OIDC isn't enabled. + 'oidc-root-discovery': { + '_skip_': lambda: not settings.OAUTH2_PROVIDER.get('OIDC_ENABLED', False), + }, }