dnscms: log in to the CMS with Galtinn over OIDC

dnscms/tests/test_oidc.py

@@ -0,0 +1,377 @@
+import logging
+import re
+
+import pytest
+import requests
+from authlib.integrations.django_client import OAuthError
+from django.contrib.auth.models import Group, Permission, User
+from django.http import HttpResponseRedirect
+from django.urls import reverse
+from joserfc.errors import JoseError
+
+from dnscms import oidc
+from dnscms.models import GaltinnUser
+from dnscms.oidc import WAGTAIL_EDITORS_GROUP, LoginRefused, resolve_user
+
+pytestmark = pytest.mark.django_db
+
+EDITOR_GROUP = "Aktive"
+
+
+def claims(**overrides):
+    base = {
+        "sub": "42",
+        "preferred_username": "kari",
+        "email": "kari@neuf.test",
+        "given_name": "Kari",
+        "family_name": "Nordmann",
+        "groups": [EDITOR_GROUP, "Styret"],
+    }
+    return {**base, **overrides}
+
+
+@pytest.fixture(autouse=True)
+def oidc_settings(settings):
+    settings.GALTINN_EDITOR_GROUP = EDITOR_GROUP
+    settings.OIDC_RP_CLIENT_ID = "client-id"
+
+
+def link(user, galtinn_id="42"):
+    return GaltinnUser.objects.create(user=user, galtinn_id=galtinn_id)
+
+
+def editor(client, django_user_model, **kwargs):
+    user = django_user_model.objects.create_user(username="editor", password="pw", **kwargs)
+    user.user_permissions.add(Permission.objects.get(codename="access_admin"))
+    client.force_login(user)
+    return user
+
+
+def test_linked_member_is_synced_from_claims():
+    user = User.objects.create_user(username="stale-name", email="stale@example.test")
+    link(user)
+
+    resolved = resolve_user(claims(preferred_username="kari-new"))
+
+    assert resolved.pk == user.pk
+    assert resolved.username == "kari-new"
+    assert resolved.email == "kari@neuf.test"
+    assert resolved.first_name == "Kari"
+
+
+def test_linked_member_is_reactivated_and_added_to_editors():
+    user = User.objects.create_user(username="kari", is_active=False)
+    link(user)
+
+    resolved = resolve_user(claims())
+
+    assert resolved.is_active
+    assert resolved.groups.filter(name=WAGTAIL_EDITORS_GROUP).exists()
+
+
+def test_linked_member_loses_local_password():
+    user = User.objects.create_user(username="kari", password="pw")
+    link(user)
+
+    resolved = resolve_user(claims())
+
+    assert not resolved.has_usable_password()
+
+
+def test_integer_sub_matches_stored_id():
+    user = User.objects.create_user(username="kari")
+    link(user)
+
+    assert resolve_user(claims(sub=42)).pk == user.pk
+
+
+def test_linked_username_collision_refuses_login(caplog):
+    linked = User.objects.create_user(username="kari-old", email="old@example.test")
+    link(linked)
+    other = User.objects.create_user(username="kari", email="other@example.test")
+
+    with caplog.at_level(logging.ERROR), pytest.raises(LoginRefused):
+        resolve_user(claims())
+
+    linked.refresh_from_db()
+    other.refresh_from_db()
+    assert linked.username == "kari-old"
+    assert linked.email == "old@example.test"
+    assert other.email == "other@example.test"
+    assert "resolve manually" in caplog.text
+
+
+def test_linked_non_member_is_deactivated_and_loses_editors():
+    user = User.objects.create_user(username="kari-old", email="old@example.test")
+    user.groups.add(Group.objects.get(name=WAGTAIL_EDITORS_GROUP))
+    link(user)
+
+    with pytest.raises(LoginRefused):
+        resolve_user(claims(groups=["Styret"]))
+
+    user.refresh_from_db()
+    assert not user.is_active
+    assert not user.groups.exists()
+    assert user.username == "kari-old"
+    assert user.email == "old@example.test"
+
+
+def test_linked_non_member_superuser_stays_active():
+    user = User.objects.create_superuser(username="kari", email="old@example.test", password="pw")
+    link(user)
+
+    with pytest.raises(LoginRefused):
+        resolve_user(claims(groups=[]))
+
+    user.refresh_from_db()
+    assert user.is_active
+    assert user.email == "old@example.test"
+    assert user.has_usable_password()
+
+
+@pytest.mark.parametrize(
+    "local",
+    [
+        {"username": "kari", "email": "other@example.test"},
+        {"username": "KARI", "email": "other@example.test"},
+        {"username": "somebody", "email": "kari@neuf.test"},
+        {"username": "somebody", "email": "KARI@neuf.test"},
+    ],
+)
+@pytest.mark.parametrize("groups", [[EDITOR_GROUP], []])
+def test_unlinked_local_user_is_never_matched(local, groups, caplog):
+    user = User.objects.create_superuser(password="pw", **local)
+
+    with caplog.at_level(logging.ERROR), pytest.raises(LoginRefused):
+        resolve_user(claims(groups=groups))
+
+    user.refresh_from_db()
+    assert user.username == local["username"]
+    assert user.email == local["email"]
+    assert user.is_active
+    assert user.has_usable_password()
+    assert User.objects.count() == 1
+    assert not GaltinnUser.objects.exists()
+    assert ("link manually" in caplog.text) == bool(groups)
+
+
+def test_link_is_never_rebound_to_another_galtinn_id():
+    user = User.objects.create_user(username="kari", email="kari@neuf.test")
+    link(user, galtinn_id="7")
+
+    with pytest.raises(LoginRefused):
+        resolve_user(claims())
+
+    assert GaltinnUser.objects.get(user=user).galtinn_id == "7"
+
+
+def test_repeat_login_keeps_single_link():
+    resolve_user(claims())
+    resolve_user(claims())
+
+    assert GaltinnUser.objects.get(galtinn_id="42").user.username == "kari"
+    assert User.objects.count() == 1
+
+
+def test_unknown_member_is_created_with_galtinn_identity():
+    resolved = resolve_user(claims())
+
+    assert resolved.username == "kari"
+    assert resolved.email == "kari@neuf.test"
+    assert resolved.first_name == "Kari"
+    assert not resolved.has_usable_password()
+    assert resolved.groups.filter(name=WAGTAIL_EDITORS_GROUP).exists()
+    assert GaltinnUser.objects.filter(user=resolved, galtinn_id="42").exists()
+
+
+def test_unknown_non_member_is_not_created():
+    with pytest.raises(LoginRefused):
+        resolve_user(claims(groups=["Styret"]))
+
+    assert not User.objects.exists()
+
+
+@pytest.mark.parametrize("name", ["sub", "preferred_username", "email"])
+@pytest.mark.parametrize("value", [None, ""])
+def test_missing_claim_refuses_login(name, value):
+    bad = claims(**{name: value}) if value is not None else claims()
+    if value is None:
+        del bad[name]
+
+    with pytest.raises(LoginRefused):
+        resolve_user(bad)
+
+    assert not User.objects.exists()
+
+
+def test_empty_editor_group_setting_fails_closed(settings):
+    settings.GALTINN_EDITOR_GROUP = ""
+
+    with pytest.raises(LoginRefused):
+        resolve_user(claims())
+
+    assert not User.objects.exists()
+
+
+def test_editors_group_missing_fails_loudly():
+    Group.objects.filter(name=WAGTAIL_EDITORS_GROUP).delete()
+
+    with pytest.raises(Group.DoesNotExist):
+        resolve_user(claims())
+
+    assert not User.objects.exists()
+
+
+def test_login_redirects_to_galtinn_with_configured_callback(client, monkeypatch, settings):
+    settings.WAGTAIL_BASE_URL = "https://cms.example.test"
+    seen = {}
+
+    def authorize_redirect(request, redirect_uri):
+        seen["redirect_uri"] = redirect_uri
+        return HttpResponseRedirect("https://galtinn.example.test/authorize")
+
+    monkeypatch.setattr(oidc.oauth.galtinn, "authorize_redirect", authorize_redirect)
+
+    response = client.get(reverse("oidc_login"), {"next": "/admin/pages/3/"})
+
+    assert response.url == "https://galtinn.example.test/authorize"
+    assert seen["redirect_uri"] == "https://cms.example.test/oidc/callback/"
+    assert client.session["oidc_next"] == "/admin/pages/3/"
+
+
+@pytest.mark.parametrize("setting", ["OIDC_RP_CLIENT_ID", "GALTINN_EDITOR_GROUP"])
+def test_views_are_404_when_not_configured(client, settings, setting):
+    setattr(settings, setting, "")
+
+    assert client.get(reverse("oidc_login")).status_code == 404
+    assert client.get(reverse("oidc_callback")).status_code == 404
+
+
+@pytest.mark.parametrize(
+    ("next_url", "expected"),
+    [("/admin/pages/3/", "/admin/pages/3/"), ("https://evil.example.test/", "/admin/")],
+)
+def test_callback_only_follows_local_next(client, monkeypatch, next_url, expected):
+    monkeypatch.setattr(
+        oidc.oauth.galtinn, "authorize_access_token", lambda request: {"userinfo": claims()}
+    )
+    session = client.session
+    session["oidc_next"] = next_url
+    session.save()
+
+    response = client.get(reverse("oidc_callback"))
+
+    assert response.url == expected
+
+
+@pytest.mark.parametrize(
+    "error", [OAuthError("denied"), JoseError("bad token"), requests.ConnectionError("down")]
+)
+def test_callback_flow_failure_redirects_to_login_with_message(client, monkeypatch, error):
+    def fail(request):
+        raise error
+
+    monkeypatch.setattr(oidc.oauth.galtinn, "authorize_access_token", fail)
+
+    response = client.get(reverse("oidc_callback"), follow=True)
+
+    assert response.redirect_chain[-1][0] == reverse("wagtailadmin_login")
+    assert any("feilet" in str(m) for m in response.context["messages"])
+    assert "_auth_user_id" not in client.session
+
+
+def test_callback_without_userinfo_is_refused(client, monkeypatch):
+    monkeypatch.setattr(oidc.oauth.galtinn, "authorize_access_token", lambda request: {})
+
+    response = client.get(reverse("oidc_callback"), follow=True)
+
+    assert any("ikke tilgang" in str(m) for m in response.context["messages"])
+    assert "_auth_user_id" not in client.session
+
+
+def test_callback_logs_in_and_redirects_to_admin(client, monkeypatch):
+    monkeypatch.setattr(
+        oidc.oauth.galtinn, "authorize_access_token", lambda request: {"userinfo": claims()}
+    )
+
+    response = client.get(reverse("oidc_callback"))
+
+    assert response.status_code == 302
+    assert response.url == reverse("wagtailadmin_home")
+    user = User.objects.get(username="kari")
+    assert client.session["_auth_user_id"] == str(user.pk)
+
+
+def test_callback_refusal_redirects_to_login_with_message(client, monkeypatch):
+    monkeypatch.setattr(
+        oidc.oauth.galtinn,
+        "authorize_access_token",
+        lambda request: {"userinfo": claims(groups=[])},
+    )
+
+    response = client.get(reverse("oidc_callback"), follow=True)
+
+    assert response.redirect_chain[-1][0] == reverse("wagtailadmin_login")
+    messages = [str(m) for m in response.context["messages"]]
+    assert any("ikke tilgang" in m for m in messages)
+    assert "_auth_user_id" not in client.session
+
+
+def test_login_page_shows_galtinn_button_once(client):
+    response = client.get(reverse("wagtailadmin_login"))
+
+    assert response.content.count(reverse("oidc_login").encode()) == 1
+    assert b"Logg inn med lokal bruker" in response.content
+    assert b"login-form" in response.content
+
+
+def test_login_page_is_local_only_when_not_configured(client, settings):
+    settings.OIDC_RP_CLIENT_ID = ""
+
+    html = client.get(reverse("wagtailadmin_login")).content.decode()
+
+    assert reverse("oidc_login") not in html
+    assert "Logg inn med lokal bruker" not in html
+    assert re.search(r'<div class="local-login" id="local-login"\s*>', html)
+
+
+def test_linked_account_page_shows_galtinn_note_and_disables_fields(client, django_user_model):
+    link(editor(client, django_user_model))
+
+    html = client.get(reverse("wagtailadmin_account")).content.decode()
+
+    assert "Navn og e-post må endres i Galtinn." in html
+    for field in ["first_name", "last_name", "email"]:
+        tag = re.search(rf'<input[^>]*name="name_email-{field}"[^>]*>', html).group(0)
+        assert "disabled" in tag
+
+
+def test_linked_account_post_cannot_change_name_or_email(client, django_user_model):
+    user = editor(client, django_user_model, email="editor@neuf.test", first_name="Ed")
+    link(user)
+
+    client.post(
+        reverse("wagtailadmin_account"),
+        {"name_email-first_name": "Hacked", "name_email-email": "hacked@example.test"},
+    )
+
+    user.refresh_from_db()
+    assert user.first_name == "Ed"
+    assert user.email == "editor@neuf.test"
+
+
+def test_local_account_page_keeps_editable_name_and_email(client, django_user_model):
+    editor(client, django_user_model)
+
+    html = client.get(reverse("wagtailadmin_account")).content.decode()
+
+    assert "Navn og e-post må endres i Galtinn." not in html
+    tag = re.search(r'<input[^>]*name="name_email-email"[^>]*>', html).group(0)
+    assert "disabled" not in tag
+
+
+def test_account_page_requires_login(client):
+    response = client.get(reverse("wagtailadmin_account"))
+
+    assert response.status_code == 302
+    assert reverse("wagtailadmin_login") in response.url