PacketStation/tests/server/test_signin_api.py

"""Sign-in, Me, sign-out, lock and unlock through the API, on PostgreSQL with the fake directory."""

from __future__ import annotations

from fakedir import FakeDirectory
from fastapi.testclient import TestClient
from pshelpers import app_conn, needs_pg, sign_in

from packetstation import audit
from packetstation.settings import Settings

pytestmark = needs_pg

GENERIC = "Sign-in failed. Check the account name and password."
NO_ROLE = "No role in Packet Station. Ask IT to add you to a group."


def test_password_signin_and_me_shape(pg_client: TestClient, pg_settings: Settings, fake_dir: FakeDirectory) -> None:
    r = sign_in(pg_client, "clerk.clara")
    assert r.status_code == 200, r.text
    cookie = r.headers["set-cookie"].lower()
    assert cookie.startswith("ps_session=") and "httponly" in cookie and "samesite=strict" in cookie
    assert "max-age" not in cookie and "secure" not in cookie  # plain http in tests
    me = r.json()
    assert set(me) == {"user", "roles", "screens", "groups", "session", "pin", "can_register_device"}
    clara = fake_dir.user("clerk.clara")
    assert me["user"] == {
        "id": str(clara.guid),
        "sam_account": "clerk.clara",
        "upn": "[email protected]",
        "display_name": "Clara Clerk",
        "emp_no": "4001",
        "email": "[email protected]",
    }
    assert me["roles"] == ["clerical"] and me["screens"] == ["clerical"] and me["groups"] == ["PS-Clerical"]
    assert set(me["session"]) == {
        "id",
        "kind",
        "created_at",
        "expires_at",
        "locked",
        "locked_since",
        "lock_minutes",
        "device",
    }
    assert me["session"]["kind"] == "password" and me["session"]["locked"] is False
    assert me["session"]["lock_minutes"] == 60  # Clerical: the desk minutes (issue #45)
    assert me["session"]["device"] is None and me["session"]["locked_since"] is None
    assert me["pin"] == {"set": False, "disabled": False} and me["can_register_device"] is False
    assert pg_client.get("/api/auth/me").json() == me

    with app_conn(pg_settings) as conn:
        hours = conn.execute(
            "SELECT extract(epoch FROM expires_at - created_at) / 3600 FROM sessions WHERE id = %s",
            (me["session"]["id"],),
        ).fetchone()[0]
        assert float(hours) == 12
        assert conn.execute("SELECT role::text, dept_code FROM user_roles").fetchall() == [("clerical", "CL")]
        ok = conn.execute("SELECT account, kind, ok FROM signin_attempts").fetchall()
        assert ok == [("clerk.clara", "password", True)]
        row = conn.execute("SELECT action, detail, role_snapshot FROM audit_events").fetchone()
        assert row[0] == "auth.signin" and row[1]["kind"] == "password" and row[2] == "clerical"
        assert conn.execute("SELECT last_signin_at IS NOT NULL FROM users").fetchone()[0]


def test_roles_union_nested_and_upn(pg_client: TestClient) -> None:
    me = sign_in(pg_client, "multi.mia").json()
    assert me["roles"] == ["clerical", "shop:06:supervisor"]
    assert me["screens"] == ["clerical", "supervisor", "stall"] and me["can_register_device"] is True
    assert sign_in(pg_client, "nested.ned").json()["roles"] == ["shop:10:worker"]
    assert sign_in(pg_client, "[email protected]").json()["roles"] == ["shop:03:supervisor"]
    admin = sign_in(pg_client, "ADMIN.ANN").json()
    assert admin["roles"] == ["admin"] and admin["screens"][-1] == "admin" and len(admin["screens"]) == 7


def test_failures_do_not_reveal_which(pg_client: TestClient, pg_settings: Settings, fake_dir: FakeDirectory) -> None:
    for login, pw in (("clerk.clara", "wrong"), ("nobody.here", None), ("disabled.dan", None), ("clerk.clara", "")):
        r = sign_in(pg_client, login, pw)
        assert r.status_code == 401, login
        assert r.json() == {"detail": GENERIC, "code": "signin_failed"}
        assert "set-cookie" not in r.headers
    r = sign_in(pg_client, "norole.nick")
    assert r.status_code == 403 and r.json() == {"detail": NO_ROLE, "code": "no_role"}
    with app_conn(pg_settings) as conn:
        reasons = [r[0] for r in conn.execute("SELECT reason FROM signin_attempts ORDER BY id").fetchall()]
        assert reasons == ["bad_password", "no_such_user", "disabled", "empty", "no_role"]
        audited = conn.execute("SELECT detail->>'reason' FROM audit_events ORDER BY id").fetchall()
        assert [a[0] for a in audited] == ["disabled", "no_role"]
        assert conn.execute("SELECT count(*) FROM sessions").fetchone()[0] == 0


