denya-onecare/tests/test_p0_auth_admin_batch.py

"""P0 security batch — admin user management, unified role model, login rate
limiting, WhatsApp webhook secret, mock-log auth, security headers, pagination.

Each item in the P0 hardening batch has an executable behavioral test here
(plus the register-removal tests living in test_p0_hardening.py).
"""

from __future__ import annotations

import pytest
from httpx import AsyncClient

from app.core.config import settings
from app.core.database import async_session_factory
from app.core.security import hash_password
from app.models.user import User
from app.services.seed import normalize_legacy_user_emails, normalize_legacy_user_roles

pytestmark = pytest.mark.asyncio


# ── helpers ───────────────────────────────────────────────────────────
async def _login(client, email="wahab@denya.com", password="denya123") -> str:
    resp = await client.post("/api/auth/login", json={"email": email, "password": password})
    assert resp.status_code == 200, resp.text
    return resp.json()["access_token"]


def _auth(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


async def _insert_user(email: str, role: str, *, active: bool = True) -> int:
    """Insert a user row directly (bypasses API role validation) — used to
    simulate legacy bootstrap/registration rows in the DB."""
    async with async_session_factory() as session:
        user = User(
            email=email,
            password_hash=hash_password("denya123"),
            full_name=f"Legacy {email}",
            role=role,
            active=active,
        )
        session.add(user)
        await session.commit()
        return user.id


async def _normalize_roles() -> None:
    async with async_session_factory() as session:
        await normalize_legacy_user_roles(session)
        await session.commit()


async def _normalize_emails() -> None:
    async with async_session_factory() as session:
        await normalize_legacy_user_emails(session)
        await session.commit()


async def _create_ticket(client, token: str, **overrides) -> dict:
    payload = {
        "unit_id": 2,
        "category_id": 3,
        "priority": "medium",
        "reporter": "P0 Batch Test",
        "reported_via": "walk-in",
        "description": "p0 batch ticket",
        **overrides,
    }
    resp = await client.post("/api/tickets", json=payload, headers=_auth(token))
    assert resp.status_code == 201, resp.text
    return resp.json()


# ── Item 2/3: admin user management ───────────────────────────────────
async def test_create_user_requires_admin(client: AsyncClient):
    token = await _login(client, email="bella@denya.com")  # CS Rep
    resp = await client.post(
        "/api/auth/users",
        json={"email": "x@example.com", "password": "password1", "full_name": "X", "role": "CS Rep"},
        headers=_auth(token),
    )
    assert resp.status_code == 403


async def test_create_user_requires_auth(client: AsyncClient):
    resp = await client.post(
        "/api/auth/users",
        json={"email": "x@example.com", "password": "password1", "full_name": "X", "role": "CS Rep"},
    )
    assert resp.status_code == 401


async def test_admin_creates_user_with_forced_role(client: AsyncClient):
    token = await _login(client)  # Admin/Wahab
    resp = await client.post(
        "/api/auth/users",
        json={"email": "New.Tech@Example.com", "password": "password1", "full_name": "New Tech", "role": "Tech"},
        headers=_auth(token),
    )
    assert resp.status_code == 201, resp.text
    created = resp.json()
    assert created["role"] == "Tech"
    assert created["active"] is True
    assert created["email"] == "new.tech@example.com"  # normalized lower-case

    # The new user can actually log in with their forced role.
    login = await client.post(
        "/api/auth/login", json={"email": "new.tech@example.com", "password": "password1"}
    )
    assert login.status_code == 200
    me = await client.get("/api/auth/me", headers=_auth(login.json()["access_token"]))
    assert me.json()["role"] == "Tech"


async def test_admin_create_user_rejects_unknown_roles(client: AsyncClient):
    """Unified role model: junk/legacy roles cannot be minted at creation."""
    token = await _login(client)
    for role in ("admin", "superadmin", "technician", "root"):
        resp = await client.post(
            "/api/auth/users",
            json={"email": f"{role.strip().lower()}@example.com", "password": "password1", "full_name": "X", "role": role},
            headers=_auth(token),
        )
        assert resp.status_code == 422, (role, resp.text)


async def test_admin_create_user_duplicate_email_conflict(client: AsyncClient):
    token = await _login(client)
    payload = {"email": "dupe2@example.com", "password": "password1", "full_name": "D", "role": "CS Rep"}
    assert (await client.post("/api/auth/users", json=payload, headers=_auth(token))).status_code == 201
    resp = await client.post("/api/auth/users", json=payload, headers=_auth(token))
    assert resp.status_code == 409


async def test_patch_user_role_change(client: AsyncClient):
    token = await _login(client)
    users = (await client.get("/api/auth/users", headers=_auth(token))).json()
    tech = next(u for u in users if u["role"] == "Tech")
    resp = await client.patch(
        f"/api/auth/users/{tech['id']}",
        json={"role": "CS Rep"},
        headers=_auth(token),
    )
    assert resp.status_code == 200, resp.text
    assert resp.json()["role"] == "CS Rep"
    assert resp.json()["active"] is True


async def test_patch_user_rejects_unknown_role(client: AsyncClient):
    token = await _login(client)
    users = (await client.get("/api/auth/users", headers=_auth(token))).json()
    tech = next(u for u in users if u["role"] == "Tech")
    resp = await client.patch(
        f"/api/auth/users/{tech['id']}",
        json={"role": "superadmin"},
        headers=_auth(token),
    )
    assert resp.status_code == 422, resp.text


async def test_patch_deactivate_blocks_login(client: AsyncClient):
    token = await _login(client)
    users = (await client.get("/api/auth/users", headers=_auth(token))).json()
    tech = next(u for u in users if u["role"] == "Tech")
    resp = await client.patch(
        f"/api/auth/users/{tech['id']}",
        json={"active": False},
        headers=_auth(token),
    )
    assert resp.status_code == 200
    assert resp.json()["active"] is False

    login = await client.post(
        "/api/auth/login", json={"email": tech["email"], "password": "denya123"}
    )
    assert login.status_code == 401


async def test_admin_cannot_modify_own_account(client: AsyncClient):
    token = await _login(client)  # wahab
    me = (await client.get("/api/auth/me", headers=_auth(token))).json()
    resp = await client.patch(
        f"/api/auth/users/{me['id']}", json={"active": False}, headers=_auth(token)
    )
    assert resp.status_code == 400


async def test_admin_can_demote_other_admin_but_not_self(client: AsyncClient):
    """An admin can manage the other admin seat, but self-removal stays blocked,
    so at least one canonical admin always remains (structural invariant)."""
    token = await _login(client)  # wahab
    users = (await client.get("/api/auth/users", headers=_auth(token))).json()
    jerome = next(u for u in users if u["role"] == "Admin/Jerome")
    wahab = next(u for u in users if u["role"] == "Admin/Wahab")

    # Demote the OTHER admin → allowed, wahab is still the acting admin.
    resp = await client.patch(
        f"/api/auth/users/{jerome['id']}", json={"role": "CEO"}, headers=_auth(token)
    )
    assert resp.status_code == 200, resp.text

    # Demoting/deactivating yourself is always rejected.
    resp = await client.patch(
        f"/api/auth/users/{wahab['id']}", json={"active": False}, headers=_auth(token)
    )
    assert resp.status_code == 400


async def test_delete_user_admin_only_and_works(client: AsyncClient):
    admin_token = await _login(client)
    users = (await client.get("/api/auth/users", headers=_auth(admin_token))).json()
    tech = next(u for u in users if u["role"] == "Tech")

    # Non-admin cannot delete.
    cs_token = await _login(client, email="bella@denya.com")
    resp = await client.delete(f"/api/auth/users/{tech['id']}", headers=_auth(cs_token))
    assert resp.status_code == 403

    # Admin deletes → 204, user gone, login fails.
    resp = await client.delete(f"/api/auth/users/{tech['id']}", headers=_auth(admin_token))
    assert resp.status_code == 204
    login = await client.post(
        "/api/auth/login", json={"email": tech["email"], "password": "denya123"}
    )
    assert login.status_code == 401


async def test_delete_user_referenced_by_ticket_is_409(client: AsyncClient):
    token = await _login(client)
    users = (await client.get("/api/auth/users", headers=_auth(token))).json()
    tech = next(u for u in users if u["role"] == "Tech")
    ticket = await _create_ticket(client, token)
    resp = await client.patch(
        f"/api/tickets/{ticket['id']}",
        json={"assigned_to": tech["id"]},
        headers=_auth(token),
    )
    assert resp.status_code == 200, resp.text

    resp = await client.delete(f"/api/auth/users/{tech['id']}", headers=_auth(token))
    assert resp.status_code == 409
    assert "related" in resp.json()["detail"].lower()


async def test_admin_cannot_delete_self(client: AsyncClient):
    token = await _login(client)
    me = (await client.get("/api/auth/me", headers=_auth(token))).json()
    resp = await client.delete(f"/api/auth/users/{me['id']}", headers=_auth(token))
    assert resp.status_code == 400


# ── Item 3: unified role model — legacy rows & JWT validation ─────────
async def test_legacy_alias_role_normalized_at_startup(client: AsyncClient):
    """A legacy 'technician' row is mapped onto canonical 'Tech' at startup."""
    await _insert_user("legacy-tech@example.com", "technician")
    await _normalize_roles()  # what lifespan does each boot

    login = await client.post(
        "/api/auth/login", json={"email": "legacy-tech@example.com", "password": "denya123"}
    )
    assert login.status_code == 200, login.text
    me = await client.get("/api/auth/me", headers=_auth(login.json()["access_token"]))
    assert me.json()["role"] == "Tech"


async def test_login_rejects_unknown_legacy_role_fail_closed(client: AsyncClient):
    """Lowercase 'superadmin' rows (mintable by the old open register) cannot
    log in — fail closed, never granted admin powers."""
    await _insert_user("legacy-admin@example.com", "superadmin")
    # NOTE: no normalize call — the row is exactly what the live DB holds today.

    login = await client.post(
        "/api/auth/login", json={"email": "legacy-admin@example.com", "password": "denya123"}
    )
    assert login.status_code == 401
    assert "role" in login.json()["detail"].lower()


async def test_jwt_validation_rejects_unknown_role(client: AsyncClient):
    """A token for a user whose row later becomes junk-role must fail closed."""
    token = await _login(client)  # wahab is a canonical admin at token time
    me = (await client.get("/api/auth/me", headers=_auth(token))).json()

    # Simulate a legacy DB row flip to a non-canonical role.
    async with async_session_factory() as session:
        from sqlalchemy import select
        user = (await session.execute(select(User).where(User.id == me["id"]))).scalar_one()
        user.role = "admin"
        await session.commit()

    resp = await client.get("/api/auth/me", headers=_auth(token))
    assert resp.status_code == 401


async def test_normalize_does_not_map_ambiguous_admin_alias(client: AsyncClient):
    """normalize_legacy_user_roles leaves identity-ambiguous 'admin' rows for
    operator remediation instead of guessing a canonical admin."""
    await _insert_user("legacy-admin2@example.com", "admin")
    await _normalize_roles()

    async with async_session_factory() as session:
        from sqlalchemy import select
        user = (
            await session.execute(select(User).where(User.email == "legacy-admin2@example.com"))
        ).scalar_one()
    assert user.role == "admin"


async def test_underscore_legacy_aliases_normalize_to_canonical():
    """Open-register rows can carry underscore role forms ('cs_rep',
    'cs_manager', 'fm_dispatcher') — the exact gap Mumuni relay #747 found on
    user 18 (test@denya.com).  Each must map onto its canonical role so startup
    normalization can converge the row instead of stranding it on /tickets."""
    from app.core.roles import normalize_role

    assert normalize_role("cs_rep") == "CS Rep"
    assert normalize_role("cs_manager") == "CS Manager"
    assert normalize_role("fm_dispatcher") == "FM Dispatcher"
    # Matching is case/whitespace tolerant, like the space-form aliases.
    assert normalize_role(" CS_REP ") == "CS Rep"
    assert normalize_role("cs_rep") is not None  # known, not fail-closed


async def test_legacy_cs_rep_row_converges_and_authenticates(client: AsyncClient):
    """A 'cs_rep' row (user 18 test@denya.com shape) is converged to canonical
    'CS Rep' by the startup self-heal and can log in again (no fail-closed
    denial, and the frontend CS nav sees the canonical role)."""
    await _insert_user("cs-rep-legacy@example.com", "cs_rep")
    await _normalize_roles()  # what lifespan does each boot

    async with async_session_factory() as session:
        from sqlalchemy import select
        user = (
            await session.execute(select(User).where(User.email == "cs-rep-legacy@example.com"))
        ).scalar_one()
    assert user.role == "CS Rep"

    login = await client.post(
        "/api/auth/login", json={"email": "cs-rep-legacy@example.com", "password": "denya123"}
    )
    assert login.status_code == 200, login.text
    me = await client.get("/api/auth/me", headers=_auth(login.json()["access_token"]))
    assert me.json()["role"] == "CS Rep"


# ── P0 email normalization (legacy mixed-case rows) ───────────────────
async def test_legacy_mixed_case_email_migrated_and_authenticates(client: AsyncClient):
    """A legacy row whose email was stored verbatim in mixed case (the old open
    register) is lowercased by the startup self-heal and still authenticates."""
    await _insert_user("DemoUser@Example.com", "Tech")
    await _normalize_emails()  # what lifespan does each boot

    async with async_session_factory() as session:
        from sqlalchemy import select
        user = (
            await session.execute(select(User).where(User.email == "demouser@example.com"))
        ).scalar_one()
    assert user.email == "demouser@example.com"

    for variant in ("demouser@example.com", "DemoUser@Example.com"):
        resp = await client.post(
            "/api/auth/login", json={"email": variant, "password": "denya123"}
        )
        assert resp.status_code == 200, resp.text


async def test_login_matches_legacy_mixed_case_email_before_migration(client: AsyncClient):
    """Login compares on the normalized form, so an un-migrated mixed-case row
    is still matched by its lowercase login (no hard dependency on the
    self-heal having run)."""
    await _insert_user("DemoUser@Example.com", "Tech")
    resp = await client.post(
        "/api/auth/login", json={"email": "demouser@example.com", "password": "denya123"}
    )
    assert resp.status_code == 200, resp.text


async def test_legacy_email_normalization_is_idempotent(client: AsyncClient):
    """The startup self-heal rewrites once and no-ops on subsequent boots."""
    await _insert_user("DemoUser@Example.com", "Tech")
    async with async_session_factory() as session:
        first = await normalize_legacy_user_emails(session)
        await session.commit()
    async with async_session_factory() as session:
        second = await normalize_legacy_user_emails(session)
        await session.commit()
    assert first == 1
    assert second == 0


async def test_create_user_rejects_case_variant_of_legacy_email(client: AsyncClient):
    """The admin create-user duplicate check compares on the normalized form:
    creating a case-variant of a legacy mixed-case row returns 409, not 201."""
    await _insert_user("DemoUser@Example.com", "Tech")
    token = await _login(client)
    resp = await client.post(
        "/api/auth/users",
        json={
            "email": "demouser@example.com",
            "password": "password1",
            "full_name": "X",
            "role": "Tech",
        },
        headers=_auth(token),
    )
    assert resp.status_code == 409, resp.text


async def test_admin_only_rbac_gate(client: AsyncClient):
    """Canonical admins pass /api/auth/admin-only; everyone else 403."""
    wahab = await _login(client)
    assert (await client.get("/api/auth/admin-only", headers=_auth(wahab))).status_code == 200

    bella = await _login(client, email="bella@denya.com")
    assert (await client.get("/api/auth/admin-only", headers=_auth(bella))).status_code == 403


# ── Item 4: login rate limiting ───────────────────────────────────────
async def test_login_rate_limited_after_five_failures(client: AsyncClient):
    email, password = "rate-limited@example.com", "denya123"
    # Make sure the account exists with a valid password.
    token = await _login(client)
    await client.post(
        "/api/auth/users",
        json={"email": email, "password": password, "full_name": "Rate", "role": "CS Rep"},
        headers=_auth(token),
    )

    for _ in range(5):
        resp = await client.post(
            "/api/auth/login", json={"email": email, "password": "wrong-password"}
        )
        assert resp.status_code == 401

    # 6th attempt — even with the CORRECT password — is throttled.
    resp = await client.post(
        "/api/auth/login", json={"email": email, "password": password}
    )
    assert resp.status_code == 429, resp.text


async def test_rate_limit_is_per_email(client: AsyncClient):
    """Failures for one account never lock out another account."""
    token = await _login(client)
    for email in ("victim@example.com", "other@example.com"):
        await client.post(
            "/api/auth/users",
            json={"email": email, "password": "password1", "full_name": "U", "role": "CS Rep"},
            headers=_auth(token),
        )

    for _ in range(6):
        resp = await client.post(
            "/api/auth/login", json={"email": "victim@example.com", "password": "bad"}
        )
        assert resp.status_code in (401, 429)

    # Unaffected account still logs in fine.
    resp = await client.post(
        "/api/auth/login", json={"email": "other@example.com", "password": "password1"}
    )
    assert resp.status_code == 200, resp.text


async def test_rate_limit_window_expires(client: AsyncClient, monkeypatch):
    """After the 15-minute window passes, the account can log in again."""
    import time as _time

    import app.core.ratelimit as ratelimit_mod

    email, password = "window@example.com", "password1"
    token = await _login(client)
    await client.post(
        "/api/auth/users",
        json={"email": email, "password": password, "full_name": "W", "role": "CS Rep"},
        headers=_auth(token),
    )

    # Pin the limiter clock so the window can be fast-forwarded deterministically.
    clock = {"now": _time.time()}
    monkeypatch.setattr(ratelimit_mod, "_now", lambda: clock["now"])
    for _ in range(5):
        await client.post("/api/auth/login", json={"email": email, "password": "bad"})

    blocked = await client.post("/api/auth/login", json={"email": email, "password": password})
    assert blocked.status_code == 429, blocked.text

    clock["now"] += settings.LOGIN_RATE_LIMIT_WINDOW_SECONDS + 1
    resp = await client.post("/api/auth/login", json={"email": email, "password": password})
    assert resp.status_code == 200, resp.text


# ── Item 5: WhatsApp webhook secret (fail closed) ─────────────────────
WEBHOOK_BODY = {
    "object": "whatsapp_business_account",
    "entry": [
        {
            "id": "1",
            "changes": [
                {
                    "id": "wamid.1",
                    "message": {"from": "+233000000000", "id": "wamid.1", "text": {"text": "AC leaking"}},
                }
            ],
        }
    ],
}


async def test_webhook_fail_closed_when_env_unset(client: AsyncClient):
    """WHATSAPP_WEBHOOK_SECRET unset ⇒ every message rejected (403)."""
    assert settings.WHATSAPP_WEBHOOK_SECRET == ""  # test env default is unset
    resp = await client.post("/api/whatsapp/webhook", json=WEBHOOK_BODY)
    assert resp.status_code == 403
    assert "not configured" in resp.json()["detail"].lower()


async def test_webhook_rejects_missing_or_wrong_secret(client: AsyncClient, monkeypatch):
    monkeypatch.setattr(settings, "WHATSAPP_WEBHOOK_SECRET", "test-webhook-secret")
    resp = await client.post("/api/whatsapp/webhook", json=WEBHOOK_BODY)
    assert resp.status_code == 403

    resp = await client.post(
        "/api/whatsapp/webhook", json=WEBHOOK_BODY, headers={"X-Webhook-Secret": "wrong"}
    )
    assert resp.status_code == 403


async def test_webhook_accepts_valid_secret_and_creates_ticket(client: AsyncClient, monkeypatch):
    from app.routers import whatsapp as whatsapp_router

    async def _fake_reply(to_phone: str, text: str):
        from app.schemas.whatsapp import WhatsAppReplyResponse
        return WhatsAppReplyResponse(success=True, message="sent")

    monkeypatch.setattr(settings, "WHATSAPP_WEBHOOK_SECRET", "test-webhook-secret")
    monkeypatch.setattr(whatsapp_router, "send_whatsapp_reply", _fake_reply)

    resp = await client.post(
        "/api/whatsapp/webhook",
        json=WEBHOOK_BODY,
        headers={"X-Webhook-Secret": "test-webhook-secret"},
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()
    assert data["status"] == "processed"
    assert data["ticket_number"].startswith("PAV-")

    # The message is logged and visible to an authenticated mock-log caller.
    token = await _login(client)
    log = await client.get("/api/whatsapp/mock-log", headers=_auth(token))
    assert log.status_code == 200
    assert len(log.json()) == 1
    assert log.json()[0]["ticket_number"] == data["ticket_number"]


async def test_webhook_verify_token_mismatch_403(client: AsyncClient, monkeypatch):
    monkeypatch.setattr(settings, "WHATSAPP_VERIFY_TOKEN", "verify-me")
    resp = await client.get(
        "/api/whatsapp/webhook",
        params={"hub.mode": "subscribe", "hub.verify_token": "nope", "hub.challenge": "1234"},
    )
    assert resp.status_code == 403


async def test_webhook_verify_token_match_returns_challenge(client: AsyncClient, monkeypatch):
    monkeypatch.setattr(settings, "WHATSAPP_VERIFY_TOKEN", "verify-me")
    resp = await client.get(
        "/api/whatsapp/webhook",
        params={"hub.mode": "subscribe", "hub.verify_token": "verify-me", "hub.challenge": "1234"},
    )
    assert resp.status_code == 200
    assert resp.json() == {"challenge": "1234"}


# ── Item 6: mock-log requires auth ────────────────────────────────────
async def test_mock_log_requires_auth(client: AsyncClient):
    resp = await client.get("/api/whatsapp/mock-log")
    assert resp.status_code == 401, resp.text

    token = await _login(client)
    resp = await client.get("/api/whatsapp/mock-log", headers=_auth(token))
    assert resp.status_code == 200
    assert resp.json() == []


# ── Item 7: security headers ──────────────────────────────────────────
async def test_security_headers_on_html_page(client: AsyncClient):
    resp = await client.get("/login")
    assert resp.status_code == 200
    assert resp.headers["x-frame-options"] == "DENY"
    assert resp.headers["x-content-type-options"] == "nosniff"
    assert "content-security-policy" in resp.headers
    assert "default-src 'self'" in resp.headers["content-security-policy"]


async def test_csp_only_on_html_not_json_api(client: AsyncClient):
    resp = await client.get("/api/tickets")
    assert resp.status_code == 200
    assert "content-type" in resp.headers and resp.headers["content-type"].startswith("application/json")
    assert "content-security-policy" not in resp.headers
    # Frame/type hardening headers apply everywhere.
    assert resp.headers["x-frame-options"] == "DENY"
    assert resp.headers["x-content-type-options"] == "nosniff"


async def test_hsts_only_when_tls_terminates(client: AsyncClient):
    plain = await client.get("/login")
    assert "strict-transport-security" not in plain.headers

    tls = await client.get("/login", headers={"X-Forwarded-Proto": "https"})
    assert tls.headers["strict-transport-security"] == "max-age=31536000; includeSubDomains"


# ── Item 8: pagination (page/limit) and sane max page size ────────────
async def test_limit_alias_over_cap_rejected(client: AsyncClient, seed_tickets):
    await seed_tickets(10)
    resp = await client.get("/api/tickets", params={"limit": 500})
    assert resp.status_code == 422


async def test_limit_alias_paginates(client: AsyncClient, seed_tickets):
    await seed_tickets(14)
    resp = await client.get("/api/tickets", params={"page": 2, "limit": 5})
    assert resp.status_code == 200
    data = resp.json()
    assert data["total"] == 14
    assert len(data["items"]) == 5
    assert data["page_size"] == 5
    assert data["page"] == 2


async def test_page_beyond_last_returns_empty_with_total(client: AsyncClient, seed_tickets):
    await seed_tickets(7)
    resp = await client.get("/api/tickets", params={"page": 999, "page_size": 10})
    assert resp.status_code == 200
    data = resp.json()
    assert data["items"] == []
    assert data["total"] == 7


async def test_page_size_and_limit_conflict_is_422(client: AsyncClient, seed_tickets):
    await seed_tickets(3)
    resp = await client.get("/api/tickets", params={"page_size": 10, "limit": 20})
    assert resp.status_code == 422