Pluggbar autentisering: PIN, JWT og OIDC
Dagens «auth» var én delt PIN i et X-Pin-header, sammenlignet med !=, og
den sa ingenting om hvem som sendte et utrop – køen viste bare IP-adressen.
Crew har et eget adminpanel som skal bli identitetsleverandør.
Modellen er én datatype og én funksjon: Principal (hvem + roller +
rettigheter) og require(action), som enten gir deg den eller stopper
forespørselen. Handlinger: view, preview, announce, stop, clear.
Tre leverandører, prøvd i rekkefølge, alle valgfrie:
- pin X-Pin som før, men med hmac.compare_digest istedenfor ==. Rolle
admin som standard, så eksisterende oppsett oppfører seg likt.
Beholdt bevisst: den virker uten nett, og det er nettopp da man
trenger å rope ut noe.
- jwt Authorization: Bearer. HS256 med delt hemmelighet, eller
RS256/ES256 mot panelets JWKS. Krever exp – et token uten
utløpstid er en evig nøkkel. Roller fra et konfigurerbart claim,
som tåler liste, streng, kommaliste og punktsti.
- oidc Autorisasjonskode med PKCE. Discovery hentes først ved første
innlogging, så tjenesten starter selv om panelet er nede.
En leverandør er aktiv når den står i auth.providers *eller* når den er
ferdig konfigurert. Derfor slår ANNOUNCER_PIN=1234 alene fortsatt på PIN,
slik det har virket til nå. Er ingen aktiv, er alt åpent – som før.
/auth/token?t=<jwt> er broa mens panelet mangler OIDC: panelet lenker hit
med et kortlevd token, vi setter sesjonscookie. Ett klikk, ingen PIN.
Kontrakten panelet skal oppfylle står i docs/crew-integration.md.
401 betyr «vi vet ikke hvem du er, logg inn», 403 «vi vet hvem du er, men
rollen din får ikke dette». Et gyldig token uten roller gir 403, ikke 401 –
å logge inn på nytt ville ikke hjulpet.
Samtidig:
- /api/status er lukket bak «view». Den var åpen, og lekket både
meldingstekstene og IP-ene til alle som hadde sendt noe, til hvem som
helst som nådde porten. IP vises nå bare for den som har «clear»; alle
andre ser navnet.
- Køen og en revisjonslogg i logs/announcements.jsonl viser hvem som sendte
hva. Loggen feiler aldri forespørselen – en logg som ikke lar seg skrive
skal ikke hindre et utrop.
- Branding flyttet fra index.html til /api/config, så config.toml fra
forrige commit faktisk brukes.
- Sesjonsnøkkel i instance/secret_key, ellers logges alle ut ved hver
restart. SESSION_COOKIE_SECURE settes bare bak TLS: appen kjører på rå
HTTP over LAN og NetBird, og med Secure på ville ingen klart å logge inn.
57 tester, de fleste på auth, inkludert hele OIDC-runden mot en stubbet
leverandør. Det er delen 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.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
87066bb8c2
commit
d6eb811ee5
10 changed files with 2090 additions and 117 deletions
535
tests/test_auth.py
Normal file
535
tests/test_auth.py
Normal file
|
|
@ -0,0 +1,535 @@
|
|||
"""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:
|
||||
assert c.get(f"/do/{action}").status_code == 200, action
|
||||
|
||||
|
||||
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"})
|
||||
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", "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"]
|
||||
Loading…
Add table
Add a link
Reference in a new issue