diff --git a/src/Server.py b/src/Server.py index 3950531..3678660 100644 --- a/src/Server.py +++ b/src/Server.py @@ -10,6 +10,8 @@ import asyncio import secrets import hashlib import hmac +import base64 +import urllib.parse import time import html import warnings @@ -276,6 +278,89 @@ def change_user_password(username: str, new_password: str, db_path: str = DEFAUL return changed +# --- Two-Factor Authentication (TOTP / RFC 6238) --- + +def generate_totp_secret(length: int = 20) -> str: + """Generates a cryptographically random Base32-encoded TOTP secret key.""" + return base64.b32encode(secrets.token_bytes(length)).decode("utf-8").replace("=", "") + + +def get_totp_token(secret: str, intervals_no: Optional[int] = None) -> str: + """Calculates a 6-digit TOTP code for the given counter interval according to RFC 6238.""" + if intervals_no is None: + intervals_no = int(time.time()) // 30 + key = base64.b32decode(secret + "=" * ((8 - len(secret) % 8) % 8), casefold=True) + msg = struct.pack(">Q", intervals_no) + h = hmac.new(key, msg, hashlib.sha1).digest() + o = h[19] & 15 + h_int = (struct.unpack(">I", h[o:o+4])[0] & 0x7fffffff) % 1000000 + return f"{h_int:06d}" + + +def verify_totp_code(secret: str, code: str, window: int = 1) -> bool: + """Verifies a 6-digit TOTP code with standard drift window (+/- 30s).""" + if not secret or not code: + return False + code = str(code).strip().replace(" ", "") + if len(code) != 6 or not code.isdigit(): + return False + now = int(time.time()) + current_step = now // 30 + for offset in range(-window, window + 1): + if hmac.compare_digest(get_totp_token(secret, current_step + offset), code): + return True + return False + + +def get_totp_uri(username: str, secret: str, issuer: str = "LOGAR Server Hub") -> str: + """Generates standard otpauth URI for QR code generation.""" + user_enc = urllib.parse.quote(username) + issuer_enc = urllib.parse.quote(issuer) + return f"otpauth://totp/{issuer_enc}:{user_enc}?secret={secret}&issuer={issuer_enc}&algorithm=SHA1&digits=6&period=30" + + +def get_user_totp_info(username: str, db_path: str = DEFAULT_DB_FILE) -> Dict[str, Any]: + """Retrieves TOTP configuration for a user from the SQLite database.""" + conn = sqlite3.connect(db_path) + c = conn.cursor() + try: + c.execute("SELECT totp_secret, totp_enabled FROM users WHERE username = ?", (username,)) + row = c.fetchone() + if row: + return {"secret": row[0] or "", "enabled": bool(row[1])} + except Exception: + pass + finally: + conn.close() + return {"secret": "", "enabled": False} + + +def enable_user_totp(username: str, secret: str, db_path: str = DEFAULT_DB_FILE) -> bool: + """Enables and activates TOTP for a user in the SQLite database.""" + conn = sqlite3.connect(db_path) + c = conn.cursor() + try: + c.execute("UPDATE users SET totp_secret = ?, totp_enabled = 1 WHERE username = ?", (secret, username)) + changed = c.rowcount > 0 + conn.commit() + return changed + finally: + conn.close() + + +def disable_user_totp(username: str, db_path: str = DEFAULT_DB_FILE) -> bool: + """Disables TOTP for a user in the SQLite database.""" + conn = sqlite3.connect(db_path) + c = conn.cursor() + try: + c.execute("UPDATE users SET totp_secret = NULL, totp_enabled = 0 WHERE username = ?", (username,)) + changed = c.rowcount > 0 + conn.commit() + return changed + finally: + conn.close() + + # --- Hermes Token Management --- def create_hermes_token(name: str, token: Optional[str] = None, db_path: str = DEFAULT_DB_FILE) -> Dict[str, Any]: @@ -765,10 +850,22 @@ def init_db(db_path: str, enrollment_secret: Optional[str] = None, max_seats: in password_hash TEXT NOT NULL, salt TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'admin', - created_at TEXT NOT NULL + created_at TEXT NOT NULL, + totp_secret TEXT, + totp_enabled INTEGER DEFAULT 0 ) """) + # Migrations for users table (2FA) + for col, col_def in [ + ("totp_secret", "TEXT"), + ("totp_enabled", "INTEGER DEFAULT 0") + ]: + try: + conn.execute(f"ALTER TABLE users ADD COLUMN {col} {col_def}") + except sqlite3.OperationalError: + pass + # Ensure default admin user exists c = conn.cursor() c.execute("SELECT COUNT(*) FROM users") @@ -1342,14 +1439,19 @@ def login_page(request: Request):
-
+
-
+
+
@@ -1358,26 +1460,37 @@ def login_page(request: Request): e.preventDefault(); const u = document.getElementById('username').value.trim(); const p = document.getElementById('password').value; + const totp = document.getElementById('totpCode').value.trim(); const errEl = document.getElementById('loginError'); const btn = document.getElementById('submitBtn'); + const totpGrp = document.getElementById('totpGroup'); errEl.style.display = 'none'; btn.disabled = true; btn.textContent = 'Authenticating...'; + const payload = { username: u, password: p }; + if (totp) payload.totp_code = totp; + try { const res = await fetch('/login', { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ username: u, password: p }) + body: JSON.stringify(payload) }); const data = await res.json(); if (res.ok && data.status === 'success') { window.location.href = '/'; + } else if (data.requires_2fa || data.status === '2fa_required') { + totpGrp.style.display = 'block'; + document.getElementById('totpCode').focus(); + btn.disabled = false; + btn.textContent = 'Verify & Sign In'; + errEl.style.display = 'none'; } else { errEl.textContent = data.detail || 'Invalid username or password'; errEl.style.display = 'block'; btn.disabled = false; - btn.textContent = 'Sign In'; + btn.textContent = totpGrp.style.display === 'block' ? 'Verify & Sign In' : 'Sign In'; } } catch(ex) { errEl.textContent = 'Connection error: ' + ex.message; @@ -1393,24 +1506,26 @@ def login_page(request: Request): @app.post("/login") async def process_login(request: Request): - """Processes login credentials and issues session cookie.""" + """Processes login credentials and issues session cookie with 2FA TOTP enforcement.""" client_ip = get_client_ip(request) if not check_ip_rate_limit(client_ip, _LOGIN_FAILURES, max_attempts=10, window_seconds=60): raise HTTPException(status_code=429, detail="Too many failed login attempts. Please try again later.") - username, password = None, None + username, password, totp_code = None, None, None content_type = request.headers.get("content-type", "") if "application/json" in content_type: try: body = await request.json() username = body.get("username") password = body.get("password") + totp_code = body.get("totp_code") except Exception: pass else: form = await request.form() username = form.get("username") password = form.get("password") + totp_code = form.get("totp_code") if not username or not password: raise HTTPException(status_code=400, detail="Username and password are required") @@ -1421,6 +1536,20 @@ async def process_login(request: Request): await asyncio.sleep(0.5) raise HTTPException(status_code=401, detail="Invalid username or password") + # Check 2FA TOTP requirement + totp_info = get_user_totp_info(username, db_path=db_path) + if totp_info.get("enabled"): + if not totp_code: + return JSONResponse(status_code=200, content={ + "status": "2fa_required", + "requires_2fa": True, + "message": "Two-factor authentication code required" + }) + if not verify_totp_code(totp_info["secret"], totp_code): + record_failed_ip_attempt(client_ip, _LOGIN_FAILURES) + await asyncio.sleep(0.3) + raise HTTPException(status_code=401, detail="Invalid 6-digit 2FA verification code") + reset_failed_ip_attempts(client_ip, _LOGIN_FAILURES) token = generate_session_token(username) is_https = request.url.scheme == "https" or request.headers.get("x-forwarded-proto") == "https" @@ -1456,6 +1585,55 @@ async def api_change_password(req: ChangePasswordRequest, current_user: str = De return {"status": "success", "message": "Password updated successfully"} +# --- 2FA Authentication Management Endpoints --- + +@app.get("/api/user/2fa/status") +def api_user_2fa_status(auth_user: str = Depends(require_admin_auth)): + """Returns 2FA status for currently logged in administrator.""" + db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE) + info = get_user_totp_info(auth_user, db_path=db_path) + return {"enabled": info["enabled"], "username": auth_user} + + +@app.get("/api/user/2fa/setup") +def api_user_2fa_setup(auth_user: str = Depends(require_admin_auth)): + """Generates a new TOTP secret and otpauth URI for 2FA provisioning.""" + secret = generate_totp_secret() + otpauth = get_totp_uri(auth_user, secret, issuer="LOGAR Server Hub") + return { + "secret": secret, + "otpauth_uri": otpauth, + "username": auth_user + } + + +@app.post("/api/user/2fa/enable") +async def api_user_2fa_enable(request: Request, auth_user: str = Depends(require_admin_auth)): + """Enables TOTP for the current user after validating 6-digit code.""" + body = await request.json() + secret = body.get("secret", "").strip() + code = body.get("code", "").strip() + if not secret or not code: + raise HTTPException(status_code=400, detail="Secret and 6-digit verification code required") + if not verify_totp_code(secret, code): + raise HTTPException(status_code=400, detail="Invalid 6-digit TOTP verification code. Please check your device clock.") + db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE) + enable_user_totp(auth_user, secret, db_path=db_path) + return {"status": "success", "message": "2FA successfully activated"} + + +@app.post("/api/user/2fa/disable") +async def api_user_2fa_disable(request: Request, auth_user: str = Depends(require_admin_auth)): + """Disables TOTP for the current user after verifying password.""" + body = await request.json() + password = body.get("password", "") + db_path = SERVER_STATE.get("config", {}).get("db_path", DEFAULT_DB_FILE) + if not verify_user_credentials(auth_user, password, db_path=db_path): + raise HTTPException(status_code=403, detail="Current password incorrect") + disable_user_totp(auth_user, db_path=db_path) + return {"status": "success", "message": "2FA successfully disabled"} + + @app.post("/api/client/enroll") def enroll_client(req: ClientEnrollRequest, request: Request = None): """ @@ -2303,7 +2481,7 @@ def serve_webui(request: Request = None): border-radius: 10px; box-shadow: 0 12px 30px rgba(0, 0, 0, 0.6); min-width: 200px; z-index: 200; overflow: hidden; }} - .user-menu-container:hover .user-dropdown, .user-menu-container:focus-within .user-dropdown {{ display: block; }} + .user-menu-container:hover .user-dropdown, .user-menu-container:focus-within .user-dropdown, .user-dropdown.show {{ display: block; }} .user-dropdown-header {{ padding: 10px 14px; font-size: 11px; font-weight: 700; text-transform: uppercase; color: var(--text-muted); background: #0f172a; border-bottom: 1px solid var(--border); }} .user-dropdown-item {{ display: flex; align-items: center; gap: 8px; padding: 10px 14px; @@ -2442,13 +2620,14 @@ def serve_webui(request: Request = None):
-
+
👤 {auth_user} ▼
-
+
Logged in as {auth_user}
- + + 🚪 Logout
@@ -2583,7 +2762,7 @@ def serve_webui(request: Request = None):
- Verified Anomaly Stream (Rolling 12h Window) + Verified Anomaly Stream (Latest 50 Events • Rolling 12h Window) Direct Hermes Intelligence
@@ -2732,7 +2911,92 @@ def serve_webui(request: Request = None):
+ + + """ diff --git a/tests/test_server.py b/tests/test_server.py index 3b9a45a..a22cf13 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -470,6 +470,46 @@ class TestServerComponent(unittest.TestCase): self.assertEqual(client_data["active_seats"], 0) self.assertEqual(len(client_data["clients"]), 1) + def test_two_factor_auth_and_totp_flow(self): + """Tests TOTP secret generation, token verification, URI generation, and enable/disable lifecycle.""" + Server.init_db(self.test_db) + + # 1. Test TOTP helpers + secret = Server.generate_totp_secret() + self.assertTrue(len(secret) >= 16) + token = Server.get_totp_token(secret) + self.assertEqual(len(token), 6) + self.assertTrue(token.isdigit()) + self.assertTrue(Server.verify_totp_code(secret, token)) + self.assertFalse(Server.verify_totp_code(secret, "000000" if token != "000000" else "999999")) + self.assertFalse(Server.verify_totp_code(secret, "invalid")) + + # 2. Test OTPAuth URI + uri = Server.get_totp_uri("admin", secret, issuer="LOGAR Fleet Hub") + self.assertTrue(uri.startswith("otpauth://totp/")) + self.assertIn("secret=" + secret, uri) + self.assertIn("LOGAR", uri) + + # 3. Test initial user 2FA status + info = Server.get_user_totp_info("admin", db_path=self.test_db) + self.assertFalse(info["enabled"]) + self.assertEqual(info["secret"], "") + + # 4. Test enabling 2FA + enabled = Server.enable_user_totp("admin", secret, db_path=self.test_db) + self.assertTrue(enabled) + + info_active = Server.get_user_totp_info("admin", db_path=self.test_db) + self.assertTrue(info_active["enabled"]) + self.assertEqual(info_active["secret"], secret) + + # 5. Test disabling 2FA + disabled = Server.disable_user_totp("admin", db_path=self.test_db) + self.assertTrue(disabled) + + info_disabled = Server.get_user_totp_info("admin", db_path=self.test_db) + self.assertFalse(info_disabled["enabled"]) + if __name__ == "__main__": unittest.main()