neuf-www/dnscms/tests/test_oidc.py

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