Erstanmeldung fragt nach, statt still ein zweites Konto anzulegen
Wer sich zum ersten Mal ueber den Anmeldedienst anmeldet und hier noch kein Konto hat, bekam bisher wortlos ein neues - mit Rolle "Benutzer" und ohne Verlauf. Wer schon ein Konto hatte, wollte aber fast immer genau das verknuepfen und stand dann mit zwei Konten da. Jetzt zeigt der Rueckweg eine Entscheidungsseite: bestehendes Konto verknuepfen oder ein neues anlegen. Das Verknuepfen verlangt einmalig Benutzername und Passwort des bestehenden Kontos. Genau dieser Nachweis ist der Grund, warum der Anmeldeweg selbst nie fragen darf, welches Konto gemeint ist - ohne ihn koennte sich jeder mit einem beliebigen Konto beim Anmeldedienst an ein fremdes Vorrania-Konto haengen. Die geprueften Angaben liegen dafuer kurzlebig und einmalig in oidc_pending_links; vom Code steht dort nur der SHA-256-Hash, wie bei den API-Tokens. Ein falsch getipptes Passwort verbraucht den Vorgang nicht, und unbekannter Benutzername und falsches Passwort antworten gleich - sonst liessen sich vorhandene Konten abzaehlen. AUTO_CREATE=false blendet den Knopf zum Anlegen aus und laesst sich auch ueber den Dialog nicht umgehen. Der App-Weg bleibt unveraendert, weil er keinen Zwischenschritt anzeigen kann. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
a134b2aad5
commit
eb29143f97
@@ -174,6 +174,31 @@ class OidcLoginState(Base):
|
||||
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
|
||||
|
||||
|
||||
class OidcPendingLink(Base):
|
||||
"""Eine gepruefte Anmeldedienst-Kennung, die noch keinem Konto gehoert.
|
||||
|
||||
Entsteht, wenn sich jemand erfolgreich beim Anmeldedienst ausgewiesen hat,
|
||||
aber weder eine bekannte Kennung noch eine passende bestaetigte E-Mail
|
||||
vorliegt. Statt still ein zweites Konto anzulegen, fragt Vorrania nach:
|
||||
neues Konto oder bestehendes verknuepfen.
|
||||
|
||||
Kurzlebig und einmalig verwendbar. Gespeichert werden nur die geprueften
|
||||
Angaben aus dem ID-Token – nie das Token selbst.
|
||||
"""
|
||||
|
||||
__tablename__ = "oidc_pending_links"
|
||||
|
||||
# Wie bei ApiToken liegt nur der SHA-256-Hash hier; den Klartext kennt allein
|
||||
# der Browser, der ihn im Adressfragment mitbekommen hat.
|
||||
code_hash: Mapped[str] = mapped_column(String(64), primary_key=True)
|
||||
provider: Mapped[str] = mapped_column(String(64), nullable=False)
|
||||
subject: Mapped[str] = mapped_column(String(255), nullable=False)
|
||||
email: Mapped[str | None] = mapped_column(String(255))
|
||||
email_verified: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
|
||||
username_vorschlag: Mapped[str] = mapped_column(String(64), nullable=False)
|
||||
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
|
||||
|
||||
|
||||
class UserIdentity(Base):
|
||||
"""Verknuepfung eines Vorrania-Kontos mit einem Konto beim Anmeldeanbieter.
|
||||
|
||||
|
||||
@@ -23,12 +23,14 @@ from sqlalchemy.orm import Session
|
||||
|
||||
from ..database import get_db
|
||||
from ..deps import get_current_user
|
||||
from ..models import User, UserIdentity
|
||||
from ..models import OidcPendingLink, User, UserIdentity
|
||||
from ..oidc_config import OidcProvider, provider_finden, registrierte_provider
|
||||
from ..schemas import (
|
||||
OidcIdentityOut,
|
||||
OidcLinkStart,
|
||||
OidcNativeLogin,
|
||||
OidcPendingOut,
|
||||
OidcPendingVerknuepfen,
|
||||
OidcProviderOut,
|
||||
Token,
|
||||
)
|
||||
@@ -48,6 +50,8 @@ WEB_FEHLER = "/login"
|
||||
# Eine Verknuepfung endet dort, wo sie angestossen wurde – nicht auf der
|
||||
# Anmeldemaske: Der Nutzer ist ja die ganze Zeit angemeldet geblieben.
|
||||
WEB_KONTO = "/konto"
|
||||
# Wohin es geht, wenn die Kennung echt ist, aber noch keinem Konto gehoert.
|
||||
WEB_ENTSCHEIDUNG = "/sso-neu"
|
||||
|
||||
|
||||
def _provider_oder_404(slug: str) -> OidcProvider:
|
||||
@@ -189,7 +193,18 @@ def anmeldung_abschliessen(
|
||||
status_code=status.HTTP_303_SEE_OTHER,
|
||||
)
|
||||
|
||||
nutzer = dienst.benutzer_aus_claims(db, provider, claims)
|
||||
nutzer = dienst.konto_suchen(db, provider, claims)
|
||||
if nutzer is None:
|
||||
# Die Kennung ist echt, gehoert aber noch niemandem. Statt still ein
|
||||
# zweites Konto anzulegen, nachfragen – wer schon ein Konto hat, will
|
||||
# meistens genau das verknuepfen.
|
||||
dienst.offen_aufraeumen(db)
|
||||
merkzettel = dienst.offen_anlegen(db, provider, claims)
|
||||
logger.info("SSO '%s': unbekannte Kennung, frage nach.", slug)
|
||||
return RedirectResponse(
|
||||
f"{WEB_ENTSCHEIDUNG}#code={quote(merkzettel)}",
|
||||
status_code=status.HTTP_303_SEE_OTHER,
|
||||
)
|
||||
except dienst.OidcFehler as fehler:
|
||||
logger.warning("SSO '%s' gescheitert: %s", slug, fehler)
|
||||
return _fehler_weiterleitung(fehler.schluessel, verknuepfung=verknuepfen_fuer is not None)
|
||||
@@ -335,3 +350,71 @@ def verknuepfung_loesen(
|
||||
# "letzte Anmeldemoeglichkeit" ist kein Serverfehler, sondern eine
|
||||
# bewusste Weigerung – der Text erklaert den Ausweg.
|
||||
raise HTTPException(status.HTTP_409_CONFLICT, str(fehler)) from fehler
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Erstanmeldung: Kennung ist echt, gehoert aber noch keinem Konto
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _offen_oder_404(db: Session, code: str) -> OidcPendingLink:
|
||||
eintrag = dienst.offen_lesen(db, code)
|
||||
if eintrag is None:
|
||||
raise HTTPException(
|
||||
status.HTTP_404_NOT_FOUND,
|
||||
"Der Anmeldevorgang ist abgelaufen. Bitte noch einmal anmelden.",
|
||||
)
|
||||
return eintrag
|
||||
|
||||
|
||||
@router.get("/pending/{code}", response_model=OidcPendingOut)
|
||||
def offene_anmeldung(code: str, db: Session = Depends(get_db)) -> OidcPendingOut:
|
||||
"""Was die Entscheidungsseite anzeigen soll.
|
||||
|
||||
Ohne Anmeldung abrufbar – der Einmalcode aus der Adresse ist der Nachweis.
|
||||
"""
|
||||
eintrag = _offen_oder_404(db, code)
|
||||
provider = _provider_oder_404(eintrag.provider)
|
||||
return OidcPendingOut(
|
||||
provider=eintrag.provider,
|
||||
display_name=provider.display_name,
|
||||
email=eintrag.email,
|
||||
username_vorschlag=eintrag.username_vorschlag,
|
||||
anlegen_erlaubt=provider.auto_create,
|
||||
)
|
||||
|
||||
|
||||
@router.post("/pending/{code}/anlegen", response_model=Token)
|
||||
def offene_anmeldung_neues_konto(code: str, db: Session = Depends(get_db)) -> Token:
|
||||
"""Legt ein neues Vorrania-Konto zu dieser Kennung an."""
|
||||
eintrag = _offen_oder_404(db, code)
|
||||
provider = _provider_oder_404(eintrag.provider)
|
||||
try:
|
||||
nutzer = dienst.offen_neues_konto(db, provider, eintrag)
|
||||
except dienst.OidcFehler as fehler:
|
||||
raise HTTPException(status.HTTP_403_FORBIDDEN, str(fehler)) from fehler
|
||||
logger.info("SSO '%s': neues Konto '%s' auf Wunsch angelegt.", provider.slug, nutzer.username)
|
||||
return _token_fuer(nutzer)
|
||||
|
||||
|
||||
@router.post("/pending/{code}/verknuepfen", response_model=Token)
|
||||
def offene_anmeldung_verknuepfen(
|
||||
code: str, payload: OidcPendingVerknuepfen, db: Session = Depends(get_db)
|
||||
) -> Token:
|
||||
"""Haengt die Kennung an ein bestehendes Konto – nach Passwort-Nachweis."""
|
||||
eintrag = _offen_oder_404(db, code)
|
||||
provider = _provider_oder_404(eintrag.provider)
|
||||
try:
|
||||
nutzer = dienst.offen_verknuepfen(
|
||||
db, provider, eintrag, payload.username, payload.password
|
||||
)
|
||||
except dienst.OidcFehler as fehler:
|
||||
# Falsches Passwort ist 401, eine belegte Kennung ein Konflikt.
|
||||
code_http = (
|
||||
status.HTTP_401_UNAUTHORIZED
|
||||
if fehler.schluessel == "anmeldung_falsch"
|
||||
else status.HTTP_409_CONFLICT
|
||||
)
|
||||
raise HTTPException(code_http, str(fehler)) from fehler
|
||||
logger.info("SSO '%s': mit bestehendem Konto '%s' verknuepft.", provider.slug, nutzer.username)
|
||||
return _token_fuer(nutzer)
|
||||
|
||||
@@ -154,6 +154,29 @@ class OidcLinkStart(BaseModel):
|
||||
authorization_url: str
|
||||
|
||||
|
||||
class OidcPendingOut(BaseModel):
|
||||
"""Eine gepruefte Kennung, die noch keinem Konto gehoert.
|
||||
|
||||
Wird der Entscheidungsseite gezeigt: „neues Konto" oder „bestehendes
|
||||
verknuepfen". Der Einmalcode aus der Adresse ist der Nachweis, dass die
|
||||
Anmeldung beim Anbieter tatsaechlich stattgefunden hat.
|
||||
"""
|
||||
|
||||
provider: str
|
||||
display_name: str
|
||||
email: str | None = None
|
||||
username_vorschlag: str
|
||||
# Ob dieser Server das Anlegen neuer Konten ueberhaupt zulaesst.
|
||||
anlegen_erlaubt: bool
|
||||
|
||||
|
||||
class OidcPendingVerknuepfen(BaseModel):
|
||||
"""Eigentumsnachweis fuer das bestehende Konto."""
|
||||
|
||||
username: str = Field(min_length=1, max_length=64)
|
||||
password: str = Field(min_length=1, max_length=255)
|
||||
|
||||
|
||||
class OidcNativeLogin(BaseModel):
|
||||
"""ID-Token aus dem App-Flow, das gegen ein Vorrania-Token getauscht wird."""
|
||||
|
||||
|
||||
+157
-15
@@ -15,8 +15,9 @@ from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
import secrets
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
@@ -27,8 +28,9 @@ from authlib.oidc.core import CodeIDToken
|
||||
from sqlalchemy import func
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from ..models import Role, User, UserIdentity
|
||||
from ..models import OidcPendingLink, Role, User, UserIdentity
|
||||
from ..oidc_config import OidcProvider
|
||||
from ..security import hash_api_token, verify_password
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -38,6 +40,8 @@ TIMEOUT_SECONDS = 10.0
|
||||
CACHE_SEKUNDEN = 3600.0
|
||||
# Etwas Spielraum fuer auseinanderlaufende Uhren zwischen Server und Anbieter.
|
||||
UHREN_SPIELRAUM = 60
|
||||
# So lange darf die Entscheidung „neues Konto oder verknuepfen" offenbleiben.
|
||||
OFFEN_GUELTIG = timedelta(minutes=10)
|
||||
|
||||
|
||||
class OidcFehler(Exception):
|
||||
@@ -345,16 +349,17 @@ def identitaet_loesen(db: Session, nutzer: User, provider_slug: str) -> None:
|
||||
logger.info("SSO '%s': Verknuepfung von '%s' geloest.", provider_slug, nutzer.username)
|
||||
|
||||
|
||||
def benutzer_aus_claims(
|
||||
def konto_suchen(
|
||||
db: Session, provider: OidcProvider, claims: dict[str, Any]
|
||||
) -> User:
|
||||
"""Findet oder erzeugt das Vorrania-Konto zu einem geprueften ID-Token.
|
||||
|
||||
Reihenfolge:
|
||||
) -> User | None:
|
||||
"""Sucht das Konto zu einem geprueften ID-Token – ohne je eines anzulegen.
|
||||
|
||||
1. Bekannte Identitaet (Anbieter + ``sub``) – der Normalfall ab dem zweiten Mal.
|
||||
2. Erstverknuepfung ueber die **bestaetigte** E-Mail-Adresse.
|
||||
3. Neues Konto, falls fuer den Anbieter erlaubt.
|
||||
|
||||
``None`` heisst: die Kennung ist echt, gehoert aber noch niemandem. Der
|
||||
Browser-Weg fragt dann nach (neues Konto oder bestehendes verknuepfen),
|
||||
statt still ein zweites Konto anzulegen.
|
||||
"""
|
||||
subject = str(claims.get("sub") or "").strip()
|
||||
if not subject:
|
||||
@@ -428,13 +433,15 @@ def benutzer_aus_claims(
|
||||
email,
|
||||
)
|
||||
|
||||
# 3) Neu anlegen, falls erlaubt.
|
||||
if not provider.auto_create:
|
||||
raise OidcFehler(
|
||||
"kein_konto",
|
||||
"Zu dieser Anmeldung gibt es kein Vorrania-Konto. Ein Administrator muss "
|
||||
"es zuerst anlegen und die E-Mail-Adresse hinterlegen.",
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def konto_anlegen(db: Session, provider: OidcProvider, claims: dict[str, Any]) -> User:
|
||||
"""Legt ein neues Konto zu einer geprueften, noch freien Kennung an."""
|
||||
subject = str(claims.get("sub") or "").strip()
|
||||
if not subject:
|
||||
raise OidcFehler("kein_subject", "ID-Token ohne 'sub'.")
|
||||
email = _email_aus_claims(claims)
|
||||
|
||||
# Die Adresse ist eindeutig indiziert. Sie kann hier trotzdem schon vergeben
|
||||
# sein – naemlich wenn Schritt 2 sie wegen fehlender Bestaetigung NICHT zum
|
||||
@@ -472,3 +479,138 @@ def benutzer_aus_claims(
|
||||
db.refresh(nutzer)
|
||||
logger.info("SSO '%s': neues Konto '%s' angelegt.", provider.slug, nutzer.username)
|
||||
return nutzer
|
||||
|
||||
|
||||
def benutzer_aus_claims(
|
||||
db: Session, provider: OidcProvider, claims: dict[str, Any]
|
||||
) -> User:
|
||||
"""Sucht das Konto und legt es notfalls an – der Weg ohne Rueckfrage.
|
||||
|
||||
Genutzt vom App-Anmeldeweg, der keinen Zwischenschritt anzeigen kann.
|
||||
"""
|
||||
nutzer = konto_suchen(db, provider, claims)
|
||||
if nutzer is not None:
|
||||
return nutzer
|
||||
if not provider.auto_create:
|
||||
raise OidcFehler(
|
||||
"kein_konto",
|
||||
"Zu dieser Anmeldung gibt es kein Vorrania-Konto. Melde dich einmal im "
|
||||
"Browser an, um es zu verknuepfen, oder lass dir eines anlegen.",
|
||||
)
|
||||
return konto_anlegen(db, provider, claims)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Offene Erstanmeldung: Kennung ist echt, gehoert aber noch keinem Konto
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _mit_zeitzone(wert: datetime) -> datetime:
|
||||
# SQLite gibt Zeitstempel ohne Zonenangabe zurueck.
|
||||
return wert if wert.tzinfo is not None else wert.replace(tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def offen_anlegen(db: Session, provider: OidcProvider, claims: dict[str, Any]) -> str:
|
||||
"""Merkt die geprueften Angaben vor und gibt den Einmalcode im Klartext zurueck.
|
||||
|
||||
Gespeichert wird nur, was aus dem bereits geprueften ID-Token stammt – nie
|
||||
das Token selbst. Vom Code liegt hier nur der Hash, wie bei den API-Tokens.
|
||||
"""
|
||||
subject = str(claims.get("sub") or "").strip()
|
||||
if not subject:
|
||||
raise OidcFehler("kein_subject", "ID-Token ohne 'sub'.")
|
||||
|
||||
roh_bestaetigt = claims.get("email_verified")
|
||||
code = secrets.token_urlsafe(32)
|
||||
db.add(
|
||||
OidcPendingLink(
|
||||
code_hash=hash_api_token(code),
|
||||
provider=provider.slug,
|
||||
subject=subject,
|
||||
email=_email_aus_claims(claims),
|
||||
email_verified=roh_bestaetigt is True or str(roh_bestaetigt).lower() == "true",
|
||||
username_vorschlag=_namensvorschlag(claims),
|
||||
created_at=_jetzt(),
|
||||
)
|
||||
)
|
||||
db.commit()
|
||||
return code
|
||||
|
||||
|
||||
def offen_lesen(
|
||||
db: Session, code: str, jetzt: datetime | None = None
|
||||
) -> OidcPendingLink | None:
|
||||
"""Liest den Merkzettel, ohne ihn zu verbrauchen (fuer die Anzeige)."""
|
||||
eintrag = db.get(OidcPendingLink, hash_api_token(code))
|
||||
if eintrag is None:
|
||||
return None
|
||||
if _mit_zeitzone(eintrag.created_at) + OFFEN_GUELTIG < (jetzt or _jetzt()):
|
||||
db.delete(eintrag)
|
||||
db.commit()
|
||||
return None
|
||||
return eintrag
|
||||
|
||||
|
||||
def offen_aufraeumen(db: Session, jetzt: datetime | None = None) -> int:
|
||||
grenze = (jetzt or _jetzt()) - OFFEN_GUELTIG
|
||||
alte = [
|
||||
e for e in db.query(OidcPendingLink).all() if _mit_zeitzone(e.created_at) < grenze
|
||||
]
|
||||
for eintrag in alte:
|
||||
db.delete(eintrag)
|
||||
if alte:
|
||||
db.commit()
|
||||
return len(alte)
|
||||
|
||||
|
||||
def _claims_aus_offen(eintrag: OidcPendingLink) -> dict[str, Any]:
|
||||
return {
|
||||
"sub": eintrag.subject,
|
||||
"email": eintrag.email,
|
||||
"email_verified": eintrag.email_verified,
|
||||
"preferred_username": eintrag.username_vorschlag,
|
||||
}
|
||||
|
||||
|
||||
def offen_neues_konto(
|
||||
db: Session, provider: OidcProvider, eintrag: OidcPendingLink
|
||||
) -> User:
|
||||
"""Legt das neue Konto an und verbraucht den Merkzettel."""
|
||||
if not provider.auto_create:
|
||||
raise OidcFehler(
|
||||
"anlegen_nicht_erlaubt",
|
||||
"Auf diesem Server duerfen sich keine neuen Konten selbst anlegen.",
|
||||
)
|
||||
claims = _claims_aus_offen(eintrag)
|
||||
db.delete(eintrag)
|
||||
db.flush()
|
||||
return konto_anlegen(db, provider, claims)
|
||||
|
||||
|
||||
def offen_verknuepfen(
|
||||
db: Session,
|
||||
provider: OidcProvider,
|
||||
eintrag: OidcPendingLink,
|
||||
username: str,
|
||||
passwort: str,
|
||||
) -> User:
|
||||
"""Haengt die Kennung an ein bestehendes Konto – nach Passwort-Nachweis.
|
||||
|
||||
Das Passwort ist hier der Eigentumsnachweis. Ohne ihn koennte sich jeder mit
|
||||
einem beliebigen Konto beim Anmeldedienst an ein fremdes Vorrania-Konto
|
||||
haengen – genau deshalb fragt der Anmeldeweg selbst nie „welches Konto?".
|
||||
"""
|
||||
nutzer = db.query(User).filter(User.username == username).first()
|
||||
if nutzer is None or not verify_password(passwort, nutzer.password_hash):
|
||||
# Bewusst dieselbe Meldung fuer „gibt es nicht" und „Passwort falsch":
|
||||
# sonst verraet die Antwort, welche Benutzernamen existieren.
|
||||
raise OidcFehler("anmeldung_falsch", "Benutzername oder Passwort falsch.")
|
||||
|
||||
claims = _claims_aus_offen(eintrag)
|
||||
identitaet_verknuepfen(db, nutzer, provider, claims)
|
||||
# Erst nach erfolgreicher Verknuepfung verbrauchen – ein Tippfehler im
|
||||
# Passwort soll nicht den ganzen Anmeldevorgang wegwerfen.
|
||||
db.delete(eintrag)
|
||||
db.commit()
|
||||
db.refresh(nutzer)
|
||||
return nutzer
|
||||
|
||||
+108
-1
@@ -11,16 +11,22 @@ from types import SimpleNamespace
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
|
||||
from app.models import Role, User, UserIdentity
|
||||
from app.models import OidcPendingLink, Role, User, UserIdentity
|
||||
from app.oidc_config import OidcProvider, provider_aus_umgebung
|
||||
from app.routers.auth import login
|
||||
from app.services import oidc_state
|
||||
from app.security import hash_password, verify_password
|
||||
from app.services.oidc import (
|
||||
OFFEN_GUELTIG,
|
||||
OidcFehler,
|
||||
benutzer_aus_claims,
|
||||
identitaet_loesen,
|
||||
identitaet_verknuepfen,
|
||||
konto_suchen,
|
||||
offen_anlegen,
|
||||
offen_lesen,
|
||||
offen_neues_konto,
|
||||
offen_verknuepfen,
|
||||
)
|
||||
|
||||
BASIS = "https://vorrania.example.com"
|
||||
@@ -431,3 +437,104 @@ def test_app_client_darf_eigene_discovery_adresse_haben():
|
||||
def test_ohne_eigene_adresse_teilen_sich_beide_wege_die_discovery():
|
||||
(p,) = provider_aus_umgebung(VOLLE_UMGEBUNG, BASIS)
|
||||
assert p.app_discovery_url == p.discovery_url
|
||||
|
||||
|
||||
# ---- Erstanmeldung ohne passendes Konto ------------------------------------
|
||||
|
||||
|
||||
def test_unbekannte_kennung_legt_nichts_an(db, provider):
|
||||
"""konto_suchen darf NIE ein Konto erzeugen – sonst gibt es nichts zu fragen."""
|
||||
assert konto_suchen(db, provider, claims()) is None
|
||||
assert db.query(User).count() == 0
|
||||
assert db.query(UserIdentity).count() == 0
|
||||
|
||||
|
||||
def test_offener_merkzettel_haelt_die_geprueften_angaben(db, provider):
|
||||
code = offen_anlegen(db, provider, claims(preferred_username="eva.mueller"))
|
||||
|
||||
eintrag = offen_lesen(db, code)
|
||||
assert eintrag.subject == "abc-123"
|
||||
assert eintrag.email == "eva@example.com"
|
||||
assert eintrag.email_verified is True
|
||||
assert eintrag.username_vorschlag == "eva.mueller"
|
||||
# Der Klartext-Code darf nirgends liegen, nur sein Hash.
|
||||
assert code not in (eintrag.code_hash, eintrag.subject)
|
||||
|
||||
|
||||
def test_offener_merkzettel_laeuft_ab(db, provider):
|
||||
code = offen_anlegen(db, provider, claims())
|
||||
spaeter = datetime.now(timezone.utc) + OFFEN_GUELTIG + timedelta(seconds=1)
|
||||
|
||||
assert offen_lesen(db, code, jetzt=spaeter) is None
|
||||
|
||||
|
||||
def test_falscher_code_findet_nichts(db, provider):
|
||||
offen_anlegen(db, provider, claims())
|
||||
assert offen_lesen(db, "ausgedacht") is None
|
||||
|
||||
|
||||
def test_verknuepfen_braucht_das_richtige_passwort(db, provider):
|
||||
"""Das Passwort ist der Eigentumsnachweis – ohne ihn waere der Dialog eine Luecke."""
|
||||
eva = User(username="eva", password_hash=hash_password("geheim123"), role=Role.admin)
|
||||
db.add(eva)
|
||||
db.commit()
|
||||
eintrag = offen_lesen(db, offen_anlegen(db, provider, claims()))
|
||||
|
||||
with pytest.raises(OidcFehler) as fehler:
|
||||
offen_verknuepfen(db, provider, eintrag, "eva", "falsch")
|
||||
|
||||
assert fehler.value.schluessel == "anmeldung_falsch"
|
||||
assert db.query(UserIdentity).count() == 0
|
||||
# Der Merkzettel bleibt: ein Tippfehler soll nicht den Vorgang wegwerfen.
|
||||
assert db.query(OidcPendingLink).count() == 1
|
||||
|
||||
|
||||
def test_unbekannter_benutzername_verraet_sich_nicht(db, provider):
|
||||
"""Gleiche Meldung wie bei falschem Passwort – sonst zaehlt man Konten ab."""
|
||||
eintrag = offen_lesen(db, offen_anlegen(db, provider, claims()))
|
||||
|
||||
with pytest.raises(OidcFehler) as fehler:
|
||||
offen_verknuepfen(db, provider, eintrag, "gibtsnicht", "beliebig")
|
||||
|
||||
assert fehler.value.schluessel == "anmeldung_falsch"
|
||||
|
||||
|
||||
def test_verknuepfen_haengt_kennung_an_das_bestaetigte_konto(db, provider):
|
||||
eva = User(username="eva", password_hash=hash_password("geheim123"), role=Role.admin)
|
||||
db.add(eva)
|
||||
db.commit()
|
||||
eintrag = offen_lesen(db, offen_anlegen(db, provider, claims()))
|
||||
|
||||
nutzer = offen_verknuepfen(db, provider, eintrag, "eva", "geheim123")
|
||||
|
||||
assert nutzer.id == eva.id
|
||||
assert nutzer.role == Role.admin # Rolle bleibt erhalten
|
||||
assert db.query(User).count() == 1 # kein Zweitkonto
|
||||
assert db.query(OidcPendingLink).count() == 0 # verbraucht
|
||||
# Ab jetzt meldet dieselbe Kennung direkt an.
|
||||
assert konto_suchen(db, provider, claims()).id == eva.id
|
||||
|
||||
|
||||
def test_neues_konto_auf_wunsch(db, provider):
|
||||
eintrag = offen_lesen(db, offen_anlegen(db, provider, claims(preferred_username="eva")))
|
||||
|
||||
neu = offen_neues_konto(db, provider, eintrag)
|
||||
|
||||
assert neu.username == "eva"
|
||||
assert neu.role == Role.user
|
||||
assert neu.password_hash is None
|
||||
assert db.query(OidcPendingLink).count() == 0
|
||||
|
||||
|
||||
def test_ohne_auto_anlage_bleibt_nur_verknuepfen(db, provider):
|
||||
"""AUTO_CREATE=false darf sich nicht ueber den Dialog umgehen lassen."""
|
||||
from dataclasses import replace
|
||||
|
||||
ohne = replace(provider, auto_create=False)
|
||||
eintrag = offen_lesen(db, offen_anlegen(db, ohne, claims()))
|
||||
|
||||
with pytest.raises(OidcFehler) as fehler:
|
||||
offen_neues_konto(db, ohne, eintrag)
|
||||
|
||||
assert fehler.value.schluessel == "anlegen_nicht_erlaubt"
|
||||
assert db.query(User).count() == 0
|
||||
|
||||
Reference in New Issue
Block a user