announcer/tests/test_auth.py

"""Tester for autentiseringen.

Auth er den ene delen av denne appen man ikke kan se etter med øyet: at en PIN
virker ser du med en gang, at et utløpt token blir avvist ser du aldri før det
smeller. Derfor står prøvene her og ikke rundt resten.

Testene kjører mot en bar Flask-app, ikke mot announcer.py – det ville dratt inn
piper, lastet stemmer og startet avspillingstråden.
"""

from __future__ import annotations

import sys
import time
import urllib.parse
from pathlib import Path

import jwt as pyjwt
import pytest
from flask import Flask, jsonify

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import auth  # noqa: E402
import config  # noqa: E402

SECRET = "en-delt-hemmelighet-mellom-panelet-og-announcer"


def build(tmp_path: Path, toml: str = "", **env: str):
    """Lager en liten app med gitt konfigurasjon. Returnerer testklienten."""
    cfg_path = tmp_path / "config.toml"
    cfg_path.write_text(f'[paths]\nstate = "{tmp_path}"\nlog = "{tmp_path}/a.jsonl"\n\n{toml}')
    for key, value in env.items():
        __import__("os").environ[key] = value
    try:
        cfg = config.load(cfg_path)
    finally:
        for key in env:
            __import__("os").environ.pop(key, None)

    app = Flask(__name__)
    app.config["TESTING"] = True
    authenticator = auth.install(app, cfg)

    @app.get("/do/<action>")
    def do(action: str):  # noqa: ANN202
        return jsonify(auth.require(action).public())

    @app.errorhandler(401)
    def _401(e):  # noqa: ANN001, ANN202
        return jsonify({"error": e.description}), 401

    @app.errorhandler(403)
    def _403(e):  # noqa: ANN001, ANN202
        return jsonify({"error": e.description}), 403

    return app.test_client(), authenticator


def token(claims: dict, *, secret: str = SECRET, alg: str = "HS256", **over) -> str:
    body = {"sub": "u1", "exp": time.time() + 600, **claims, **over}
    return pyjwt.encode(body, secret, algorithm=alg)


# --------------------------------------------------------------------------
# Uten konfigurasjon skal ingenting være stengt
# --------------------------------------------------------------------------


def test_uten_konfig_er_alt_apent(tmp_path):
    c, a = build(tmp_path)
    assert a.enabled is False
    for action in config.ACTIONS:
        if action == "play":
            continue
        assert c.get(f"/do/{action}").status_code == 200, action
    # Det ene unntaket: «play» er for en sentral med mesh-nøkkel. En åpen
    # sone skal ikke spille lyd for hvem som helst som finner den.
    assert c.get("/do/play").status_code == 401


def test_feil_legitimasjon_er_feil_ogsa_pa_apen_sone(tmp_path):
    """En PIN som vises fram og er feil, skal avvises – ikke behandles som
    gjest fordi gjester tilfeldigvis har lov."""
    c, _ = build(tmp_path, '[mesh]\nsecret = "riktig"\n')
    assert c.get("/do/view").status_code == 200                                  # gjest: lov
    assert c.get("/do/view", headers={"X-Mesh-Key": "feil"}).status_code == 401  # feil nøkkel: nei
    assert c.get("/do/play", headers={"X-Mesh-Key": "riktig"}).status_code == 200


def test_tom_providers_liste_er_fortsatt_apen(tmp_path):
    c, a = build(tmp_path, "[auth]\nproviders = []\n")
    assert a.enabled is False
    assert c.get("/do/clear").status_code == 200


# --------------------------------------------------------------------------
# PIN
# --------------------------------------------------------------------------


def test_pin_kreves_nar_den_er_satt(tmp_path):
    c, a = build(tmp_path, '[auth.pin]\npin = "1234"\n')
    assert a.enabled is True
    assert [p.name for p in a.providers] == ["pin"]
    assert c.get("/do/announce").status_code == 401
    assert c.get("/do/announce", headers={"X-Pin": "feil"}).status_code == 401
    assert c.get("/do/announce", headers={"X-Pin": "1234"}).status_code == 200


