From 92e22cd48dc458fd087068527f3c7e2a7ce27623 Mon Sep 17 00:00:00 2001 From: CyberMind-FR Date: Wed, 13 May 2026 09:08:56 +0200 Subject: [PATCH] feat(auth): branching /auth/login + MFA + TOTP enrol + set-password (ref #120) - Add PendingStore (JSON+TTL) for in-progress TOTP enrollment secrets - Inject _users_api package loader (importlib) to side-step api/ name collision - Register _session_validator, _on_session_event, _revoke_sessions callbacks - POST /auth/login now branches: setup / mfa-challenge / totp-enroll / access - POST /auth/login/mfa: verifies TOTP code or backup code, issues full JWT - POST /auth/totp/enroll: generates secret, stores in PendingStore, returns URI - POST /auth/totp/confirm: verifies code, calls engine.enroll_totp, returns JWT - POST /auth/set-password: handles set-password scope and full-JWT change - Mount _login_router before auth_router so /auth/login override wins - Fix conftest sys.path order so secubox-auth/api takes priority for auth tests Co-Authored-By: Claude Sonnet 4.6 --- packages/secubox-auth/api/main.py | 308 +++++++++++++++++- packages/secubox-auth/api/totp_pending.py | 51 +++ packages/secubox-auth/tests/conftest.py | 5 +- .../secubox-auth/tests/test_auth_flows.py | 120 +++++++ .../secubox-auth/tests/test_totp_pending.py | 32 ++ 5 files changed, 514 insertions(+), 2 deletions(-) create mode 100644 packages/secubox-auth/api/totp_pending.py create mode 100644 packages/secubox-auth/tests/test_auth_flows.py create mode 100644 packages/secubox-auth/tests/test_totp_pending.py diff --git a/packages/secubox-auth/api/main.py b/packages/secubox-auth/api/main.py index e700af02..293c68c8 100644 --- a/packages/secubox-auth/api/main.py +++ b/packages/secubox-auth/api/main.py @@ -4,6 +4,7 @@ from pydantic import BaseModel, Field, field_validator from secubox_core.auth import router as auth_router, require_jwt, create_token, set_session_callback from secubox_core.config import get_config import json +import os import secrets import time import threading @@ -27,12 +28,317 @@ async def health_check(): """Public health check endpoint for sidebar status.""" return {"status": "ok", "module": "deb"} +# ────────────────────────────────────────────────────────────────────── +# Task 13: branching login, MFA, TOTP enroll/confirm, set-password +# Inserted BEFORE the legacy auth_router mount so the overriding +# _login_router (mounted AFTER) takes precedence on /auth/login. +# ────────────────────────────────────────────────────────────────────── +from secubox_core import user_store +from secubox_core.auth import ( + set_session_validator, + _emit_session_event, + _decode_token as _jwt_decode, +) + +# Make secubox-users engine importable (in-process path, no IPC). +# We cannot use `from api import engine` because Python resolves "api" to the +# secubox-auth api package (the one we're inside). Instead we register the +# secubox-users api/ directory as a *separate* package named `_users_api` in +# sys.modules so that the relative imports inside engine.py (`from . import totp`) +# resolve correctly. +import importlib.util as _ilu +import sys as _sys +import types as _types + + +def _bootstrap_users_api_package() -> str: + """Register secubox-users/api as '_users_api' package; return its root dir.""" + candidates = [ + Path("/usr/lib/secubox/users/api"), + Path(__file__).resolve().parents[3] / "packages" / "secubox-users" / "api", + ] + pkg_root = next((p for p in candidates if (p / "engine.py").exists()), None) + if pkg_root is None: + raise ImportError("secubox-users api/ not found in any candidate path") + + pkg_name = "_users_api" + if pkg_name not in _sys.modules: + # Create a package object pointing at the discovered directory. + pkg = _types.ModuleType(pkg_name) + pkg.__path__ = [str(pkg_root)] + pkg.__package__ = pkg_name + pkg.__spec__ = _ilu.spec_from_file_location( + pkg_name, str(pkg_root / "__init__.py"), + submodule_search_locations=[str(pkg_root)], + ) + _sys.modules[pkg_name] = pkg + # Execute __init__.py if it exists. + init_py = pkg_root / "__init__.py" + if init_py.exists(): + spec = _ilu.spec_from_file_location(pkg_name, str(init_py), + submodule_search_locations=[str(pkg_root)]) + mod = _ilu.module_from_spec(spec) # type: ignore[arg-type] + _sys.modules[pkg_name] = mod + spec.loader.exec_module(mod) # type: ignore[union-attr] + + return pkg_name + + +def _load_users_submod(pkg_name: str, name: str) -> object: + full_name = f"{pkg_name}.{name}" + if full_name in _sys.modules: + return _sys.modules[full_name] + pkg_root = Path(_sys.modules[pkg_name].__path__[0]) + spec = _ilu.spec_from_file_location(full_name, str(pkg_root / f"{name}.py"), + submodule_search_locations=[]) + mod = _ilu.module_from_spec(spec) # type: ignore[arg-type] + mod.__package__ = pkg_name + _sys.modules[full_name] = mod + spec.loader.exec_module(mod) # type: ignore[union-attr] + return mod + + +_users_api_pkg = _bootstrap_users_api_package() +_users_engine_mod = _load_users_submod(_users_api_pkg, "engine") +_users_totp_mod = _load_users_submod(_users_api_pkg, "totp") +from api.totp_pending import PendingStore + +_DATA_DIR = Path(os.environ.get("SECUBOX_AUTH_DATA_DIR", "/var/lib/secubox/auth")) +_DATA_DIR.mkdir(parents=True, exist_ok=True) +_SESSIONS_FILE = Path(os.environ.get("SECUBOX_AUTH_SESSIONS", str(_DATA_DIR / "sessions.json"))) +_AUDIT_FILE = Path(os.environ.get("SECUBOX_AUTH_AUDIT", str(_DATA_DIR / "audit.log"))) +_TOTP_PENDING_FILE = Path(os.environ.get("SECUBOX_AUTH_TOTP_PENDING", str(_DATA_DIR / "totp-pending.json"))) +_USERS_FILE = Path(os.environ.get("USERS_FILE", "/etc/secubox/users.json")) + +_pending = PendingStore(_TOTP_PENDING_FILE, ttl_seconds=900) +_users_engine = _users_engine_mod.Engine(users_path=_USERS_FILE) + + +def _read_sessions() -> list: + if not _SESSIONS_FILE.exists(): + return [] + try: + return json.loads(_SESSIONS_FILE.read_text()) + except Exception: + return [] + + +def _write_sessions(rows: list) -> None: + _SESSIONS_FILE.write_text(json.dumps(rows)) + + +def _append_audit(event: str, username: str, details: dict) -> None: + line = json.dumps({"ts": time.time(), "event": event, "user": username, **details}) + "\n" + with _AUDIT_FILE.open("a") as _f: + _f.write(line) + + +def _session_validator(jti: str) -> bool: + return any(s.get("id") == jti for s in _read_sessions()) + + +def _on_session_event(event: str, username: str, details: dict) -> None: + _append_audit(event, username, details) + if event == "login_success": + rows = _read_sessions() + rows.append({ + "id": details.get("jti", secrets.token_hex(8)), + "username": username, + "ip": details.get("ip", ""), + "user_agent": details.get("user_agent", ""), + "created": datetime.utcnow().isoformat(), + "expires": int(time.time()) + details.get("expires_in", 86400), + "type": "jwt", + }) + _write_sessions(rows) + + +def _revoke_sessions(username: str) -> int: + rows = _read_sessions() + keep = [r for r in rows if r.get("username") != username] + n = len(rows) - len(keep) + _write_sessions(keep) + _append_audit("sessions_revoked", username, {"count": n}) + return n + + +set_session_validator(_session_validator) +# set_session_callback already called in startup(); _on_session_event replaces it. +set_session_callback(_on_session_event) +_users_engine.set_revoke_callback(_revoke_sessions) +_users_engine.set_audit_callback(lambda evt, user, d: _append_audit(evt, user, d)) + + +# ─── Branching login router ──────────────────────────────────────────── +from fastapi import APIRouter as _APIRouter, Request as _Request + +_login_router = _APIRouter(tags=["auth-v2"]) + + +class _LoginIn(BaseModel): + username: str + password: str + + +class _MfaIn(BaseModel): + code: str + + +class _SetPasswordIn(BaseModel): + new_password: str + old_password: Optional[str] = None + + +def _check_scope(authorization: Optional[str], expected_scope: str) -> dict: + if not authorization or not authorization.lower().startswith("bearer "): + raise HTTPException(status_code=401, detail="Token Bearer manquant") + payload = _jwt_decode(authorization.split(" ", 1)[1]) + if payload.get("scope") != expected_scope: + raise HTTPException(status_code=403, detail="Token hors scope") + return payload + + +@_login_router.post("/login") +async def _login_v2(req: _LoginIn, request: _Request): + """Branching login: setup_token / mfa_token / enrollment_token / access_token.""" + ip = (request.headers.get("X-Forwarded-For", "").split(",")[0].strip() + or request.headers.get("X-Real-IP", "") + or (request.client.host if request.client else "")) + ua = request.headers.get("User-Agent", "")[:100] + user = user_store.get_user(req.username) + + if not user or not user.get("enabled"): + _emit_session_event("login_failed", req.username, { + "reason": "unknown_user" if not user else "disabled", + "ip": ip, "user_agent": ua, + }) + raise HTTPException(status_code=401, detail="Identifiants incorrects") + + # Initial-setup branch (empty password + must_change_password flag) + if user.get("must_change_password") and req.password == "": + setup_tok = create_token(req.username, scope="set-password", expires_in=900) + _append_audit("setup_token_issued", req.username, {"ip": ip}) + return {"setup_required": True, "setup_token": setup_tok} + + if not user.get("password_hash"): + _emit_session_event("login_failed", req.username, {"reason": "no_local_password", "ip": ip}) + raise HTTPException(status_code=401, detail="Aucun mot de passe local") + + if not user_store.verify_password(req.username, req.password): + _emit_session_event("login_failed", req.username, {"reason": "invalid_credentials", "ip": ip}) + raise HTTPException(status_code=401, detail="Identifiants incorrects") + + totp_block = user.get("totp") or {} + if totp_block.get("enabled"): + mfa_tok = create_token(req.username, scope="mfa-challenge", expires_in=300) + _append_audit("mfa_challenge_issued", req.username, {"ip": ip}) + return {"mfa_required": True, "mfa_token": mfa_tok} + + # Admin without TOTP → force enrollment + if user.get("role") == "admin": + enroll_tok = create_token(req.username, scope="totp-enroll", expires_in=900) + _append_audit("totp_enrollment_required", req.username, {"ip": ip}) + return {"enrollment_required": True, "enrollment_token": enroll_tok} + + jti = secrets.token_hex(8) + tok = create_token(req.username, jti=jti) + _on_session_event("login_success", req.username, { + "jti": jti, "expires_in": 86400, "ip": ip, "user_agent": ua, + }) + _users_engine.touch_last_login(req.username) + return {"access_token": tok, "token_type": "bearer", "expires_in": 86400} + + +@_login_router.post("/login/mfa") +async def _login_mfa(req: _MfaIn, request: _Request): + payload = _check_scope(request.headers.get("Authorization"), "mfa-challenge") + username = payload["sub"] + if not user_store.is_enabled(username): + raise HTTPException(status_code=401, detail="Compte désactivé") + ok = (_users_engine.verify_totp_for_user(username, req.code) + or _users_engine.consume_backup_code(username, req.code)) + if not ok: + _append_audit("mfa_failed", username, {}) + raise HTTPException(status_code=401, detail="Code invalide") + jti = secrets.token_hex(8) + tok = create_token(username, jti=jti) + _on_session_event("login_success", username, {"jti": jti, "expires_in": 86400, "ip": ""}) + _users_engine.touch_last_login(username) + return {"access_token": tok, "token_type": "bearer", "expires_in": 86400} + + +@_login_router.post("/totp/enroll") +async def _totp_enroll(request: _Request): + payload = _check_scope(request.headers.get("Authorization"), "totp-enroll") + username = payload["sub"] + existing = user_store.get_user(username) or {} + if (existing.get("totp") or {}).get("enabled"): + raise HTTPException(status_code=409, detail="Déjà enrôlé") + secret = _users_totp_mod.generate_secret() + _pending.put(payload["jti"], secret) + import socket as _socket + issuer = f"SecuBox · {_socket.gethostname()}" + uri = _users_totp_mod.provisioning_uri(username, secret, issuer=issuer) + return {"secret": secret, "otpauth_uri": uri, "qr_png_b64": ""} + + +@_login_router.post("/totp/confirm") +async def _totp_confirm(req: _MfaIn, request: _Request): + payload = _check_scope(request.headers.get("Authorization"), "totp-enroll") + username = payload["sub"] + secret = _pending.get(payload["jti"]) + if not secret: + raise HTTPException(status_code=410, detail="Enrôlement expiré") + import pyotp as _pyotp + if not _pyotp.TOTP(secret).verify(req.code, valid_window=1): + raise HTTPException(status_code=401, detail="Code invalide") + backup_plain = _users_engine.enroll_totp(username, secret) + _pending.delete(payload["jti"]) + jti = secrets.token_hex(8) + tok = create_token(username, jti=jti) + _on_session_event("login_success", username, {"jti": jti, "expires_in": 86400, "ip": ""}) + return { + "access_token": tok, "token_type": "bearer", "expires_in": 86400, + "backup_codes": backup_plain, + "backup_codes_note": "Conservez ces codes. Affichés une seule fois.", + } + + +@_login_router.post("/set-password") +async def _set_password(req: _SetPasswordIn, request: _Request): + auth_h = request.headers.get("Authorization", "") + if not auth_h.lower().startswith("bearer "): + raise HTTPException(status_code=401, detail="Token Bearer manquant") + payload = _jwt_decode(auth_h.split(" ", 1)[1]) + scope = payload.get("scope") + username = payload["sub"] + if scope == "set-password": + _users_engine.set_password(username, req.new_password) + _users_engine.revoke_sessions(username) + return {"ok": True, "message": "Mot de passe défini, veuillez vous reconnecter"} + if scope is None: + # Change-my-password with full JWT + if not req.old_password: + raise HTTPException(status_code=400, detail="Ancien mot de passe requis") + if not user_store.verify_password(username, req.old_password): + raise HTTPException(status_code=401, detail="Ancien mot de passe incorrect") + _users_engine.set_password(username, req.new_password) + return {"ok": True, "message": "Mot de passe modifié"} + raise HTTPException(status_code=403, detail="Token hors scope") + + +# Mount v2 router FIRST so its /login overrides the legacy auth_router /login. +# FastAPI uses the first matching route, so _login_router must come before auth_router. +app.include_router(_login_router, prefix="/auth") app.include_router(auth_router, prefix="/auth") router = APIRouter() # Configuration DATA_DIR = Path("/var/lib/secubox/auth") -DATA_DIR.mkdir(parents=True, exist_ok=True) +try: + DATA_DIR.mkdir(parents=True, exist_ok=True) +except PermissionError: + pass # dev/test environment without root — DATA_DIR fallback is unused VOUCHERS_FILE = DATA_DIR / "vouchers.json" SESSIONS_FILE = DATA_DIR / "sessions.json" HISTORY_FILE = DATA_DIR / "history.json" diff --git a/packages/secubox-auth/api/totp_pending.py b/packages/secubox-auth/api/totp_pending.py new file mode 100644 index 00000000..2b74299c --- /dev/null +++ b/packages/secubox-auth/api/totp_pending.py @@ -0,0 +1,51 @@ +"""Pending-TOTP-enrollment store. JSON-backed, TTL'd.""" +from __future__ import annotations + +import json +import os +import tempfile +import time +from pathlib import Path +from typing import Optional + + +class PendingStore: + def __init__(self, path: Path, ttl_seconds: int = 900): + self.path = Path(path) + self.ttl = ttl_seconds + if not self.path.exists(): + self.path.parent.mkdir(parents=True, exist_ok=True) + self.path.write_text("{}") + + def _load(self) -> dict: + try: + return json.loads(self.path.read_text()) + except Exception: + return {} + + def _save(self, doc: dict) -> None: + fd, tmp = tempfile.mkstemp(prefix=".pending.", dir=str(self.path.parent)) + with os.fdopen(fd, "w") as f: + json.dump(doc, f) + os.replace(tmp, self.path) + + def put(self, key: str, secret: str) -> None: + doc = self._load() + doc[key] = {"secret": secret, "expires_at": int(time.time()) + self.ttl} + self._save(doc) + + def get(self, key: str) -> Optional[str]: + doc = self._load() + entry = doc.get(key) + if not entry: + return None + if entry.get("expires_at", 0) < int(time.time()): + self.delete(key) + return None + return entry.get("secret") + + def delete(self, key: str) -> None: + doc = self._load() + if key in doc: + del doc[key] + self._save(doc) diff --git a/packages/secubox-auth/tests/conftest.py b/packages/secubox-auth/tests/conftest.py index 0302cbdb..1273e6d5 100644 --- a/packages/secubox-auth/tests/conftest.py +++ b/packages/secubox-auth/tests/conftest.py @@ -7,8 +7,11 @@ import pytest ROOT = Path(__file__).resolve().parents[3] sys.path.insert(0, str(ROOT / "common")) -sys.path.insert(0, str(ROOT / "packages" / "secubox-auth")) +# secubox-users must be reachable for the engine/totp modules, but secubox-auth's +# own api/ package must take priority for `from api import …` in auth tests. +# Insert auth LAST so it lands at index 0 (highest priority). sys.path.insert(0, str(ROOT / "packages" / "secubox-users")) +sys.path.insert(0, str(ROOT / "packages" / "secubox-auth")) @pytest.fixture diff --git a/packages/secubox-auth/tests/test_auth_flows.py b/packages/secubox-auth/tests/test_auth_flows.py new file mode 100644 index 00000000..c7c8ab7b --- /dev/null +++ b/packages/secubox-auth/tests/test_auth_flows.py @@ -0,0 +1,120 @@ +"""End-to-end auth flows: password + TOTP + set-password + disable.""" +import json +from pathlib import Path + +import pyotp +import pytest +from argon2 import PasswordHasher +from fastapi.testclient import TestClient + + +@pytest.fixture +def client(tmp_path: Path, monkeypatch): + # Configure paths + users_path = tmp_path / "users.json" + sessions_path = tmp_path / "sessions.json" + audit_path = tmp_path / "audit.log" + pending_path = tmp_path / "totp-pending.json" + + # Pre-seed admin with password set, no TOTP yet (will trigger enrollment branch). + pw_hash = PasswordHasher().hash("GoodPass!42xyz") + users_path.write_text(json.dumps({ + "version": 2, + "users": [{ + "username": "admin", + "email": "a@b.c", + "role": "admin", + "enabled": True, + "password_hash": pw_hash, + "must_change_password": False, + "totp": None, + "google": None, + "services": [], + "created": "2026-05-13T00:00:00+00:00", + "last_login": None, + }], + "groups": [], + })) + sessions_path.write_text("[]") + pending_path.write_text("{}") + + monkeypatch.setenv("USERS_FILE", str(users_path)) + monkeypatch.setenv("SECUBOX_AUTH_DATA_DIR", str(tmp_path)) + monkeypatch.setenv("SECUBOX_AUTH_SESSIONS", str(sessions_path)) + monkeypatch.setenv("SECUBOX_AUTH_AUDIT", str(audit_path)) + monkeypatch.setenv("SECUBOX_AUTH_TOTP_PENDING", str(pending_path)) + monkeypatch.setenv("SECUBOX_JWT_SECRET", "test-secret") + + # Point user_store at the temp file + from secubox_core import user_store + monkeypatch.setattr(user_store, "USERS_PATH", users_path) + + # Re-import the app for a clean state + import importlib + from api import main as auth_main + importlib.reload(auth_main) + + return TestClient(auth_main.app), users_path, sessions_path + + +def test_admin_password_login_returns_enrollment_token(client): + c, users_path, _ = client + r = c.post("/auth/login", json={"username": "admin", "password": "GoodPass!42xyz"}) + assert r.status_code == 200 + body = r.json() + assert body.get("enrollment_required") is True + assert "enrollment_token" in body + + +def test_wrong_password_returns_401(client): + c, *_ = client + r = c.post("/auth/login", json={"username": "admin", "password": "wrong"}) + assert r.status_code == 401 + + +def test_disabled_user_blocked(client): + c, users_path, sessions_path = client + # disable admin + doc = json.loads(users_path.read_text()) + doc["users"][0]["enabled"] = False + users_path.write_text(json.dumps(doc)) + r = c.post("/auth/login", json={"username": "admin", "password": "GoodPass!42xyz"}) + assert r.status_code == 401 + + +def test_full_totp_enrollment_then_login(client): + c, users_path, sessions_path = client + + # 1. Login → enrollment_token + r1 = c.post("/auth/login", json={"username": "admin", "password": "GoodPass!42xyz"}) + enroll_tok = r1.json()["enrollment_token"] + + # 2. Enroll → secret + QR + r2 = c.post("/auth/totp/enroll", headers={"Authorization": f"Bearer {enroll_tok}"}) + assert r2.status_code == 200 + secret = r2.json()["secret"] + assert "otpauth_uri" in r2.json() + + # 3. Confirm with wrong code → 401, pending kept + r3 = c.post( + "/auth/totp/confirm", + json={"code": "000000"}, + headers={"Authorization": f"Bearer {enroll_tok}"}, + ) + assert r3.status_code == 401 + + # 4. Confirm with correct code → access_token + backup codes + code = pyotp.TOTP(secret).now() + r4 = c.post( + "/auth/totp/confirm", + json={"code": code}, + headers={"Authorization": f"Bearer {enroll_tok}"}, + ) + assert r4.status_code == 200 + assert "access_token" in r4.json() + assert len(r4.json()["backup_codes"]) == 10 + + # 5. Next login now gets mfa_required + r5 = c.post("/auth/login", json={"username": "admin", "password": "GoodPass!42xyz"}) + assert r5.status_code == 200 + assert r5.json()["mfa_required"] is True diff --git a/packages/secubox-auth/tests/test_totp_pending.py b/packages/secubox-auth/tests/test_totp_pending.py new file mode 100644 index 00000000..0d4fc18a --- /dev/null +++ b/packages/secubox-auth/tests/test_totp_pending.py @@ -0,0 +1,32 @@ +"""TOTP pending-enrollment store with TTL.""" +import time +from pathlib import Path + +from api import totp_pending + + +def test_put_get_roundtrip(tmp_path: Path): + store = totp_pending.PendingStore(tmp_path / "pending.json", ttl_seconds=600) + store.put("jti-abc", "SECRET123") + assert store.get("jti-abc") == "SECRET123" + + +def test_get_returns_none_after_ttl(tmp_path: Path, monkeypatch): + store = totp_pending.PendingStore(tmp_path / "pending.json", ttl_seconds=1) + store.put("jti-abc", "SECRET123") + # Fast-forward clock. + real_time = time.time + monkeypatch.setattr(totp_pending.time, "time", lambda: real_time() + 5) + assert store.get("jti-abc") is None + + +def test_delete_removes_entry(tmp_path: Path): + store = totp_pending.PendingStore(tmp_path / "pending.json", ttl_seconds=600) + store.put("jti-abc", "SECRET123") + store.delete("jti-abc") + assert store.get("jti-abc") is None + + +def test_get_returns_none_for_missing(tmp_path: Path): + store = totp_pending.PendingStore(tmp_path / "pending.json", ttl_seconds=600) + assert store.get("unknown") is None