diff --git a/packages/secubox-auth/api/main.py b/packages/secubox-auth/api/main.py index ab3b4200..501ac5dc 100644 --- a/packages/secubox-auth/api/main.py +++ b/packages/secubox-auth/api/main.py @@ -16,6 +16,7 @@ import subprocess from datetime import datetime, timedelta from pathlib import Path from typing import Dict, Any, List, Optional +from api import ntp_health app = FastAPI(title="secubox-auth", version="2.0.0", root_path="/api/v1/auth") @@ -173,6 +174,35 @@ _users_engine.set_revoke_callback(_revoke_sessions) _users_engine.set_audit_callback(lambda evt, user, d: _append_audit(evt, user, d)) +def _verify_totp_ntp_aware(username: str, code: str) -> bool: + """Verify TOTP with a window scaled by NTP health (offline-mode resilient).""" + u = user_store.get_user(username) or {} + block = (u.get("totp") or {}) + if not block.get("enabled"): + return False + secret = block.get("secret") + last_step = block.get("last_step") + window = ntp_health.recommended_totp_window() + step_size = 30 + now = int(time.time()) + current = now // step_size + import pyotp + for delta in range(-window, window + 1): + step = current + delta + if pyotp.TOTP(secret).at(step * step_size) == code: + if last_step is not None and step <= last_step: + return False + # Persist the consumed step + doc = json.loads(_USERS_FILE.read_text()) + for entry in doc["users"]: + if entry["username"] == username and entry.get("totp"): + entry["totp"]["last_step"] = step + break + _USERS_FILE.write_text(json.dumps(doc, indent=2, sort_keys=True)) + return True + return False + + # ─── Branching login router ──────────────────────────────────────────── from fastapi import APIRouter as _APIRouter, Request as _Request @@ -259,7 +289,7 @@ async def _login_mfa(req: _MfaIn, request: _Request): 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) + ok = (_verify_totp_ntp_aware(username, req.code) or _users_engine.consume_backup_code(username, req.code)) if not ok: _append_audit("mfa_failed", username, {}) @@ -283,7 +313,8 @@ async def _totp_enroll(request: _Request): 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": ""} + qr_png = _users_totp_mod.qr_png_b64(uri) + return {"secret": secret, "otpauth_uri": uri, "qr_png_b64": qr_png} @_login_router.post("/totp/confirm") @@ -331,6 +362,22 @@ async def _set_password(req: _SetPasswordIn, request: _Request): raise HTTPException(status_code=403, detail="Token hors scope") +@_login_router.get("/health") +async def _auth_health(): + """Public-readable health surface for the UI banner. + + Returns NTP-health + identity-store source. UI uses this to render warnings + like "Identity store in fallback mode" or "Clock not synced — TOTP may fail". + """ + src = user_store.load_with_fallback() + return { + "ntp": ntp_health.probe(), + "totp_window": ntp_health.recommended_totp_window(), + "identity_source": src.get("source"), + "identity_fallback": src.get("source") == "auth.toml.fallback", + } + + # 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. # Mounted under both prefixes: "" for the canonical URL (/api/v1/auth/login after nginx diff --git a/packages/secubox-auth/api/ntp_health.py b/packages/secubox-auth/api/ntp_health.py new file mode 100644 index 00000000..a2e35fbb --- /dev/null +++ b/packages/secubox-auth/api/ntp_health.py @@ -0,0 +1,70 @@ +"""NTP health probe — read chronyc tracking. Used to widen TOTP window on drift.""" +from __future__ import annotations + +import logging +import re +import subprocess +import time +from typing import Dict + +log = logging.getLogger("secubox.ntp_health") + +_OFFSET_RE = re.compile(r"^System time\s*:\s*([\d.eE+-]+)\s*seconds", re.MULTILINE) +_LEAP_RE = re.compile(r"^Leap status\s*:\s*(\S.*)$", re.MULTILINE) +_REF_RE = re.compile(r"^Reference ID\s*:\s*(\S+)", re.MULTILINE) + +_cache: Dict[str, object] = {"ts": 0, "result": None} +_CACHE_TTL = 30 # seconds + + +def probe() -> Dict[str, object]: + """Return {synced, drift_seconds, leap_status, reference_id} or {synced: False, error: ...}. + + Cached for _CACHE_TTL seconds so the auth path doesn't spawn chronyc on every call. + """ + now = time.time() + if now - _cache["ts"] < _CACHE_TTL and _cache["result"] is not None: + return dict(_cache["result"]) + result = _probe_uncached() + _cache.update({"ts": now, "result": result}) + return dict(result) + + +def _probe_uncached() -> Dict[str, object]: + try: + out = subprocess.run( + ["chronyc", "-n", "tracking"], + capture_output=True, text=True, timeout=2, check=False, + ) + except (FileNotFoundError, subprocess.TimeoutExpired) as exc: + return {"synced": False, "error": f"chronyc unavailable: {exc}"} + if out.returncode != 0: + return {"synced": False, "error": out.stderr.strip() or "chronyc failed"} + m_offset = _OFFSET_RE.search(out.stdout) + m_leap = _LEAP_RE.search(out.stdout) + m_ref = _REF_RE.search(out.stdout) + drift = float(m_offset.group(1)) if m_offset else None + leap = m_leap.group(1).strip() if m_leap else "Unknown" + ref = m_ref.group(1) if m_ref else "?" + synced = leap.lower().startswith("normal") and ref != "7F7F0101" # 7F7F0101 = unsynchronised + return { + "synced": synced, + "drift_seconds": drift, + "leap_status": leap, + "reference_id": ref, + } + + +def recommended_totp_window() -> int: + """Return the TOTP `valid_window` to use given current NTP health. + + synced → 1 (±30s) + degraded → 3 (±90s) — accept drift up to 90s + unknown → 2 (±60s) — middle ground + """ + h = probe() + if h.get("synced"): + return 1 + if "error" in h: + return 2 + return 3 diff --git a/packages/secubox-auth/tests/test_auth_flows.py b/packages/secubox-auth/tests/test_auth_flows.py index c7c8ab7b..ac3b1864 100644 --- a/packages/secubox-auth/tests/test_auth_flows.py +++ b/packages/secubox-auth/tests/test_auth_flows.py @@ -118,3 +118,44 @@ def test_full_totp_enrollment_then_login(client): r5 = c.post("/auth/login", json={"username": "admin", "password": "GoodPass!42xyz"}) assert r5.status_code == 200 assert r5.json()["mfa_required"] is True + + +def test_auth_health_endpoint(client, monkeypatch): + c, *_ = client + # Force unsynced + from api import ntp_health + monkeypatch.setattr(ntp_health, "probe", lambda: {"synced": False, "error": "no chronyc"}) + monkeypatch.setattr(ntp_health, "recommended_totp_window", lambda: 3) + r = c.get("/auth/health") + assert r.status_code == 200 + body = r.json() + assert body["ntp"]["synced"] is False + assert body["totp_window"] == 3 + assert body["identity_fallback"] in (True, False) + + +def test_totp_widened_window_accepts_drifted_code(client, monkeypatch): + """With NTP degraded (window=3), a code from a step ~60s in the past still validates.""" + import time as _time + import pyotp + c, users_path, _ = client + + # Enroll admin + r1 = c.post("/auth/login", json={"username": "admin", "password": "GoodPass!42xyz"}) + enroll_tok = r1.json()["enrollment_token"] + r2 = c.post("/auth/totp/enroll", headers={"Authorization": f"Bearer {enroll_tok}"}) + secret = r2.json()["secret"] + code = pyotp.TOTP(secret).now() + c.post("/auth/totp/confirm", json={"code": code}, + headers={"Authorization": f"Bearer {enroll_tok}"}) + + # Degrade NTP and present a code from 60s ago + from api import ntp_health + monkeypatch.setattr(ntp_health, "recommended_totp_window", lambda: 3) + r3 = c.post("/auth/login", json={"username": "admin", "password": "GoodPass!42xyz"}) + mfa_tok = r3.json()["mfa_token"] + drifted_code = pyotp.TOTP(secret).at(int(_time.time()) - 60) + r4 = c.post("/auth/login/mfa", json={"code": drifted_code}, + headers={"Authorization": f"Bearer {mfa_tok}"}) + assert r4.status_code == 200 + assert "access_token" in r4.json() diff --git a/packages/secubox-auth/tests/test_ntp_health.py b/packages/secubox-auth/tests/test_ntp_health.py new file mode 100644 index 00000000..0c10a590 --- /dev/null +++ b/packages/secubox-auth/tests/test_ntp_health.py @@ -0,0 +1,96 @@ +"""ntp_health: parsing, caching, fallback when chronyc missing.""" +from pathlib import Path + +import pytest + +from api import ntp_health + + +SAMPLE_SYNCED = """\ +Reference ID : 8A1B0C2D (ntp.example.org) +Stratum : 3 +Ref time (UTC) : Wed May 13 04:30:00 2026 +System time : 0.000123456 seconds fast of NTP time +Last offset : -0.000098 seconds +RMS offset : 0.000123 seconds +Frequency : 12.345 ppm slow +Residual freq : 0.001 ppm +Skew : 1.234 ppm +Root delay : 0.012345 seconds +Root dispersion : 0.001234 seconds +Update interval : 64.0 seconds +Leap status : Normal +""" + +SAMPLE_UNSYNCED = """\ +Reference ID : 7F7F0101 () +Stratum : 0 +Ref time (UTC) : Thu Jan 1 00:00:00 1970 +System time : 0.000000000 seconds fast of NTP time +Last offset : +0.000000000 seconds +RMS offset : 0.000000000 seconds +Frequency : 0.000 ppm slow +Residual freq : +0.000 ppm +Skew : 0.000 ppm +Root delay : 1.000000000 seconds +Root dispersion : 1.000000000 seconds +Update interval : 0.0 seconds +Leap status : Not synchronised +""" + + +@pytest.fixture(autouse=True) +def _bust_cache(): + ntp_health._cache.update({"ts": 0, "result": None}) + + +def _fake_chronyc(stdout: str, returncode: int = 0): + class _Proc: + def __init__(self): + self.stdout = stdout + self.stderr = "" + self.returncode = returncode + def runner(*a, **kw): + return _Proc() + return runner + + +def test_synced(monkeypatch): + monkeypatch.setattr(ntp_health.subprocess, "run", _fake_chronyc(SAMPLE_SYNCED)) + h = ntp_health.probe() + assert h["synced"] is True + assert h["leap_status"].lower().startswith("normal") + assert ntp_health.recommended_totp_window() == 1 + + +def test_unsynced(monkeypatch): + monkeypatch.setattr(ntp_health.subprocess, "run", _fake_chronyc(SAMPLE_UNSYNCED)) + h = ntp_health.probe() + assert h["synced"] is False + assert ntp_health.recommended_totp_window() == 3 + + +def test_chronyc_missing_returns_unknown(monkeypatch): + def raise_fnf(*a, **kw): + raise FileNotFoundError("no chronyc") + monkeypatch.setattr(ntp_health.subprocess, "run", raise_fnf) + h = ntp_health.probe() + assert h["synced"] is False + assert "error" in h + assert ntp_health.recommended_totp_window() == 2 # unknown → middle + + +def test_cache_is_honoured(monkeypatch): + calls = [] + def runner(*a, **kw): + calls.append(1) + class _Proc: + stdout = SAMPLE_SYNCED + stderr = "" + returncode = 0 + return _Proc() + monkeypatch.setattr(ntp_health.subprocess, "run", runner) + ntp_health.probe() + ntp_health.probe() + ntp_health.probe() + assert len(calls) == 1 # cached diff --git a/packages/secubox-users/api/totp.py b/packages/secubox-users/api/totp.py index a654a12a..bc57f40a 100644 --- a/packages/secubox-users/api/totp.py +++ b/packages/secubox-users/api/totp.py @@ -1,6 +1,8 @@ """RFC 6238 TOTP wrapper + backup-code generator (argon2id-hashed).""" from __future__ import annotations +import base64 +import io import secrets from typing import List, Optional, Tuple @@ -65,3 +67,22 @@ def verify_backup_code(hash_: str, code: str) -> bool: return False except Exception: return False + + +import qrcode # noqa: E402 — placed after stdlib/third-party block intentionally + + +def qr_png_b64(otpauth_uri: str, *, box_size: int = 6, border: int = 2) -> str: + """Render the otpauth:// URI as a base64-encoded PNG (no external service).""" + qr = qrcode.QRCode( + version=None, + error_correction=qrcode.constants.ERROR_CORRECT_M, + box_size=box_size, + border=border, + ) + qr.add_data(otpauth_uri) + qr.make(fit=True) + img = qr.make_image(fill_color="black", back_color="white") + buf = io.BytesIO() + img.save(buf, format="PNG") + return base64.b64encode(buf.getvalue()).decode("ascii") diff --git a/packages/secubox-users/tests/test_totp.py b/packages/secubox-users/tests/test_totp.py index a46cdca1..8cf486a7 100644 --- a/packages/secubox-users/tests/test_totp.py +++ b/packages/secubox-users/tests/test_totp.py @@ -61,3 +61,28 @@ def test_hash_and_verify_backup_code(): assert h.startswith("$argon2id$") assert totp.verify_backup_code(h, "ABCDEFGHIJ") is True assert totp.verify_backup_code(h, "OTHERCODE!") is False + + +def test_qr_png_b64_returns_valid_png_base64(): + uri = totp.provisioning_uri("alice", "JBSWY3DPEHPK3PXP", issuer="SecuBox") + b64 = totp.qr_png_b64(uri) + import base64 + raw = base64.b64decode(b64) + # PNG signature: 89 50 4E 47 0D 0A 1A 0A + assert raw[:8] == b"\x89PNG\r\n\x1a\n" + assert len(raw) > 100 # non-trivial payload + + +def test_qr_png_b64_is_self_contained(monkeypatch): + """Render must not require any network call — verified by failing socket.""" + import socket + real_socket = socket.socket + def boom(*a, **kw): + raise OSError("network disabled") + monkeypatch.setattr(socket, "socket", boom) + try: + uri = totp.provisioning_uri("alice", "JBSWY3DPEHPK3PXP") + b64 = totp.qr_png_b64(uri) + assert b64 + finally: + monkeypatch.setattr(socket, "socket", real_socket)