def test_pin_fra_miljovariabel_slar_pa_leverandoren(tmp_path):
    """Bakoverkompatibilitet: dette er alt systemd-enheten har satt fram til nå."""
    c, a = build(tmp_path, ANNOUNCER_PIN="9999")
    assert [p.name for p in a.providers] == ["pin"]
    assert c.get("/do/clear", headers={"X-Pin": "9999"}).status_code == 200
    assert c.get("/do/clear").status_code == 401


def test_feil_pin_gir_forstaelig_melding(tmp_path):
    c, _ = build(tmp_path, '[auth.pin]\npin = "1234"\n')
    assert c.get("/do/announce", headers={"X-Pin": "nope"}).get_json()["error"] == "Feil PIN"


def test_pin_gir_admin_som_standard(tmp_path):
    c, _ = build(tmp_path, '[auth.pin]\npin = "1234"\n')
    body = c.get("/do/clear", headers={"X-Pin": "1234"}).get_json()
    assert body["roles"] == ["admin"]
    assert sorted(body["can"]) == sorted(config.ACTIONS)


def test_pin_med_crew_rolle_far_ikke_tomme_koen(tmp_path):
    c, _ = build(tmp_path, '[auth.pin]\npin = "1234"\nrole = "crew"\n')
    assert c.get("/do/announce", headers={"X-Pin": "1234"}).status_code == 200
    r = c.get("/do/clear", headers={"X-Pin": "1234"})
    assert r.status_code == 403
    assert "crew" in r.get_json()["error"]


def test_pin_godtas_ogsa_i_authorization_header(tmp_path):
    c, _ = build(tmp_path, '[auth.pin]\npin = "1234"\n')
    assert c.get("/do/announce", headers={"Authorization": "Pin 1234"}).status_code == 200


# --------------------------------------------------------------------------
# Anonym rolle
# --------------------------------------------------------------------------


def test_anonym_kan_gis_leserett(tmp_path):
    """For statusskjermen som henger på veggen uten noen til å logge inn."""
    c, _ = build(tmp_path, '[auth.pin]\npin = "1234"\n\n[roles]\nanonymous = ["view"]\n')
    assert c.get("/do/view").status_code == 200
    assert c.get("/do/announce").status_code == 401


# --------------------------------------------------------------------------
# JWT
# --------------------------------------------------------------------------


JWT_CFG = f'''
[auth.jwt]
secret = "{SECRET}"
audience = "announcer"
issuer = "https://crew.example.no"
'''


def test_gyldig_jwt_slipper_inn_med_rolle_og_navn(tmp_path):
    c, a = build(tmp_path, JWT_CFG)
    assert [p.name for p in a.providers] == ["jwt"]
    t = token({"aud": "announcer", "iss": "https://crew.example.no",
               "name": "Kari Nordmann", "roles": ["crew"], "sub": "42"})
    body = c.get("/do/announce", headers={"Authorization": f"Bearer {t}"}).get_json()
    assert body["name"] == "Kari Nordmann"
    assert body["roles"] == ["crew"]
    assert body["subject"] == "jwt:42"
    assert body["method"] == "jwt"


def test_utlopt_jwt_avvises(tmp_path):
    c, _ = build(tmp_path, JWT_CFG)
    t = token({"aud": "announcer", "iss": "https://crew.example.no", "roles": ["admin"]},
              exp=time.time() - 3600)
    r = c.get("/do/announce", headers={"Authorization": f"Bearer {t}"})
    assert r.status_code == 401
    assert "utløpt" in r.get_json()["error"]


def test_jwt_for_en_annen_tjeneste_avvises(tmp_path):
    c, _ = build(tmp_path, JWT_CFG)
    t = token({"aud": "et-annet-system", "iss": "https://crew.example.no", "roles": ["admin"]})
    r = c.get("/do/announce", headers={"Authorization": f"Bearer {t}"})
    assert r.status_code == 401
    assert "annen tjeneste" in r.get_json()["error"]


def test_jwt_fra_feil_utsteder_avvises(tmp_path):
    c, _ = build(tmp_path, JWT_CFG)
    t = token({"aud": "announcer", "iss": "https://ondsinnet.example", "roles": ["admin"]})
    assert c.get("/do/announce", headers={"Authorization": f"Bearer {t}"}).status_code == 401


