kirche-einfach/backend/tests/test_auth.py

"""Phase 1 tests: magic-link auth, access requests, RBAC, cookie flags."""

import logging
from datetime import timedelta
from uuid import uuid4

import pytest
from sqlmodel import select

from app.auth import service
from app.auth.deps import SESSION_COOKIE
from app.models import MagicLinkToken, User
from app.settings import settings

ADMIN = "admin@example.com"
MEMBER = "member@example.com"


@pytest.fixture(autouse=True)
def _no_smtp_many_admins(monkeypatch):
    # Never touch the network: dev fallback logs instead of mailing.
    # (The real repo .env configures SMTP; tests must not use it.)
    monkeypatch.setattr(settings, "smtp_host", "")
    monkeypatch.setattr(settings, "admin_emails", ADMIN)


def _get_or_make_user(session, email, role="member"):
    user = service.get_user_by_email(session, email)
    if user is None:
        user = User(email=email, name="Test", role=role)
        session.add(user)
        session.commit()
        session.refresh(user)
    return user


async def _login(client, session, email, role="member"):
    """Consume a fresh magic token; persist the session cookie manually.

    httpx honours the Secure cookie flag and would withhold it over the
    http://test transport, so the session id is re-set as a plain cookie.
    """
    user = _get_or_make_user(session, email, role)
    token = service.issue_magic_token(session, user)
    response = await client.post("/api/auth/magic/consume", json={"token": token})
    assert response.status_code == 200, response.text
    sid = response.cookies[SESSION_COOKIE]
    client.cookies.set(SESSION_COOKIE, sid)
    return response.json()


async def test_request_approve_consume_me_logout(client, session):
    assert (
        await client.post("/api/auth/admin-bootstrap", json={"email": ADMIN})
    ).status_code == 201
    admin = await _login(client, session, ADMIN, role="admin")
    assert admin["role"] == "admin"

    created = await client.post(
        "/api/access/requests",
        json={"name": "Maria", "email": MEMBER, "reason": "choir planning"},
    )
    assert created.status_code == 201, created.text
    assert created.json()["status"] == "pending"

    queue = await client.get("/api/access/requests")
    assert queue.status_code == 200
    assert any(r["email"] == MEMBER for r in queue.json())

    approved = await client.post(f"/api/access/requests/{created.json()['id']}/approve")
    assert approved.status_code == 200, approved.text
    body = approved.json()
    assert body["request"]["status"] == "approved"
    assert body["email_sent"] is False  # no SMTP in tests

    # New client (no admin cookie) logs in as the approved member.
    from httpx import ASGITransport, AsyncClient

    from app.main import app

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as member:
        user = service.get_user_by_email(session, MEMBER)
        assert user is not None and user.role == "member"
        token = service.issue_magic_token(session, user)
        consumed = await member.post("/api/auth/magic/consume", json={"token": token})
        assert consumed.status_code == 200
        assert consumed.json()["email"] == MEMBER
        member.cookies.set(SESSION_COOKIE, consumed.cookies[SESSION_COOKIE])

        me = await member.get("/api/auth/me")
        assert me.status_code == 200
        assert me.json()["tenant_id"] == "default"

        assert (await member.post("/api/auth/logout")).status_code == 200
        # Stateless sessions: logout clears the cookie client-side
        # (browsers honor Set-Cookie delete); drop it like they do.
        member.cookies.clear()
        assert (await member.get("/api/auth/me")).status_code == 401


async def test_consume_unknown_token_400(client):
    response = await client.post("/api/auth/magic/consume", json={"token": "nope"})
    assert response.status_code == 400


async def test_token_expiry_boundary(client, session):
    user = _get_or_make_user(session, MEMBER)
    token = service.issue_magic_token(session, user)
    record = session.exec(
        select(MagicLinkToken).where(MagicLinkToken.token_hash == service.hash_token(token))
    ).one()
    record.expires_at = service.utcnow() - timedelta(seconds=1)
    session.add(record)
    session.commit()

    response = await client.post("/api/auth/magic/consume", json={"token": token})
    assert response.status_code == 400
    assert "expired" in response.json()["detail"]


