feat(auth): offline-mode hardening — server-side QR + NTP-aware TOTP window + /auth/health (ref #120)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

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

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

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()

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

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")

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)