def test_jwt_med_feil_signatur_avvises(tmp_path):
    c, _ = build(tmp_path, JWT_CFG)
    t = token({"aud": "announcer", "iss": "https://crew.example.no", "roles": ["admin"]},
              secret="en-helt-annen-hemmelighet-som-er-lang-nok-for-hmac")
    assert c.get("/do/announce", headers={"Authorization": f"Bearer {t}"}).status_code == 401


def test_jwt_uten_exp_avvises(tmp_path):
    """Et token uten utløpstid er en evig nøkkel; den vil vi ikke ha."""
    c, _ = build(tmp_path, JWT_CFG)
    body = {"sub": "u1", "aud": "announcer", "iss": "https://crew.example.no", "roles": ["admin"]}
    t = pyjwt.encode(body, SECRET, algorithm="HS256")
    assert c.get("/do/announce", headers={"Authorization": f"Bearer {t}"}).status_code == 401


def test_jwt_rolle_uten_rettighet_gir_403_ikke_401(tmp_path):
    """Forskjellen betyr noe for klienten: 401 = logg inn, 403 = du får ikke."""
    c, _ = build(tmp_path, JWT_CFG)
    t = token({"aud": "announcer", "iss": "https://crew.example.no", "roles": ["crew"]})
    h = {"Authorization": f"Bearer {t}"}
    assert c.get("/do/announce", headers=h).status_code == 200
    assert c.get("/do/clear", headers=h).status_code == 403


def test_ingen_authorization_header_gir_401(tmp_path):
    c, _ = build(tmp_path, JWT_CFG)
    assert c.get("/do/view").status_code == 401


@pytest.mark.parametrize(
    "claims, role_claim, forventet",
    [
        ({"roles": ["crew", "admin"]}, "roles", ["crew", "admin"]),
        ({"roles": "crew admin"}, "roles", ["crew", "admin"]),
        ({"roles": "crew,admin"}, "roles", ["crew", "admin"]),
        ({"roles": "crew"}, "roles", ["crew"]),
        ({"realm_access": {"roles": ["crew"]}}, "realm_access.roles", ["crew"]),
        ({"groups": ["admin"]}, "groups", ["admin"]),
        ({}, "roles", []),
    ],
)
def test_rolleclaim_tolkes_i_alle_formene(tmp_path, claims, role_claim, forventet):
    c, _ = build(tmp_path, JWT_CFG + f'role_claim = "{role_claim}"\n')
    t = token({"aud": "announcer", "iss": "https://crew.example.no", **claims})
    r = c.get("/do/view", headers={"Authorization": f"Bearer {t}"})
    if forventet:
        assert r.get_json()["roles"] == forventet
    else:
        # Gyldig token uten roller: vi *vet* hvem dette er, han får bare ikke
        # gjøre noe. Da er 403 riktig – å logge inn på nytt hjelper ikke.
        assert r.status_code == 403


def test_default_role_brukes_nar_tokenet_mangler_roller(tmp_path):
    c, _ = build(tmp_path, JWT_CFG + 'default_role = "crew"\n')
    t = token({"aud": "announcer", "iss": "https://crew.example.no"})
    body = c.get("/do/announce", headers={"Authorization": f"Bearer {t}"}).get_json()
    assert body["roles"] == ["crew"]


@pytest.mark.parametrize(
    "claims, forventet",
    [
        ({"name": "Kari"}, "Kari"),
        ({"preferred_username": "kari"}, "kari"),
        ({"email": "kari@example.no"}, "kari@example.no"),
        ({}, "u1"),
    ],
)
def test_visningsnavn_finnes_fram_fra_det_som_er(tmp_path, claims, forventet):
    c, _ = build(tmp_path, JWT_CFG + 'default_role = "admin"\n')
    t = token({"aud": "announcer", "iss": "https://crew.example.no", **claims})
    assert c.get("/do/view", headers={"Authorization": f"Bearer {t}"}).get_json()["name"] == forventet


# --------------------------------------------------------------------------
# Token-lenka fra crew-panelet
# --------------------------------------------------------------------------