async def test_token_reuse_400(client, session):
    user = _get_or_make_user(session, MEMBER)
    token = service.issue_magic_token(session, user)
    assert (await client.post("/api/auth/magic/consume", json={"token": token})).status_code == 200
    again = await client.post("/api/auth/magic/consume", json={"token": token})
    assert again.status_code == 400
    assert "used" in again.json()["detail"]


async def test_admin_only_endpoints(client, session, anon_client):
    await _login(client, session, MEMBER, role="member")
    created = await client.post(
        "/api/access/requests",
        json={"name": "X", "email": "x@example.com", "reason": ""},
    )
    assert created.status_code == 201

    assert (await client.get("/api/access/requests")).status_code == 403
    approve = await client.post(f"/api/access/requests/{created.json()['id']}/approve")
    assert approve.status_code == 403

    async for anon in anon_client():
        assert (await anon.get("/api/access/requests")).status_code == 401
        assert (
            await anon.post(f"/api/access/requests/{created.json()['id']}/approve")
        ).status_code == 401
        assert (await anon.get("/api/auth/me")).status_code == 401


async def test_approve_unknown_and_double_decide(client, session):
    await client.post("/api/auth/admin-bootstrap", json={"email": ADMIN})
    await _login(client, session, ADMIN, role="admin")

    assert (await client.post(f"/api/access/requests/{uuid4()}/approve")).status_code == 404

    created = await client.post(
        "/api/access/requests",
        json={"name": "Y", "email": "y@example.com", "reason": ""},
    )
    rid = created.json()["id"]
    assert (await client.post(f"/api/access/requests/{rid}/reject")).status_code == 200
    again = await client.post(f"/api/access/requests/{rid}/approve")
    assert again.status_code == 409


async def test_bootstrap_once_and_forbidden(client):
    assert (
        await client.post("/api/auth/admin-bootstrap", json={"email": "stranger@example.com"})
    ).status_code == 403
    assert (
        await client.post("/api/auth/admin-bootstrap", json={"email": ADMIN})
    ).status_code == 201
    assert (
        await client.post("/api/auth/admin-bootstrap", json={"email": ADMIN})
    ).status_code == 409


async def test_session_cookie_flags(client, session):
    user = _get_or_make_user(session, MEMBER)
    token = service.issue_magic_token(session, user)
    response = await client.post("/api/auth/magic/consume", json={"token": token})
    assert response.status_code == 200
    cookies = response.headers.get_list("set-cookie")
    assert any("httponly" in c.lower() for c in cookies), cookies
    assert any("secure" in c.lower() for c in cookies), cookies
    assert any("samesite=lax" in c.lower() for c in cookies), cookies
    assert any("path=/" in c.lower() for c in cookies), cookies


async def test_no_token_or_session_in_logs(client, session, caplog):
    user = _get_or_make_user(session, MEMBER)
    token = service.issue_magic_token(session, user)
    with caplog.at_level(logging.INFO):
        caplog.clear()
        response = await client.post("/api/auth/magic/consume", json={"token": token})
    assert response.status_code == 200
    assert token not in caplog.text
    assert response.cookies[SESSION_COOKIE] not in caplog.text
    assert service.hash_token(token) not in caplog.text


def test_session_jwt_short_expiry_and_claims(session):
    """Stateless sessions expire in days (no revocation — bound the window)."""
    user = _get_or_make_user(session, MEMBER)
    claims = service.decode_session_token(service.mint_session_token(user))
    assert claims["sub"] == str(user.id)
    assert claims["tenant_id"] == user.tenant_id
    assert claims["exp"] - claims["iat"] == 7 * 24 * 3600


async def test_deleted_user_token_fails_closed(client, session):
    """Tokens of deleted users stop working (row reloaded per request)."""
    user = _get_or_make_user(session, MEMBER)
    token = service.issue_magic_token(session, user)
    response = await client.post("/api/auth/magic/consume", json={"token": token})
    assert response.status_code == 200, response.text
    jwt_cookie = response.cookies[SESSION_COOKIE]
    session.delete(user)
    session.commit()
    client.cookies.set(SESSION_COOKIE, jwt_cookie)
    assert (await client.get("/api/auth/me")).status_code == 401


