AR-Autopilot/arautopilot/core/user.py

"""User entity + PIN hashing.

Sprint 2.5: minimal model to back the Studio login. PINs are 4-8 digit
numeric strings hashed with PBKDF2-HMAC-SHA256 (stdlib, no extra
dependency) using a per-user 16-byte salt and a 200k-iteration work
factor. Hash format follows a self-describing string so future migrations
(argon2, scrypt) can co-exist.

Format::

    pbkdf2_sha256$<iterations>$<base64(salt)>$<base64(hash)>
"""

from __future__ import annotations

import base64
import hashlib
import hmac
import os
from datetime import UTC, datetime

from pydantic import BaseModel, ConfigDict, Field, field_validator

from arautopilot.core.ids import new_vessel_id
from arautopilot.core.rbac import Role

_PBKDF2_ITERATIONS = 200_000
_PBKDF2_SALT_LEN = 16
_PBKDF2_HASH_LEN = 32
_PBKDF2_ALGO = "sha256"


def _hash_pin(pin: str, *, iterations: int = _PBKDF2_ITERATIONS) -> str:
    """Hash a PIN with PBKDF2-HMAC-SHA256. Returns the self-describing string."""
    if not _looks_like_pin(pin):
        raise ValueError(
            "PIN must be 4-8 digits (numeric). Use ASCII digits only."
        )
    salt = os.urandom(_PBKDF2_SALT_LEN)
    digest = hashlib.pbkdf2_hmac(
        _PBKDF2_ALGO, pin.encode("utf-8"), salt, iterations, _PBKDF2_HASH_LEN
    )
    return (
        f"pbkdf2_{_PBKDF2_ALGO}${iterations}"
        f"${base64.b64encode(salt).decode('ascii')}"
        f"${base64.b64encode(digest).decode('ascii')}"
    )


def _verify_pin(pin: str, hashed: str) -> bool:
    """Constant-time verification of a PIN against a stored hash."""
    if not _looks_like_pin(pin):
        return False
    try:
        scheme, iters_s, salt_b64, hash_b64 = hashed.split("$", 3)
    except ValueError:
        return False
    if scheme != f"pbkdf2_{_PBKDF2_ALGO}":
        return False
    try:
        iterations = int(iters_s)
        salt = base64.b64decode(salt_b64)
        expected = base64.b64decode(hash_b64)
    except (ValueError, TypeError):
        return False
    candidate = hashlib.pbkdf2_hmac(
        _PBKDF2_ALGO, pin.encode("utf-8"), salt, iterations, len(expected)
    )
    return hmac.compare_digest(candidate, expected)


def _looks_like_pin(pin: str) -> bool:
    return bool(pin) and 4 <= len(pin) <= 8 and pin.isdigit()


class User(BaseModel):
    """A user of the Studio or the bridge display.

    Field ``pin_hash`` is the only sensitive value persisted -- the plain
    PIN never lives in memory longer than the duration of a verify call.
    """

    model_config = ConfigDict(extra="forbid", validate_assignment=True)

    user_id: str = Field(default_factory=lambda: new_vessel_id())
    display_name: str = Field(min_length=1, max_length=80)
    role: Role
    pin_hash: str = Field(min_length=8, max_length=300)
    vessel_id: str | None = Field(
        default=None,
        description="If set, this user belongs to one specific vessel "
                    "(Owners + their crew). None means cross-vessel scope "
                    "(Super Admin + Engineer).",
    )
    active: bool = True
    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
    last_login_at: datetime | None = None

    @field_validator("pin_hash")
    @classmethod
    def _looks_like_hash(cls, v: str) -> str:
        # Cheap structural validation -- does not verify the hash itself.
        parts = v.split("$")
        if len(parts) != 4 or not parts[0].startswith("pbkdf2_"):
            raise ValueError(
                "pin_hash must be in the form pbkdf2_<algo>$<iter>$<salt>$<hash>"
            )
        return v

    # ----- Construction helpers -------------------------------------------
    @classmethod
    def create(
        cls,
        *,
        display_name: str,
        role: Role,
        pin: str,
        vessel_id: str | None = None,
    ) -> "User":
        """Construct a new user from a plaintext PIN.

        The PIN is hashed before the model is built; the plaintext does
        not survive this call.
        """
        return cls(
            display_name=display_name,
            role=role,
            pin_hash=_hash_pin(pin),
            vessel_id=vessel_id,
        )

    # ----- Authentication --------------------------------------------------
    def verify_pin(self, pin: str) -> bool:
        """Return True iff ``pin`` matches this user's stored hash."""
        return _verify_pin(pin, self.pin_hash)

    def set_pin(self, pin: str) -> "User":
        """Return a copy with a freshly-hashed new PIN."""
        return self.model_copy(update={"pin_hash": _hash_pin(pin)})

    def touch_login(self) -> "User":
        """Return a copy with ``last_login_at`` refreshed to now (UTC)."""
        return self.model_copy(update={"last_login_at": datetime.now(UTC)})