def test_token_lenke_gir_sesjon_sa_resten_virker_uten_header(tmp_path):
    c, _ = build(tmp_path, JWT_CFG)
    t = token({"aud": "announcer", "iss": "https://crew.example.no",
               "name": "Kent", "roles": ["admin"]})
    r = c.get(f"/auth/token?t={t}")
    assert r.status_code == 302
    body = c.get("/do/clear").get_json()   # ingen header denne gangen
    assert body["name"] == "Kent"
    assert body["method"] == "jwt"


def test_token_lenke_med_ugyldig_token_gir_ingen_sesjon(tmp_path):
    c, _ = build(tmp_path, JWT_CFG)
    assert c.get("/auth/token?t=aldeles-ugyldig").status_code == 401
    assert c.get("/do/view").status_code == 401


def test_token_lenke_finnes_ikke_uten_jwt_oppsett(tmp_path):
    c, _ = build(tmp_path, '[auth.pin]\npin = "1234"\n')
    assert c.get("/auth/token?t=hva-som-helst").status_code == 404


def test_utlogging_tommer_sesjonen(tmp_path):
    c, _ = build(tmp_path, JWT_CFG)
    t = token({"aud": "announcer", "iss": "https://crew.example.no", "roles": ["admin"]})
    c.get(f"/auth/token?t={t}")
    assert c.get("/do/clear").status_code == 200
    c.get("/auth/logout")
    assert c.get("/do/clear").status_code == 401


def test_rettigheter_slas_opp_pa_nytt_for_hver_forespørsel(tmp_path):
    """Endrer man [roles] skal det slå inn uten at folk må logge inn på nytt."""
    c, a = build(tmp_path, JWT_CFG)
    t = token({"aud": "announcer", "iss": "https://crew.example.no", "roles": ["crew"]})
    c.get(f"/auth/token?t={t}")
    assert c.get("/do/clear").status_code == 403
    a.cfg.data["roles"]["crew"] = ["*"]
    assert c.get("/do/clear").status_code == 200


# --------------------------------------------------------------------------
# Flere metoder samtidig
# --------------------------------------------------------------------------


def test_pin_og_jwt_kan_virke_side_om_side(tmp_path):
    c, a = build(tmp_path, JWT_CFG + '\n[auth.pin]\npin = "1234"\nrole = "crew"\n')
    assert sorted(p.name for p in a.providers) == ["jwt", "pin"]
    assert c.get("/do/announce", headers={"X-Pin": "1234"}).status_code == 200
    t = token({"aud": "announcer", "iss": "https://crew.example.no", "roles": ["admin"]})
    assert c.get("/do/clear", headers={"Authorization": f"Bearer {t}"}).status_code == 200


def test_rekkefolgen_i_providers_bestemmer(tmp_path):
    c, a = build(tmp_path, '[auth]\nproviders = ["jwt", "pin"]\n' + JWT_CFG + '\n[auth.pin]\npin = "1"\n')
    assert [p.name for p in a.providers] == ["jwt", "pin"]


# --------------------------------------------------------------------------
# Enhetstester på småtingene
# --------------------------------------------------------------------------


@pytest.mark.parametrize(
    "value, forventet",
    [
        (["a", "b"], ("a", "b")),
        ("a b", ("a", "b")),
        ("a,b", ("a", "b")),
        ("a", ("a",)),
        (None, ()),
        (42, ()),
        ([" a ", ""], ("a",)),
    ],
)
def test_roller_tolkes(value, forventet):
    assert auth._roles(value) == forventet


def test_nestet_claim_hentes():
    claims = {"realm_access": {"roles": ["crew"]}}
    assert auth._claim(claims, "realm_access.roles") == ["crew"]
    assert auth._claim(claims, "finnes.ikke") is None
    assert auth._claim(claims, "realm_access.roles.dypere") is None


def test_rolle_til_handling():
    cfg = config.load("/finnes-ikke.toml")
    assert cfg.permissions(("admin",)) == frozenset(config.ACTIONS)
    assert cfg.permissions(("crew",)) == frozenset({"view", "preview", "announce", "schedule"})
    assert cfg.permissions(("ukjent-rolle",)) == frozenset()
    assert cfg.permissions(()) == frozenset()
    # Roller legges sammen
    cfg.data["roles"]["bare_stopp"] = ["stop"]
    assert cfg.permissions(("crew", "bare_stopp")) == frozenset(
        {"view", "preview", "announce", "schedule", "stop"})


