feat: add operator landing and login

backend/app/services/auth_service.py

@@ -0,0 +1,175 @@
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+import secrets
+import threading
+import time
+from collections import deque
+from dataclasses import dataclass
+
+from app.core.config import Settings
+
+
+@dataclass(frozen=True)
+class AuthPrincipal:
+    username: str
+    expires_at: int
+
+
+class AuthService:
+    HASH_NAME = "pbkdf2_sha256"
+    HASH_ITERATIONS = 600_000
+    MAX_FAILURES = 5
+    FAILURE_WINDOW_SECONDS = 300
+    _failures: dict[str, deque[float]] = {}
+    _failure_lock = threading.Lock()
+
+    @staticmethod
+    def _b64_encode(value: bytes) -> str:
+        return base64.urlsafe_b64encode(value).decode("ascii").rstrip("=")
+
+    @staticmethod
+    def _b64_decode(value: str) -> bytes:
+        return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
+
+    @classmethod
+    def hash_password(
+        cls,
+        password: str,
+        *,
+        salt: bytes | None = None,
+        iterations: int | None = None,
+    ) -> str:
+        resolved_salt = salt or secrets.token_bytes(18)
+        resolved_iterations = iterations or cls.HASH_ITERATIONS
+        digest = hashlib.pbkdf2_hmac(
+            "sha256",
+            password.encode("utf-8"),
+            resolved_salt,
+            resolved_iterations,
+        )
+        return "$".join(
+            (
+                cls.HASH_NAME,
+                str(resolved_iterations),
+                cls._b64_encode(resolved_salt),
+                cls._b64_encode(digest),
+            )
+        )
+
+    @classmethod
+    def verify_password(cls, password: str, encoded: str) -> bool:
+        try:
+            algorithm, iterations_raw, salt_raw, expected_raw = encoded.split("$", 3)
+            if algorithm != cls.HASH_NAME:
+                return False
+            iterations = int(iterations_raw)
+            if iterations < 100_000 or iterations > 2_000_000:
+                return False
+            salt = cls._b64_decode(salt_raw)
+            expected = cls._b64_decode(expected_raw)
+            actual = hashlib.pbkdf2_hmac(
+                "sha256",
+                password.encode("utf-8"),
+                salt,
+                iterations,
+            )
+            return hmac.compare_digest(actual, expected)
+        except (TypeError, ValueError):
+            return False
+
+    @classmethod
+    def credentials_match(cls, username: str, password: str, settings: Settings) -> bool:
+        expected_username = settings.auth_username or ""
+        expected_password_hash = settings.auth_password_hash or ""
+        username_matches = hmac.compare_digest(
+            username.encode("utf-8"),
+            expected_username.encode("utf-8"),
+        )
+        password_matches = cls.verify_password(password, expected_password_hash)
+        return username_matches and password_matches
+
+    @classmethod
+    def create_session_token(cls, username: str, settings: Settings, *, now: int | None = None) -> str:
+        issued_at = int(time.time() if now is None else now)
+        payload = {
+            "exp": issued_at + settings.auth_session_ttl_seconds,
+            "iat": issued_at,
+            "jti": secrets.token_urlsafe(12),
+            "sub": username,
+            "v": 1,
+        }
+        encoded_payload = cls._b64_encode(
+            json.dumps(payload, separators=(",", ":"), sort_keys=True).encode("utf-8")
+        )
+        signature = hmac.new(
+            (settings.auth_session_secret or "").encode("utf-8"),
+            encoded_payload.encode("ascii"),
+            hashlib.sha256,
+        ).digest()
+        return f"{encoded_payload}.{cls._b64_encode(signature)}"
+
+    @classmethod
+    def verify_session_token(
+        cls,
+        token: str | None,
+        settings: Settings,
+        *,
+        now: int | None = None,
+    ) -> AuthPrincipal | None:
+        if not token:
+            return None
+        try:
+            encoded_payload, encoded_signature = token.split(".", 1)
+            expected_signature = hmac.new(
+                (settings.auth_session_secret or "").encode("utf-8"),
+                encoded_payload.encode("ascii"),
+                hashlib.sha256,
+            ).digest()
+            supplied_signature = cls._b64_decode(encoded_signature)
+            if not hmac.compare_digest(expected_signature, supplied_signature):
+                return None
+            payload = json.loads(cls._b64_decode(encoded_payload))
+            username = str(payload.get("sub") or "")
+            expires_at = int(payload.get("exp") or 0)
+            issued_at = int(payload.get("iat") or 0)
+            current = int(time.time() if now is None else now)
+            if payload.get("v") != 1 or username != settings.auth_username:
+                return None
+            if issued_at <= 0 or issued_at > current + 60 or expires_at <= current:
+                return None
+            if expires_at - issued_at > settings.auth_session_ttl_seconds:
+                return None
+            return AuthPrincipal(username=username, expires_at=expires_at)
+        except (TypeError, ValueError, json.JSONDecodeError, UnicodeDecodeError):
+            return None
+
+    @classmethod
+    def retry_after_seconds(cls, key: str, *, now: float | None = None) -> int:
+        current = time.monotonic() if now is None else now
+        with cls._failure_lock:
+            attempts = cls._failures.setdefault(key, deque())
+            while attempts and current - attempts[0] >= cls.FAILURE_WINDOW_SECONDS:
+                attempts.popleft()
+            if len(attempts) < cls.MAX_FAILURES:
+                if not attempts:
+                    cls._failures.pop(key, None)
+                return 0
+            return max(1, int(cls.FAILURE_WINDOW_SECONDS - (current - attempts[0])))
+
+    @classmethod
+    def record_failure(cls, key: str, *, now: float | None = None) -> None:
+        current = time.monotonic() if now is None else now
+        with cls._failure_lock:
+            attempts = cls._failures.setdefault(key, deque())
+            while attempts and current - attempts[0] >= cls.FAILURE_WINDOW_SECONDS:
+                attempts.popleft()
+            attempts.append(current)
+
+    @classmethod
+    def clear_failures(cls, key: str) -> None:
+        with cls._failure_lock:
+            cls._failures.pop(key, None)