PacketStation/server/packetstation/auth/pin.py

"""The optional Packet Station PIN: Argon2id hash on the server, set by the user, cleared by an Admin."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any

from argon2 import PasswordHasher
from argon2.exceptions import InvalidHashError, VerificationError, VerifyMismatchError

from .. import audit
from ..settings import Settings

_hasher = PasswordHasher()


@dataclass
class PinState:
    set: bool
    disabled: bool
    failed_count: int = 0

    def public(self) -> dict[str, bool]:
        return {"set": self.set, "disabled": self.disabled}


def valid_format(pin: str, settings: Settings) -> bool:
    """Digits only (ASCII), PIN_MIN_DIGITS to PIN_MAX_DIGITS long."""
    return pin.isascii() and pin.isdigit() and settings.pin_min_digits <= len(pin) <= settings.pin_max_digits


def hash_pin(pin: str) -> str:
    return _hasher.hash(pin)


def verify_hash(stored: str, pin: str) -> bool:
    if not pin:
        return False
    try:
        return _hasher.verify(stored, pin)
    except (VerifyMismatchError, VerificationError, InvalidHashError):
        return False


def state(conn: Any, user_id: Any) -> PinState:
    row = conn.execute("SELECT failed_count, disabled_at FROM user_pins WHERE user_id = %s", (user_id,)).fetchone()
    if row is None:
        return PinState(False, False)
    return PinState(True, row[1] is not None, int(row[0]))


def stored_hash(conn: Any, user_id: Any) -> tuple[str, bool] | None:
    """(hash, disabled) or None when the user has no PIN."""
    row = conn.execute("SELECT hash, disabled_at FROM user_pins WHERE user_id = %s", (user_id,)).fetchone()
    return None if row is None else (str(row[0]), row[1] is not None)


def set_pin(conn: Any, user_id: Any, pin: str) -> None:
    """Store a new PIN hash and clear the failure count and the disabled mark."""
    h = hash_pin(pin)
    conn.execute(
        "INSERT INTO user_pins (user_id, hash, failed_count, disabled_at, set_at) VALUES (%s, %s, 0, NULL, now())"
        " ON CONFLICT (user_id) DO UPDATE SET hash = EXCLUDED.hash, failed_count = 0, disabled_at = NULL,"
        " set_at = now()",
        (user_id, h),
    )


def reset_failures(conn: Any, user_id: Any) -> None:
    conn.execute("UPDATE user_pins SET failed_count = 0 WHERE user_id = %s AND failed_count <> 0", (user_id,))


def register_failure(
    conn: Any,
    settings: Settings,
    user_id: Any,
    *,
    session_id: Any = None,
    role_snapshot: str | None = None,
    ip: str | None = None,
    where: str = "signin",
) -> bool:
    """Count one wrong PIN. At PIN_MAX_FAILURES the PIN is disabled and `pin.disabled` audited. True if disabled now."""
    with conn.transaction():
        row = conn.execute(
            "UPDATE user_pins SET failed_count = failed_count + 1 WHERE user_id = %s AND disabled_at IS NULL"
            " RETURNING failed_count",
            (user_id,),
        ).fetchone()
        if row is None or int(row[0]) < settings.pin_max_failures:
            return False
        conn.execute("UPDATE user_pins SET disabled_at = now() WHERE user_id = %s", (user_id,))
        audit.append(
            conn,
            action="pin.disabled",
            user_id=user_id,
            session_id=session_id,
            role_snapshot=role_snapshot,
            subject_type="user",
            subject_id=user_id,
            detail={"failures": int(row[0]), "where": where},
            ip=ip,
        )
        from ..domain import security_alerts

        security_alerts.pin_disabled(conn, user_id, int(row[0]), where)
    return True


def clear(conn: Any, user_id: Any) -> bool:
    row = conn.execute("DELETE FROM user_pins WHERE user_id = %s RETURNING user_id", (user_id,)).fetchone()
    return row is not None