def test_ukjente_handlinger_i_roles_ignoreres():
    cfg = config.load("/finnes-ikke.toml")
    cfg.data["roles"]["rar"] = ["announce", "slett-alt"]
    assert cfg.permissions(("rar",)) == frozenset({"announce"})


def test_sesjonsnokkel_er_stabil_mellom_omstarter(tmp_path):
    cfg = config.load("/finnes-ikke.toml")
    cfg.data["paths"]["state"] = str(tmp_path)
    first = auth.secret_key(cfg)
    assert auth.secret_key(cfg) == first
    assert (tmp_path / "secret_key").stat().st_mode & 0o777 == 0o600


# --------------------------------------------------------------------------
# OIDC
# --------------------------------------------------------------------------
# Crew-panelet har ikke OIDC-endepunkter ennå, så her står en stubbet
# identitetsleverandør i stedet. Det er HTTP-kallene som stubbes; discovery,
# PKCE, state-sjekken, kodebyttet og verifiseringen av id_token er vår egen
# kode og kjøres på ordentlig.

OIDC_CFG = '''
[auth.oidc]
issuer = "https://crew.example.no"
client_id = "announcer"
client_secret = "panel-hemmelighet-som-er-lang-nok-for-hmac-sha256"
'''

DISCOVERY = {
    "issuer": "https://crew.example.no",
    "authorization_endpoint": "https://crew.example.no/oauth/authorize",
    "token_endpoint": "https://crew.example.no/oauth/token",
    "jwks_uri": "",
}


@pytest.fixture
def idp(monkeypatch):
    """Stubber HTTP-laget og husker hva vi sendte til tokenendepunktet."""
    state: dict = {"posts": [], "claims": {}}

    def fake_get(url, timeout=10):
        assert url == "https://crew.example.no/.well-known/openid-configuration"
        return dict(DISCOVERY)

    def fake_post(url, form, headers=None, timeout=10):
        state["posts"].append({"url": url, "form": form, "headers": headers or {}})
        claims = {
            "iss": "https://crew.example.no",
            "aud": "announcer",
            "exp": time.time() + 600,
            "sub": "142",
            **state["claims"],
        }
        # Panelet signerer id_token med client_secret (HS256) i dette oppsettet.
        return {"id_token": pyjwt.encode(claims, "panel-hemmelighet-som-er-lang-nok-for-hmac-sha256", algorithm="HS256")}

    monkeypatch.setattr(auth, "_get_json", fake_get)
    monkeypatch.setattr(auth, "_post_json", fake_post)
    return state


def test_oidc_login_sender_brukeren_til_panelet_med_pkce(tmp_path, idp):
    c, _ = build(tmp_path, OIDC_CFG)
    r = c.get("/auth/login")
    assert r.status_code == 302
    url = urllib.parse.urlparse(r.headers["Location"])
    q = urllib.parse.parse_qs(url.query)
    assert url.netloc == "crew.example.no" and url.path == "/oauth/authorize"
    assert q["response_type"] == ["code"]
    assert q["client_id"] == ["announcer"]
    assert q["code_challenge_method"] == ["S256"]
    assert q["redirect_uri"][0].endswith("/auth/callback")
    assert len(q["code_challenge"][0]) == 43  # sha256, base64url uten padding
    assert q["state"][0]


def test_oidc_hele_runden_gir_innlogget_sesjon(tmp_path, idp):
    idp["claims"] = {"name": "Kari Nordmann", "roles": ["crew"]}
    c, _ = build(tmp_path, OIDC_CFG)
    q = urllib.parse.parse_qs(urllib.parse.urlparse(c.get("/auth/login").headers["Location"]).query)
    r = c.get(f"/auth/callback?code=en-kode&state={q['state'][0]}")
    assert r.status_code == 302 and r.headers["Location"] == "/"

    body = c.get("/do/announce").get_json()
    assert body["name"] == "Kari Nordmann"
    assert body["roles"] == ["crew"]
    assert body["subject"] == "oidc:142"
    assert body["method"] == "oidc"
    assert c.get("/do/clear").status_code == 403

    # Kodebyttet: PKCE-verifier med, hemmeligheten som Basic-auth, ikke i body.
    post = idp["posts"][0]
    assert post["url"] == "https://crew.example.no/oauth/token"
    assert post["form"]["grant_type"] == "authorization_code"
    assert post["form"]["code"] == "en-kode"
    assert len(post["form"]["code_verifier"]) > 40
    assert "client_secret" not in post["form"]
    assert post["headers"]["Authorization"].startswith("Basic ")