def test_account_rate_limit_before_bind(pg_client: TestClient, fake_dir: FakeDirectory) -> None:
    for _ in range(5):
        assert sign_in(pg_client, "weld.andy", "wrong").status_code == 401
    binds = fake_dir.binds
    r = sign_in(pg_client, "weld.andy")  # even the right password is refused now
    assert r.status_code == 429
    body = r.json()
    assert body["code"] == "rate_limited" and 590 <= body["retry_after"] <= 600
    assert r.headers["retry-after"] == str(body["retry_after"])
    assert fake_dir.binds == binds  # no bind while limited
    r = sign_in(pg_client, "[email protected]")  # the UPN alias is limited too
    assert r.status_code == 429 and fake_dir.binds == binds
    assert sign_in(pg_client, "clerk.clara").status_code == 200  # other accounts are not


def test_ldap_unavailable(pg_client: TestClient, fake_dir: FakeDirectory) -> None:
    fake_dir.down = True
    r = sign_in(pg_client, "clerk.clara")
    assert r.status_code == 503 and r.json()["code"] == "ldap_unavailable"


def test_signout(pg_client: TestClient, pg_settings: Settings) -> None:
    sign_in(pg_client, "clerk.clara")
    r = pg_client.post("/api/auth/signout")
    assert r.status_code == 204
    assert pg_client.get("/api/auth/me").json()["code"] == "no_session"
    with app_conn(pg_settings) as conn:
        assert conn.execute("SELECT revoked_reason FROM sessions").fetchone()[0] == "signout"


def test_lock_unlock_and_audit_chain(pg_client: TestClient, pg_settings: Settings) -> None:
    sign_in(pg_client, "clerk.clara")
    assert pg_client.post("/api/auth/lock").status_code == 204
    r = pg_client.get("/api/account")
    assert r.status_code == 423 and r.json()["code"] == "locked"
    me = pg_client.get("/api/auth/me").json()
    assert me["session"]["locked"] is True and me["session"]["locked_since"]
    r = pg_client.post("/api/auth/unlock", json={"factor": "password", "secret": "wrong"})
    assert r.status_code == 403 and r.json()["code"] == "wrong_secret" and r.json()["attempts_left"] == 2
    r = pg_client.post("/api/auth/unlock", json={"factor": "pin", "secret": "123456"})
    assert r.status_code == 409 and r.json()["code"] == "pin_not_set"
    r = pg_client.post("/api/auth/unlock", json={"factor": "password", "secret": "fake-password-1"})
    assert r.status_code == 200 and r.json()["session"]["locked"] is False
    assert pg_client.get("/api/account").status_code == 200
    assert pg_client.post("/api/auth/visibility", json={"hidden": False}).status_code == 204
    assert pg_client.get("/api/account").status_code == 200
    assert pg_client.post("/api/auth/visibility", json={"hidden": True}).status_code == 204
    assert pg_client.get("/api/account").status_code == 200  # hidden under a minute: no lock
    assert pg_client.post("/api/auth/signout").status_code == 204
    with app_conn(pg_settings) as conn:
        actions = [r[0] for r in conn.execute("SELECT action FROM audit_events ORDER BY id").fetchall()]
        assert actions == ["auth.signin", "auth.lock", "auth.unlock", "auth.signout"]
        sources = conn.execute("SELECT detail->>'source' FROM audit_events WHERE action = 'auth.lock' ORDER BY id")
        assert [s[0] for s in sources.fetchall()] == ["manual"]
        ok, n, detail = audit.verify_chain(conn)
        assert ok and n == 4, detail


def test_three_wrong_unlocks_end_the_session(pg_client: TestClient, pg_settings: Settings) -> None:
    sign_in(pg_client, "clerk.clara")
    pg_client.post("/api/auth/lock")
    for left in (2, 1):
        r = pg_client.post("/api/auth/unlock", json={"factor": "password", "secret": "wrong"})
        assert r.json()["attempts_left"] == left
    r = pg_client.post("/api/auth/unlock", json={"factor": "password", "secret": "wrong"})
    assert r.status_code == 401 and r.json()["code"] == "session_ended"
    assert pg_client.get("/api/auth/me").json()["code"] == "no_session"
    with app_conn(pg_settings) as conn:
        assert conn.execute("SELECT revoked_reason FROM sessions").fetchone()[0] == "stepup_failures"
        kinds = conn.execute("SELECT kind, account FROM signin_attempts WHERE NOT ok").fetchall()
        assert len(kinds) == 3 and all(k == "unlock" and a.startswith("user:") for k, a in kinds)
        assert audit.verify_chain(conn)[0]


def test_departments(pg_client: TestClient) -> None:
    sign_in(pg_client, "weld.andy")
    deps = pg_client.get("/api/departments").json()
    assert [d["code"] for d in deps][:5] == ["CL", "WH", "QC", "PM", "01"]
    weld = next(d for d in deps if d["code"] == "03")
    assert {k: v for k, v in weld.items() if k != "stations"} == {
        "code": "03",
        "slug": "WeldParts",
        "name": "Weld Parts",
        "kind": "shop",
        "weld": True,
    }
    assert [s["name"] for s in weld["stations"]] == [f"Stall {i}" for i in range(1, 7)]
    assert all(isinstance(s["id"], int) for s in weld["stations"])
    hydro = next(d for d in deps if d["code"] == "09")
    assert [s["name"] for s in hydro["stations"]] == ["Hydro Bay 1", "Hydro Bay 2"] and hydro["weld"] is False