PacketStation/tests/server/test_pin.py

"""The Packet Station PIN: format, hashing, and the whole lifecycle through the API."""

from __future__ import annotations

from pathlib import Path

import pytest
from fakedir import FAKE_PASSWORD, FakeDirectory
from fastapi.testclient import TestClient
from pshelpers import PIN, app_conn, make_settings, needs_pg, pin_signin, register_device, set_pin, sign_in

from packetstation.auth import pin as pins
from packetstation.settings import Settings


@pytest.mark.parametrize(
    ("value", "ok"),
    [
        ("123456", True),
        ("123456789012", True),
        ("12345", False),
        ("1234567890123", False),
        ("12345a", False),
        ("١٢٣٤٥٦", False),
        ("", False),
    ],
)
def test_format(tmp_path: Path, value: str, ok: bool) -> None:
    assert pins.valid_format(value, make_settings(tmp_path)) is ok


def test_hash_is_argon2id_and_verifies() -> None:
    h = pins.hash_pin(PIN)
    assert h.startswith("$argon2id$") and PIN not in h
    assert pins.verify_hash(h, PIN) and not pins.verify_hash(h, "135791") and not pins.verify_hash("junk", PIN)


@needs_pg
def test_pin_needs_a_registered_device(pg_client: TestClient) -> None:
    set_pin(pg_client, "weld.andy")
    r = pg_client.post("/api/auth/signin/pin", json={"emp_no": "3003", "pin": PIN})
    assert r.status_code == 403 and r.json()["code"] == "device_required"
    assert pg_client.get("/api/auth/context").json()["device"] is None


@needs_pg
def test_only_registrars_register_devices(pg_client: TestClient) -> None:
    sign_in(pg_client, "weld.andy")
    r = pg_client.post("/api/account/device", json={"name": "x"})
    assert r.status_code == 403
    sign_in(pg_client, "lead.lou")
    assert pg_client.post("/api/account/device", json={"name": "x"}).status_code == 403
    sign_in(pg_client, "admin.ann")
    assert pg_client.post("/api/account/device", json={"name": "Office PC"}).status_code == 200


@needs_pg
def test_pin_set_rules(pg_client: TestClient, pg_settings: Settings) -> None:
    sign_in(pg_client, "weld.andy")
    r = pg_client.post("/api/account/pin", json={"password": FAKE_PASSWORD, "pin": "12ab56"})
    assert r.status_code == 400 and r.json()["code"] == "invalid_pin"
    r = pg_client.post("/api/account/pin", json={"password": "wrong", "pin": PIN})
    assert r.status_code == 403 and r.json()["code"] == "bad_password"
    assert pg_client.get("/api/account").json()["pin"] == {"set": False, "disabled": False}
    with app_conn(pg_settings) as conn:
        row = conn.execute("SELECT account, kind, reason FROM signin_attempts WHERE NOT ok").fetchone()
        assert row[0].startswith("user:") and row[1:] == ("pin_set", "bad_password")
        assert conn.execute("SELECT count(*) FROM user_pins").fetchone()[0] == 0


@needs_pg
def test_pin_lifecycle(pg_client: TestClient, pg_settings: Settings, fake_dir: FakeDirectory) -> None:
    device = register_device(pg_client)
    assert pg_client.get("/api/auth/context").json()["device"] == device
    set_pin(pg_client, "weld.andy")

    r = pin_signin(pg_client, "3003")
    assert r.status_code == 200, r.text
    me = r.json()
    assert me["session"]["kind"] == "pin" and me["session"]["device"] == device
    assert me["roles"] == ["shop:03:worker"] and me["pin"] == {"set": True, "disabled": False}
    # A PIN session cannot change account settings
    r = pg_client.post("/api/account/pin", json={"password": FAKE_PASSWORD, "pin": "135791"})
    assert r.status_code == 403 and r.json()["code"] == "password_session_required"
    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) == 10
        detail = conn.execute(
            "SELECT detail FROM audit_events WHERE action = 'auth.signin' ORDER BY id DESC"
        ).fetchone()
        assert detail[0] == {"kind": "pin", "device_id": device["id"], "device_name": device["name"]}
    log = pg_client.get("/api/account/access-log").json()
    assert any(e["action"] == "auth.signin" and e["detail"]["device_name"] == device["name"] for e in log)
    pg_client.post("/api/auth/signout")

    # Unknown, duplicate and wrong all answer the same
    for emp in ("0000", "9001"):
        r = pin_signin(pg_client, emp)
        assert r.status_code == 401 and r.json() == {"detail": "Sign-in failed.", "code": "signin_failed"}

    # Five wrong PINs disable it
    for _ in range(5):
        assert pin_signin(pg_client, "3003", "000000").status_code == 401
    with app_conn(pg_settings) as conn:
        failed, disabled = conn.execute("SELECT failed_count, disabled_at IS NOT NULL FROM user_pins").fetchone()
        assert failed == 5 and disabled
        assert conn.execute("SELECT count(*) FROM audit_events WHERE action = 'pin.disabled'").fetchone()[0] == 1
    assert pin_signin(pg_client, "3003").status_code in (401, 429)  # disabled (and the account is now limited)

    # A password sign-in and a new PIN bring it back
    set_pin(pg_client, "weld.andy", "13579135")
    with app_conn(pg_settings) as conn:
        assert conn.execute("SELECT failed_count, disabled_at FROM user_pins").fetchone() == (0, None)
        assert conn.execute("SELECT count(*) FROM audit_events WHERE action = 'pin.set'").fetchone()[0] == 2


@needs_pg
def test_pin_unlock_failures_count_and_end_session(pg_client: TestClient, pg_settings: Settings) -> None:
    register_device(pg_client)
    set_pin(pg_client, "weld.andy")
    assert pin_signin(pg_client, "3003").status_code == 200
    pg_client.post("/api/auth/lock")
    r = pg_client.post("/api/auth/unlock", json={"factor": "pin", "secret": "111111"})
    assert r.status_code == 403 and r.json()["attempts_left"] == 2
    r = pg_client.post("/api/auth/unlock", json={"factor": "pin", "secret": PIN})
    assert r.status_code == 200 and r.json()["session"]["locked"] is False
    with app_conn(pg_settings) as conn:
        assert conn.execute("SELECT failed_count FROM user_pins").fetchone()[0] == 0
    pg_client.post("/api/auth/lock")
    for _ in range(2):
        assert pg_client.post("/api/auth/unlock", json={"factor": "pin", "secret": "111111"}).status_code == 403
    r = pg_client.post("/api/auth/unlock", json={"factor": "pin", "secret": "111111"})
    assert r.status_code == 401 and r.json()["code"] == "session_ended"
    with app_conn(pg_settings) as conn:
        assert conn.execute("SELECT failed_count FROM user_pins").fetchone()[0] == 3
        reason = conn.execute("SELECT revoked_reason FROM sessions WHERE kind = 'pin'").fetchone()[0]
        assert reason == "stepup_failures"