PacketStation/server/packetstation/auth/stepup.py

"""Step-up checks inside a session (screen unlock and signatures): PIN or domain password.

Wrong attempts are recorded (account `user:<uuid>`), count toward the PIN's failures when the factor is the
PIN, and raise `sessions.stepup_failures`. At STEPUP_MAX_FAILURES the session ends. A right answer resets
the session's count.
"""

from __future__ import annotations

from typing import Any

from .. import audit
from ..api.errors import ApiError
from ..settings import Settings
from . import pin as pins
from . import ratelimit, sessions
from .ldap import DirectoryProtocol, LdapUnavailable
from .signin import ldap_down, rate_gate

FACTORS = ("pin", "password")


def end_session(conn: Any, actor: sessions.Actor, reason: str, ip: str | None) -> None:
    with conn.transaction():
        if sessions.revoke(conn, actor.session_id, reason):
            audit.append(
                conn,
                action="auth.session_ended",
                user_id=actor.user_id,
                session_id=actor.session_id,
                role_snapshot=actor.role_snapshot,
                detail={"reason": reason},
                ip=ip,
            )


def verify(
    conn: Any,
    directory: DirectoryProtocol,
    settings: Settings,
    actor: sessions.Actor,
    *,
    factor: str,
    secret: str,
    kind: str,
    ip: str | None,
) -> None:
    """Return when the secret is right; raise ApiError otherwise.

    Wrong: 403 `wrong_secret` with `attempts_left`. The last allowed miss: 401 `session_ended`.
    """
    if factor not in FACTORS:
        raise ApiError(400, "bad_factor", "factor must be pin or password")
    if kind not in ("unlock", "signature"):
        raise ValueError(f"bad step-up kind {kind!r}")
    account = f"user:{actor.user_id}"
    secret = secret or ""
    if factor == "pin":
        stored = pins.stored_hash(conn, actor.user_id)
        if stored is None:
            raise ApiError(409, "pin_not_set", "No PIN is set. Use your password.")
        if stored[1]:
            raise ApiError(409, "pin_disabled", "Your PIN is disabled. Use your password.")
        ok = pins.verify_hash(stored[0], secret)
    else:
        rate_gate(
            conn,
            settings,
            accounts=[account, actor.sam_account.lower()],
            ip=ip,
            kind=kind,
            record_account=account,
            user_id=actor.user_id,
            session_id=actor.session_id,
        )
        try:
            u = directory.get_user_by_guid(actor.user_id)
            if u is None or u.disabled:
                end_session(conn, actor, "roles_lost", ip)
                raise ApiError(401, "session_ended", "Your account is no longer active. Sign in again.")
            ok = bool(secret) and directory.verify_password(u.dn, secret)
        except LdapUnavailable as e:
            raise ldap_down(e) from None

    if ok:
        with conn.transaction():
            conn.execute(
                "UPDATE sessions SET stepup_failures = 0 WHERE id = %s AND stepup_failures <> 0", (actor.session_id,)
            )
            if factor == "pin":
                pins.reset_failures(conn, actor.user_id)
            ratelimit.record(
                conn, account=account, ip=ip, kind=kind, ok=True, user_id=actor.user_id, session_id=actor.session_id
            )
        return

    pin_disabled = False
    with conn.transaction():
        ratelimit.record(
            conn,
            account=account,
            ip=ip,
            kind=kind,
            ok=False,
            reason="bad_pin" if factor == "pin" else "bad_password",
            user_id=actor.user_id,
            session_id=actor.session_id,
        )
        if factor == "pin":
            pin_disabled = pins.register_failure(
                conn,
                settings,
                actor.user_id,
                session_id=actor.session_id,
                role_snapshot=actor.role_snapshot,
                ip=ip,
                where=kind,
            )
        row = conn.execute(
            "UPDATE sessions SET stepup_failures = stepup_failures + 1 WHERE id = %s RETURNING stepup_failures",
            (actor.session_id,),
        ).fetchone()
        failures = int(row[0]) if row else settings.stepup_max_failures
    if failures >= settings.stepup_max_failures:
        end_session(conn, actor, "stepup_failures", ip)
        raise ApiError(
            401,
            "session_ended",
            f"{settings.stepup_max_failures} wrong attempts. The session has ended. Sign in again.",
        )
    raise ApiError(
        403,
        "wrong_secret",
        "Wrong PIN." if factor == "pin" else "Wrong password.",
        extra={"attempts_left": settings.stepup_max_failures - failures, "pin_disabled": pin_disabled},
    )