def test_oidc_callback_med_feil_state_avvises(tmp_path, idp):
    """Uten denne sjekken kan en angriper få deg til å logge inn som seg selv."""
    c, _ = build(tmp_path, OIDC_CFG)
    c.get("/auth/login")
    r = c.get("/auth/callback?code=en-kode&state=tilfeldig-tull")
    assert r.status_code == 400
    assert c.get("/do/view").status_code == 401


def test_oidc_callback_uten_forutgaende_login_avvises(tmp_path, idp):
    c, _ = build(tmp_path, OIDC_CFG)
    assert c.get("/auth/callback?code=x&state=y").status_code == 400


def test_oidc_feil_fra_panelet_vises(tmp_path, idp):
    c, _ = build(tmp_path, OIDC_CFG)
    c.get("/auth/login")
    r = c.get("/auth/callback?error=access_denied&error_description=Avvist+av+bruker")
    assert r.status_code == 401
    assert "access_denied" in r.get_json()["error"]


def test_oidc_id_token_for_en_annen_klient_avvises(tmp_path, idp, monkeypatch):
    c, _ = build(tmp_path, OIDC_CFG)
    monkeypatch.setattr(auth, "_post_json", lambda url, form, headers=None, timeout=10: {
        "id_token": pyjwt.encode(
            {"iss": "https://crew.example.no", "aud": "et-annet-system",
             "exp": time.time() + 600, "sub": "1"},
            "panel-hemmelighet-som-er-lang-nok-for-hmac-sha256", algorithm="HS256")})
    q = urllib.parse.parse_qs(urllib.parse.urlparse(c.get("/auth/login").headers["Location"]).query)
    assert c.get(f"/auth/callback?code=k&state={q['state'][0]}").status_code == 401


def test_oidc_state_kan_ikke_brukes_to_ganger(tmp_path, idp):
    c, _ = build(tmp_path, OIDC_CFG)
    q = urllib.parse.parse_qs(urllib.parse.urlparse(c.get("/auth/login").headers["Location"]).query)
    state = q["state"][0]
    assert c.get(f"/auth/callback?code=k&state={state}").status_code == 302
    c.get("/auth/logout")
    assert c.get(f"/auth/callback?code=k&state={state}").status_code == 400


def test_oidc_manuelle_endepunkter_hopper_over_discovery(tmp_path, monkeypatch):
    """Panelet har ikke .well-known ennå, så dette er veien dit først."""
    def nei(*a, **kw):
        raise AssertionError("skulle ikke slått opp discovery")
    monkeypatch.setattr(auth, "_get_json", nei)
    c, _ = build(tmp_path, OIDC_CFG + '''
authorization_endpoint = "https://crew.example.no/oauth/authorize"
token_endpoint = "https://crew.example.no/oauth/token"
''')
    assert c.get("/auth/login").status_code == 302


def test_oidc_login_finnes_ikke_uten_oppsett(tmp_path):
    c, _ = build(tmp_path, '[auth.pin]\npin = "1"\n')
    assert c.get("/auth/login").status_code == 404
    assert c.get("/auth/callback?code=x&state=y").status_code == 404


def test_public_url_styrer_redirect_uri(tmp_path, idp):
    """Bak en omvendt proxy må redirect_uri peke utover, ikke på localhost."""
    c, _ = build(tmp_path, '[server]\npublic_url = "https://utrop.example.no"\n' + OIDC_CFG)
    q = urllib.parse.parse_qs(urllib.parse.urlparse(c.get("/auth/login").headers["Location"]).query)
    assert q["redirect_uri"] == ["https://utrop.example.no/auth/callback"]