async def test_access_request_rate_limit(client):
    email = f"impatient-{uuid4()}@example.com"
    for _ in range(service.REQUESTS_PER_EMAIL_PER_HOUR):
        response = await client.post(
            "/api/access/requests",
            json={"name": "I", "email": email, "reason": ""},
        )
        assert response.status_code == 201, response.text
    limited = await client.post(
        "/api/access/requests", json={"name": "I", "email": email, "reason": ""}
    )
    assert limited.status_code == 429


def _broken_smtp(monkeypatch):
    from app.auth import mail

    def _explode(message):
        raise OSError("relay down")

    monkeypatch.setattr(mail, "_send", _explode)


def test_magic_link_delivery_failure_returns_false(monkeypatch):
    from app.auth import mail

    monkeypatch.setattr(settings, "smtp_host", "dummy")
    _broken_smtp(monkeypatch)
    assert mail.send_magic_link("a@example.com", "http://x/login?token=t") is False
    assert mail.send_access_notification("req-1", "A", "a@example.com") is False


async def test_approve_survives_delivery_failure(client, session, monkeypatch):
    monkeypatch.setattr(settings, "smtp_host", "dummy")
    _broken_smtp(monkeypatch)
    await client.post("/api/auth/admin-bootstrap", json={"email": ADMIN})
    await _login(client, session, ADMIN, role="admin")
    created = await client.post(
        "/api/access/requests",
        json={"name": "M", "email": "m@example.com", "reason": ""},
    )
    assert created.status_code == 201, created.text
    approved = await client.post(f"/api/access/requests/{created.json()['id']}/approve")
    assert approved.status_code == 200, approved.text
    body = approved.json()
    assert body["request"]["status"] == "approved"
    assert body["email_sent"] is False


async def test_magic_request_for_known_user(client, session):
    _get_or_make_user(session, MEMBER)
    response = await client.post("/api/auth/magic/request", json={"email": MEMBER})
    assert response.status_code == 200, response.text
    body = response.json()
    assert body["message"] == "login link sent"
    assert body["email_sent"] is False  # no SMTP in tests
    assert "token" not in response.text
    tokens = session.exec(
        select(MagicLinkToken).where(
            MagicLinkToken.user_id == service.get_user_by_email(session, MEMBER).id
        )
    ).all()
    assert len(tokens) == 1


async def test_magic_request_unknown_email_403(client):
    response = await client.post("/api/auth/magic/request", json={"email": "stranger@example.com"})
    assert response.status_code == 403


async def test_magic_request_admin_email_creates_admin(client, session):
    response = await client.post("/api/auth/magic/request", json={"email": ADMIN})
    assert response.status_code == 200, response.text
    user = service.get_user_by_email(session, ADMIN)
    assert user is not None and user.role == "admin"


async def test_magic_request_rate_limit(client, session):
    _get_or_make_user(session, MEMBER)
    for _ in range(service.MAGIC_LINKS_PER_EMAIL_PER_HOUR):
        response = await client.post("/api/auth/magic/request", json={"email": MEMBER})
        assert response.status_code == 200, response.text
    limited = await client.post("/api/auth/magic/request", json={"email": MEMBER})
    assert limited.status_code == 429


def test_magic_link_mail_has_clickable_branded_html(monkeypatch):
    from app.auth import mail

    monkeypatch.setattr(settings, "smtp_host", "dummy")
    message = mail._branded_mail(
        subject="Dein Login-Link",
        to_email="a@example.com",
        headline="Anmeldung bei Kirche Einfach",
        intro="Klicke auf den Button.",
        button_label="Jetzt anmelden",
        button_url="http://localhost:3000/login?token=GEHEIM",
        note="Gültig 15 Minuten.",
    )
    html_part = next(p for p in message.walk() if p.get_content_type() == "text/html")
    html = html_part.get_content()
    # Button = anchor with href around the label; token never in text-only form.
    assert "http://localhost:3000/login?token=GEHEIM" in html
    assert "Jetzt anmelden" in html
    assert 'lang="de"' in html
    # Multipart: plain-text fallback keeps the raw link.
    parts = [p.get_content_type() for p in message.walk()]
    assert "text/plain" in parts and "text/html" in parts
    plain = next(p.get_content() for p in message.walk() if p.get_content_type() == "text/plain")
    assert "http://localhost:3000/login?token=GEHEIM" in plain