diff --git a/src/password_reset.py b/src/password_reset.py new file mode 100644 index 0000000..ac80e54 --- /dev/null +++ b/src/password_reset.py @@ -0,0 +1,59 @@ +#!/usr/bin/env python3 +"""Interactively reset an application's user password.""" +import getpass +import sys + +from webssh_backend.auth import hash_password +from webssh_backend.database import creds_manager + + +MIN_PASSWORD_LENGTH = 12 + + +def main() -> int: + if not sys.stdin.isatty() or not sys.stdout.isatty(): + print("password-reset requires an interactive terminal.", file=sys.stderr) + return 2 + if len(sys.argv) > 2: + print("Usage: password-reset [username]", file=sys.stderr) + return 2 + + username = sys.argv[1] if len(sys.argv) == 2 else input("Username: ").strip() + if not username: + print("Username is required.", file=sys.stderr) + return 2 + + try: + user = creds_manager.get_user(username) + if not user: + print(f'User "{username}" does not exist.', file=sys.stderr) + return 1 + + password = getpass.getpass("New password (minimum 12 characters): ") + if len(password) < MIN_PASSWORD_LENGTH: + print("Password must be at least 12 characters.", file=sys.stderr) + return 2 + if len(password.encode("utf-8")) > 1024: + print("Password must not exceed 1024 UTF-8 bytes.", file=sys.stderr) + return 2 + confirmation = getpass.getpass("Confirm new password: ") + if password != confirmation: + print("Passwords do not match.", file=sys.stderr) + return 2 + + if not creds_manager.reset_user_password(username, hash_password(password)): + print(f'Failed to reset password for "{username}".', file=sys.stderr) + return 1 + + print(f'Password reset for "{username}". They must change it at next login.') + return 0 + except (EOFError, KeyboardInterrupt): + print("\nPassword reset cancelled.", file=sys.stderr) + return 130 + except Exception as error: + print(f"Password reset failed: {error}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/run.py b/src/run.py index bcaa041..301f2e1 100644 --- a/src/run.py +++ b/src/run.py @@ -12,7 +12,9 @@ Usage: python run.py Environment variables: - WEBSSH_USERS - Comma-separated user:pass pairs (default: admin:admin) + WEBSSH_USERS - Optional one-time import of comma-separated user:pass pairs. + Imported users must change their password at first login. + If unset and the database has no users, create the first admin in the browser. JWT_SECRET - Secret key for JWT tokens (auto-generated if not set) MAX_UPLOAD_SIZE - Maximum file upload size (default: 100M) RECORDING_ENABLED - Enable session recording (default: true) diff --git a/src/static/webssh.css b/src/static/webssh.css index 4d3857e..89088fe 100644 --- a/src/static/webssh.css +++ b/src/static/webssh.css @@ -19,6 +19,24 @@ body { flex-direction: column; } +.app-version { + padding: 9px 15px; + color: #9ca3af; + font-size: 12px; + background: #111827; + border-bottom: 1px solid #1f2937; +} + +.version-update-notice { + display: block; + margin-top: 3px; + color: #fbbf24; +} + +.version-update-notice[hidden] { + display: none; +} + .sidebar-header { padding: 15px; background: #1f2937; @@ -45,6 +63,50 @@ body { .btn-add:hover { background: #059669; } +.btn-user-management { + width: 100%; + margin-top: 8px; + padding: 7px; + background: #374151; + color: #e5e7eb; + border: 1px solid #4b5563; + border-radius: 6px; + cursor: pointer; + font-size: 12px; +} + +.btn-user-management:hover { background: #4b5563; } + +.users-list { + margin-bottom: 18px; +} + +.user-list-item { + display: flex; + justify-content: space-between; + align-items: center; + gap: 12px; + padding: 9px 0; + border-bottom: 1px solid #374151; +} + +.user-list-details { + min-width: 0; + overflow-wrap: anywhere; +} + +.user-list-role { + margin-left: 6px; + color: #9ca3af; + font-size: 12px; +} + +.user-management-error { + color: #f87171; + font-size: 13px; + margin-bottom: 10px; +} + .sidebar-actions { display: flex; gap: 5px; diff --git a/src/static/webssh.js b/src/static/webssh.js index 5d07896..7f4c0e0 100644 --- a/src/static/webssh.js +++ b/src/static/webssh.js @@ -4,13 +4,16 @@ let connections = []; let currentConnectionId = null; let authToken = localStorage.getItem('webssh_token'); let currentUser = localStorage.getItem('webssh_user'); -let storedUsers = {}; // Will be loaded from /api/config let maxUploadSize = localStorage.getItem('maxUploadSize') ? parseInt(localStorage.getItem('maxUploadSize')) : 100 * 1024 * 1024; // Dynamically determine API URL based on current host (same origin since everything is in container) let apiBaseUrl = window.location.origin; let backendWsPort = null; let backendWsHost = null; let hostStatusTimer = null; +let authScreenMode = 'login'; +let currentUserRole = null; +let runningVersion = null; +let availableLatestVersion = null; // Use the frontend proxy so the browser does not need direct access to the backend port. function getWsUrl() { @@ -19,6 +22,108 @@ function getWsUrl() { } const CONNECTIONS_KEY = 'webssh_connections'; +const LATEST_VERSION_URL = 'https://strixoss.github.io/strixoss/webssh/latest.json'; +const LATEST_VERSION_CACHE_KEY = 'webssh_latest_version_check'; +const VERSION_CHECK_INTERVAL = 24 * 60 * 60 * 1000; + +function parseVersion(version) { + if (typeof version !== 'string' || !/^\d+(?:\.\d+)*$/.test(version)) return null; + return version.split('.').map(Number); +} + +function isNewerVersion(latest, current) { + const latestParts = parseVersion(latest); + const currentParts = parseVersion(current); + if (!latestParts || !currentParts) return false; + + const partCount = Math.max(latestParts.length, currentParts.length); + for (let i = 0; i < partCount; i++) { + const latestPart = latestParts[i] || 0; + const currentPart = currentParts[i] || 0; + if (latestPart !== currentPart) return latestPart > currentPart; + } + return false; +} + +function showLatestVersion(latestVersion, currentVersion) { + runningVersion = currentVersion; + availableLatestVersion = latestVersion; + renderLatestVersionNotice(); +} + +function renderLatestVersionNotice() { + const notice = document.getElementById('versionUpdateNotice'); + if (!notice) return; + if (currentUserRole !== 'admin' || !isNewerVersion(availableLatestVersion, runningVersion)) { + notice.hidden = true; + notice.textContent = ''; + return; + } + + notice.textContent = `New version available: v${availableLatestVersion}`; + notice.hidden = false; +} + +async function loadVersionAndCheckForUpdates() { + const versionLabel = document.getElementById('appVersion'); + const notice = document.getElementById('versionUpdateNotice'); + + try { + const response = await fetch(`${apiBaseUrl}/api/version`); + if (!response.ok) throw new Error(`Version API returned HTTP ${response.status}`); + const data = await response.json(); + const currentVersion = data.version; + if (!parseVersion(currentVersion)) throw new Error('Version API returned an invalid version'); + + versionLabel.textContent = `WebSSH v${currentVersion}`; + + let cachedCheck = null; + try { + cachedCheck = JSON.parse(localStorage.getItem(LATEST_VERSION_CACHE_KEY) || 'null'); + } catch (error) { + console.warn('Unable to read cached latest version:', error); + } + + if ( + cachedCheck && + Number.isFinite(cachedCheck.checkedAt) && + Date.now() - cachedCheck.checkedAt >= 0 && + Date.now() - cachedCheck.checkedAt < VERSION_CHECK_INTERVAL && + parseVersion(cachedCheck.response?.stable?.version) + ) { + showLatestVersion(cachedCheck.response.stable.version, currentVersion); + return; + } + + try { + const latestResponse = await fetch(LATEST_VERSION_URL, { cache: 'no-store' }); + if (!latestResponse.ok) throw new Error(`Latest-version endpoint returned HTTP ${latestResponse.status}`); + const latestData = await latestResponse.json(); + const latestVersion = latestData?.stable?.version; + if (!parseVersion(latestVersion)) throw new Error('Latest-version endpoint returned an invalid version'); + + try { + localStorage.setItem(LATEST_VERSION_CACHE_KEY, JSON.stringify({ + checkedAt: Date.now(), + response: latestData + })); + } catch (error) { + console.warn('Unable to cache latest version response:', error); + } + + showLatestVersion(latestVersion, currentVersion); + } catch (error) { + console.warn('Failed to check for a new WebSSH version:', error); + if (parseVersion(cachedCheck?.response?.stable?.version)) { + showLatestVersion(cachedCheck.response.stable.version, currentVersion); + } + } + } catch (error) { + console.warn('Failed to load the current WebSSH version:', error); + if (versionLabel) versionLabel.textContent = 'WebSSH'; + if (notice) notice.hidden = true; + } +} // Load connections from server (with encrypted credentials) async function loadSavedConnections() { @@ -226,6 +331,7 @@ resizeObserver.observe(document.getElementById('terminal')); if (authToken) { verifyToken(); } +initializeAuthScreen(); // Load saved connections loadSavedConnections(); @@ -239,32 +345,69 @@ term.onData((data) => { }); async function login() { - const username = document.getElementById('loginUsername').value; + const username = document.getElementById('loginUsername').value.trim(); const password = document.getElementById('loginPassword').value; - const apiUrl = apiBaseUrl; - - if (!username || !password) { - document.getElementById('loginError').textContent = 'Please enter username and password'; + const confirmPassword = document.getElementById('confirmPassword').value; + + if ((authScreenMode !== 'change-password' && !username) || !password) { + document.getElementById('loginError').textContent = authScreenMode === 'change-password' + ? 'Please enter a new password' + : 'Please enter username and password'; return; } - + + if (authScreenMode !== 'login' && password !== confirmPassword) { + document.getElementById('loginError').textContent = 'Passwords do not match'; + return; + } + + if (authScreenMode !== 'login' && password.length < 12) { + document.getElementById('loginError').textContent = 'Password must be at least 12 characters'; + return; + } + try { - const response = await fetch(`${apiUrl}/api/login`, { + const endpoint = authScreenMode === 'setup' + ? '/api/setup' + : authScreenMode === 'change-password' + ? '/api/change-password' + : '/api/login'; + const headers = { 'Content-Type': 'application/json' }; + if (authScreenMode === 'change-password') { + headers.Authorization = `Bearer ${authToken}`; + } + + const response = await fetch(`${apiBaseUrl}${endpoint}`, { method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ username, password }) + headers, + body: JSON.stringify(authScreenMode === 'change-password' ? { password } : { username, password }) }); - const data = await response.json(); - - if (response.ok && data.token) { + + if (response.ok && authScreenMode === 'setup') { + setAuthScreenMode('login'); + await login(); + return; + } + + if (response.ok && authScreenMode === 'change-password' && data.token) { authToken = data.token; localStorage.setItem('webssh_token', authToken); - document.getElementById('loginModal').classList.remove('active'); - document.getElementById('loggedInUser').textContent = data.username; - term.writeln(`[Authenticated as ${data.username}]`); - loadSavedConnections(); - setTimeout(loadConfig, 500); // Load configuration after successful login with delay + finishLogin(data); + } else if (response.ok && data.token && data.password_change_required) { + authToken = data.token; + localStorage.setItem('webssh_token', authToken); + currentUser = data.username; + localStorage.setItem('webssh_user', currentUser); + setUserManagementAccess(null); + setAuthScreenMode('change-password'); + document.getElementById('loginPassword').value = ''; + document.getElementById('confirmPassword').value = ''; + document.getElementById('loginError').textContent = 'Please set a new password to continue.'; + } else if (response.ok && data.token) { + authToken = data.token; + localStorage.setItem('webssh_token', authToken); + finishLogin(data); } else { document.getElementById('loginError').textContent = data.error || 'Login failed'; } @@ -273,6 +416,165 @@ async function login() { } } +function finishLogin(data) { + currentUser = data.username; + localStorage.setItem('webssh_user', currentUser); + document.getElementById('loginModal').classList.remove('active'); + document.getElementById('loggedInUser').textContent = `${data.username} (${data.role})`; + setUserManagementAccess(data.role); + term.writeln(`[Authenticated as ${data.username}]`); + loadSavedConnections(); + setTimeout(loadConfig, 500); +} + +function setUserManagementAccess(role) { + currentUserRole = role; + document.getElementById('manageUsersBtn').hidden = role !== 'admin'; + renderLatestVersionNotice(); +} + +async function showUserManagement() { + document.getElementById('userManagementModal').classList.add('active'); + document.getElementById('userManagementError').textContent = ''; + await loadUsers(); +} + +function hideUserManagement() { + document.getElementById('userManagementModal').classList.remove('active'); +} + +async function loadUsers() { + const usersList = document.getElementById('usersList'); + usersList.textContent = 'Loading users...'; + try { + const response = await fetch(`${apiBaseUrl}/api/users`, { + headers: { 'Authorization': 'Bearer ' + authToken } + }); + const data = await response.json(); + if (!response.ok) throw new Error(data.error || 'Failed to load users'); + + usersList.replaceChildren(); + data.users.forEach(user => { + const row = document.createElement('div'); + row.className = 'user-list-item'; + const details = document.createElement('div'); + details.className = 'user-list-details'; + const name = document.createElement('strong'); + name.textContent = user.username; + const role = document.createElement('span'); + role.className = 'user-list-role'; + role.textContent = user.role; + details.append(name, role); + if (user.must_change_password) { + const changeNotice = document.createElement('div'); + changeNotice.className = 'user-list-role'; + changeNotice.textContent = 'Password change required at next login'; + details.append(changeNotice); + } + row.append(details); + + if (user.username !== currentUser) { + const deleteButton = document.createElement('button'); + deleteButton.className = 'btn-small btn-delete'; + deleteButton.textContent = 'Delete'; + deleteButton.addEventListener('click', () => deleteUser(user.username)); + row.append(deleteButton); + } + usersList.append(row); + }); + } catch (error) { + usersList.textContent = error.message; + } +} + +async function createUser() { + const errorDisplay = document.getElementById('userManagementError'); + errorDisplay.textContent = ''; + try { + const response = await fetch(`${apiBaseUrl}/api/users`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'Authorization': 'Bearer ' + authToken + }, + body: JSON.stringify({ + username: document.getElementById('newAppUsername').value.trim(), + password: document.getElementById('newAppPassword').value, + role: document.getElementById('newAppRole').value + }) + }); + const data = await response.json(); + if (!response.ok) throw new Error(data.error || 'Failed to create user'); + + document.getElementById('newAppUsername').value = ''; + document.getElementById('newAppPassword').value = ''; + errorDisplay.textContent = 'User created. They must change the temporary password at first login.'; + errorDisplay.style.color = '#34d399'; + await loadUsers(); + } catch (error) { + errorDisplay.style.color = '#f87171'; + errorDisplay.textContent = error.message; + } +} + +async function deleteUser(username) { + if (!window.confirm(`Delete user "${username}"?`)) return; + const errorDisplay = document.getElementById('userManagementError'); + errorDisplay.textContent = ''; + errorDisplay.style.color = '#f87171'; + try { + const response = await fetch(`${apiBaseUrl}/api/users/delete`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'Authorization': 'Bearer ' + authToken + }, + body: JSON.stringify({ username }) + }); + const data = await response.json(); + if (!response.ok) throw new Error(data.error || 'Failed to delete user'); + await loadUsers(); + } catch (error) { + errorDisplay.textContent = error.message; + } +} + +function setAuthScreenMode(mode) { + authScreenMode = mode; + const isLogin = mode === 'login'; + const isSetup = mode === 'setup'; + const title = document.getElementById('authTitle'); + const usernameGroup = document.getElementById('loginUsernameGroup'); + const passwordLabel = document.getElementById('loginPasswordLabel'); + const confirmGroup = document.getElementById('confirmPasswordGroup'); + const submitButton = document.getElementById('authSubmitButton'); + const username = document.getElementById('loginUsername'); + const password = document.getElementById('loginPassword'); + + title.textContent = isSetup ? 'Create Administrator' : isLogin ? 'WebSSH Login' : 'Change Password'; + usernameGroup.hidden = mode === 'change-password'; + passwordLabel.textContent = isLogin ? 'Password' : 'New password'; + confirmGroup.hidden = isLogin; + submitButton.textContent = isSetup ? 'Create administrator' : isLogin ? 'Login' : 'Change password'; + username.autocomplete = 'username'; + password.autocomplete = isLogin ? 'current-password' : 'new-password'; + password.placeholder = isLogin ? 'Password' : 'At least 12 characters'; + document.getElementById('loginError').textContent = ''; +} + +async function initializeAuthScreen() { + try { + const response = await fetch(`${apiBaseUrl}/api/setup-status`); + if (!response.ok) throw new Error(`Setup status returned HTTP ${response.status}`); + const data = await response.json(); + if (data.needs_setup && !authToken) { + setAuthScreenMode('setup'); + } + } catch (error) { + console.error('Unable to determine initial setup status:', error); + } +} + async function verifyToken() { try { const response = await fetch(`${apiBaseUrl}/api/verify`, { @@ -280,27 +582,37 @@ async function verifyToken() { headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ token: authToken }) }); - + const data = await response.json(); - - if (data.valid) { + + if (data.valid && data.password_change_required) { + currentUser = data.username; + setUserManagementAccess(null); + setAuthScreenMode('change-password'); + document.getElementById('loginModal').classList.add('active'); + document.getElementById('loginError').textContent = 'Please set a new password to continue.'; + } else if (data.valid) { document.getElementById('loginModal').classList.remove('active'); - document.getElementById('loggedInUser').textContent = data.username; + document.getElementById('loggedInUser').textContent = `${data.username} (${data.role})`; + setUserManagementAccess(data.role); term.writeln(`[Authenticated as ${data.username}]`); - loadSavedConnections(); // Load encrypted credentials from server - setTimeout(loadConfig, 500); // Load configuration after successful token verification with delay + loadSavedConnections(); + setTimeout(loadConfig, 500); } else { localStorage.removeItem('webssh_token'); authToken = null; + setUserManagementAccess(null); } } catch (e) { localStorage.removeItem('webssh_token'); authToken = null; + setUserManagementAccess(null); } } function logout() { localStorage.removeItem('webssh_token'); + localStorage.removeItem('webssh_user'); localStorage.removeItem('webssh_api_url'); authToken = null; location.reload(); @@ -816,6 +1128,7 @@ function sendResizeForConnection(conn) { // Initialize toggleAuthFields(); +loadVersionAndCheckForUpdates(); term.writeln('WebSSH Multi-Connection Client'); term.writeln('Click "+ New Connection" to get started'); @@ -862,11 +1175,6 @@ function loadConfig() { backendWsHost = config.config.ws_host || window.location.hostname || 'localhost'; console.log(`Loaded backend WebSocket port: ${backendWsPort}`); } - // Load stored users for lock screen verification - if (config.config.stored_users) { - storedUsers = config.config.stored_users; - console.log('Loaded stored users for lock screen'); - } // Load current user if (config.config.current_user) { currentUser = config.config.current_user; @@ -2505,52 +2813,33 @@ async function unlockSession() { return; } - // Hash the password and verify against stored users - const hashedInput = await hashPassword(password); - - // Try to find a matching user - let validUser = null; - for (const [username, storedHash] of Object.entries(storedUsers || {})) { - if (storedHash === hashedInput) { - validUser = username; - break; - } - } - - if (!validUser) { - // Also try current session - const response = await fetch(`${apiBaseUrl}/api/verify`, { + try { + const response = await fetch(`${apiBaseUrl}/api/verify-password`, { method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ token: authToken }) + headers: { + 'Content-Type': 'application/json', + 'Authorization': `Bearer ${authToken}` + }, + body: JSON.stringify({ password }) }); - - if (response.ok) { - validUser = currentUser; - } - } - - if (validUser) { - isLocked = false; - lastActivityTime = Date.now(); - document.getElementById('lockScreenModal').style.display = 'none'; - resetInactivityTimers(); - console.log('Session unlocked'); - } else { - document.getElementById('unlockError').textContent = 'Incorrect password'; - document.getElementById('unlockError').style.display = 'block'; - document.getElementById('unlockPassword').value = ''; - document.getElementById('unlockPassword').focus(); - } -} + const result = await response.json(); -// Hash password for verification -async function hashPassword(password) { - const encoder = new TextEncoder(); - const data = encoder.encode(password); - const hashBuffer = await crypto.subtle.digest('SHA-256', data); - const hashArray = Array.from(new Uint8Array(hashBuffer)); - return hashArray.map(b => b.toString(16).padStart(2, '0')).join(''); + if (response.ok && result.valid) { + isLocked = false; + lastActivityTime = Date.now(); + document.getElementById('lockScreenModal').style.display = 'none'; + resetInactivityTimers(); + console.log('Session unlocked'); + return; + } + } catch (error) { + console.error('Password verification failed:', error); + } + + document.getElementById('unlockError').textContent = 'Incorrect password'; + document.getElementById('unlockError').style.display = 'block'; + document.getElementById('unlockPassword').value = ''; + document.getElementById('unlockPassword').focus(); } // Auto-disconnect due to inactivity diff --git a/src/webssh_backend/__init__.py b/src/webssh_backend/__init__.py index dceadb6..0cd17a0 100644 --- a/src/webssh_backend/__init__.py +++ b/src/webssh_backend/__init__.py @@ -18,7 +18,7 @@ Usage: from webssh_backend.main import main asyncio.run(main()) """ -__version__ = "0.5.1" +__version__ = "0.6.0" # Note: Specific imports are intentionally NOT done at package level # to avoid loading all dependencies when only specific modules are needed. diff --git a/src/webssh_backend/auth.py b/src/webssh_backend/auth.py index 2b337a8..cbcf3b3 100644 --- a/src/webssh_backend/auth.py +++ b/src/webssh_backend/auth.py @@ -1,6 +1,10 @@ """ Auth module - JWT authentication and user management. """ +import base64 +import hashlib +import hmac +import os from datetime import datetime, timedelta, timezone from typing import Optional from urllib.parse import parse_qs, urlparse @@ -8,11 +12,49 @@ from urllib.parse import parse_qs, urlparse import jwt from websockets.server import WebSocketServerProtocol -from .config import JWT_ALGORITHM, JWT_EXPIRE_HOURS, JWT_SECRET, USERS, logger +from .config import JWT_ALGORITHM, JWT_EXPIRE_HOURS, JWT_SECRET, logger -__all__ = ['verify_token', 'verify_token_ws', 'verify_token_http', 'generate_token', 'JWT_ALGORITHM', 'JWT_EXPIRE_HOURS', 'JWT_SECRET', 'USERS'] +__all__ = [ + 'verify_token', 'verify_token_ws', 'verify_token_http', 'generate_token', + 'hash_password', 'verify_password', 'decode_token', + 'JWT_ALGORITHM', 'JWT_EXPIRE_HOURS', 'JWT_SECRET' +] from .utils import get_client_ip +PASSWORD_HASH_ITERATIONS = 600_000 + + +def hash_password(password: str) -> str: + """Hash a password using PBKDF2-HMAC-SHA256 with a unique random salt.""" + salt = os.urandom(16) + digest = hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'), salt, PASSWORD_HASH_ITERATIONS) + return 'pbkdf2_sha256${}${}${}'.format( + PASSWORD_HASH_ITERATIONS, + base64.urlsafe_b64encode(salt).decode('ascii'), + base64.urlsafe_b64encode(digest).decode('ascii') + ) + + +def verify_password(password: str, stored_hash: str) -> bool: + """Constant-time verify a password against a stored PBKDF2 hash.""" + try: + algorithm, iterations, salt, expected = stored_hash.split('$', 3) + iterations = int(iterations) + if algorithm != 'pbkdf2_sha256' or not 100_000 <= iterations <= 2_000_000: + return False + digest = hashlib.pbkdf2_hmac( + 'sha256', + password.encode('utf-8'), + base64.urlsafe_b64decode(salt.encode('ascii')), + iterations + ) + return hmac.compare_digest( + base64.urlsafe_b64encode(digest).decode('ascii'), + expected + ) + except (AttributeError, TypeError, ValueError): + return False + async def verify_token_ws(ws: WebSocketServerProtocol) -> Optional[str]: """Verify JWT token from WebSocket query parameters. Returns username or None.""" @@ -27,8 +69,19 @@ async def verify_token_ws(ws: WebSocketServerProtocol) -> Optional[str]: logger.warning(f"No token provided from {get_client_ip(ws)}") return None - payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM]) - return payload['sub'] + payload = decode_token(token) + if not payload or payload.get('pwd_change'): + return None + + from .database import creds_manager + user = creds_manager.get_user(payload.get('sub')) + if ( + not user + or user['must_change_password'] + or payload.get('auth_version', 0) != user['auth_version'] + ): + return None + return user['username'] except jwt.ExpiredSignatureError: logger.warning(f"Expired token from {get_client_ip(ws)}") return None @@ -40,29 +93,42 @@ async def verify_token_ws(ws: WebSocketServerProtocol) -> Optional[str]: return None -def generate_token(username: str) -> str: +def generate_token( + username: str, + password_change_required: bool = False, + auth_version: int = 0, +) -> str: """Generate a new JWT token for the user.""" expire = datetime.now(timezone.utc) + timedelta(hours=JWT_EXPIRE_HOURS) - token = jwt.encode( - {'sub': username, 'exp': expire, 'iat': datetime.now(timezone.utc)}, - JWT_SECRET, - algorithm=JWT_ALGORITHM - ) + payload = { + 'sub': username, + 'exp': expire, + 'iat': datetime.now(timezone.utc), + 'auth_version': auth_version, + } + if password_change_required: + payload['pwd_change'] = True + token = jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM) # Handle both PyJWT v1 (bytes) and v2 (str) if isinstance(token, bytes): token = token.decode('utf-8') return token -def verify_token_http(token: str) -> Optional[str]: - """Verify JWT token from HTTP request. Returns username or None.""" +def decode_token(token: str) -> Optional[dict]: + """Decode a JWT token, returning None when invalid or expired.""" try: - payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM]) - return payload.get('sub') - except jwt.ExpiredSignatureError: + return jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM]) + except (jwt.InvalidTokenError, TypeError): return None - except jwt.InvalidTokenError: + + +def verify_token_http(token: str, allow_password_change: bool = False) -> Optional[str]: + """Verify an HTTP JWT, rejecting password-change-only tokens by default.""" + payload = decode_token(token) + if not payload or (payload.get('pwd_change') and not allow_password_change): return None + return payload.get('sub') # For backward compatibility, keep the original function name diff --git a/src/webssh_backend/config.py b/src/webssh_backend/config.py index 0c3d32d..a79aeaf 100644 --- a/src/webssh_backend/config.py +++ b/src/webssh_backend/config.py @@ -1,7 +1,6 @@ """ Configuration module - all constants and environment variables. """ -import hashlib import json import logging import os @@ -201,16 +200,16 @@ except Exception as exc: logger.info(f"Resolved backend ports: ws={WS_PORT}, http={HTTP_PORT}, sftp={SFTP_WS_PORT}") -# User Configuration (username: password_hash) -# Set via environment variable: WEBSSH_USERS=user1:pass1,user2:pass2 -USERS = {} -users_env = os.environ.get("WEBSSH_USERS", "admin:admin") -for user_cred in users_env.split(","): +# Legacy user configuration is imported into the database once during startup. +# Format: WEBSSH_USERS=user1:pass1,user2:pass2 +legacy_users = {} +for user_cred in os.environ.get("WEBSSH_USERS", "").split(","): if ":" in user_cred: - u, p = user_cred.split(":", 1) - USERS[u.strip()] = hashlib.sha256(p.strip().encode()).hexdigest() - -logger.info(f"Loaded {len(USERS)} users: {list(USERS.keys())}") + username, password = user_cred.split(":", 1) + username, password = username.strip(), password.strip() + if username and password: + legacy_users[username] = password +LEGACY_USERS = list(legacy_users.items()) # Log storage directory status logger.info(f"Using storage directory: {STORAGE_DIR}") diff --git a/src/webssh_backend/database.py b/src/webssh_backend/database.py index cb25e59..141b0de 100644 --- a/src/webssh_backend/database.py +++ b/src/webssh_backend/database.py @@ -7,7 +7,8 @@ from typing import Any, Dict, List, Optional from cryptography.fernet import Fernet -from .config import ENCRYPTION_KEY, STORAGE_FILE, logger +from .config import ENCRYPTION_KEY, LEGACY_USERS, STORAGE_FILE, logger +from .auth import hash_password class CredentialsManager: @@ -101,10 +102,229 @@ class CredentialsManager: renewal_request_id TEXT ) ''') + cursor.execute(''' + CREATE TABLE IF NOT EXISTS users ( + username TEXT PRIMARY KEY, + password_hash TEXT NOT NULL, + role TEXT NOT NULL CHECK (role IN ('admin', 'user')), + must_change_password INTEGER NOT NULL DEFAULT 0, + auth_version INTEGER NOT NULL DEFAULT 0, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP + ) + ''') + cursor.execute(''' + CREATE TABLE IF NOT EXISTS schema_migrations ( + migration_id TEXT PRIMARY KEY, + applied_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP + ) + ''') + user_columns = { + row[1] for row in cursor.execute('PRAGMA table_info(users)').fetchall() + } + if 'auth_version' not in user_columns: + cursor.execute('ALTER TABLE users ADD COLUMN auth_version INTEGER NOT NULL DEFAULT 0') + cursor.execute( + "INSERT OR IGNORE INTO schema_migrations (migration_id) VALUES (?)", + ('003_user_auth_version',) + ) + cursor.execute( + "INSERT OR IGNORE INTO schema_migrations (migration_id) VALUES (?)", + ('001_users_table',) + ) conn.commit() conn.close() + self._import_legacy_users() logger.info(f"Credentials database initialized at {self.db_path}") + def _import_legacy_users(self): + """Import environment users once; legacy passwords must be changed at first login.""" + conn = sqlite3.connect(self.db_path) + try: + conn.execute('BEGIN IMMEDIATE') + already_imported = conn.execute( + 'SELECT 1 FROM schema_migrations WHERE migration_id = ?', + ('002_legacy_users_import',) + ).fetchone() + if already_imported: + conn.commit() + return + + user_count = conn.execute('SELECT COUNT(*) FROM users').fetchone()[0] + imported_count = 0 + if user_count == 0: + for index, (username, password) in enumerate(LEGACY_USERS): + conn.execute(''' + INSERT INTO users (username, password_hash, role, must_change_password) + VALUES (?, ?, ?, 1) + ''', ( + username, + hash_password(password), + 'admin' if index == 0 else 'user' + )) + imported_count += 1 + + conn.execute( + 'INSERT INTO schema_migrations (migration_id) VALUES (?)', + ('002_legacy_users_import',) + ) + conn.commit() + if imported_count: + logger.info(f"Imported {imported_count} legacy user(s); password change required at next login") + except Exception: + conn.rollback() + raise + finally: + conn.close() + + def get_user(self, username: str) -> Optional[Dict[str, Any]]: + """Return a user record without exposing it to client responses.""" + conn = sqlite3.connect(self.db_path) + try: + row = conn.execute(''' + SELECT username, password_hash, role, must_change_password, created_at, auth_version + FROM users WHERE username = ? + ''', (username,)).fetchone() + if not row: + return None + return { + 'username': row[0], + 'password_hash': row[1], + 'role': row[2], + 'must_change_password': bool(row[3]), + 'created_at': row[4], + 'auth_version': row[5], + } + finally: + conn.close() + + def has_users(self) -> bool: + conn = sqlite3.connect(self.db_path) + try: + return conn.execute('SELECT 1 FROM users LIMIT 1').fetchone() is not None + finally: + conn.close() + + def list_users(self) -> List[Dict[str, Any]]: + conn = sqlite3.connect(self.db_path) + try: + rows = conn.execute(''' + SELECT username, role, must_change_password, created_at + FROM users ORDER BY username COLLATE NOCASE + ''').fetchall() + return [ + { + 'username': row[0], + 'role': row[1], + 'must_change_password': bool(row[2]), + 'created_at': row[3], + } + for row in rows + ] + finally: + conn.close() + + def create_user(self, username: str, password_hash: str, role: str) -> bool: + conn = sqlite3.connect(self.db_path) + try: + conn.execute(''' + INSERT INTO users (username, password_hash, role, must_change_password) + VALUES (?, ?, ?, 1) + ''', (username, password_hash, role)) + conn.commit() + return True + except sqlite3.IntegrityError: + conn.rollback() + return False + except Exception: + conn.rollback() + raise + finally: + conn.close() + + def delete_user(self, username: str, requested_by: str) -> str: + """Delete a user without permitting self-deletion or removal of the last admin.""" + if username == requested_by: + return 'self' + + conn = sqlite3.connect(self.db_path) + try: + conn.execute('BEGIN IMMEDIATE') + row = conn.execute('SELECT role FROM users WHERE username = ?', (username,)).fetchone() + if not row: + conn.rollback() + return 'not_found' + if row[0] == 'admin': + admin_count = conn.execute( + "SELECT COUNT(*) FROM users WHERE role = 'admin'" + ).fetchone()[0] + if admin_count <= 1: + conn.rollback() + return 'last_admin' + conn.execute('DELETE FROM users WHERE username = ?', (username,)) + conn.commit() + return 'deleted' + except Exception: + conn.rollback() + raise + finally: + conn.close() + + def create_first_admin(self, username: str, password_hash: str) -> bool: + """Atomically create the initial administrator only when no users exist.""" + conn = sqlite3.connect(self.db_path) + try: + conn.execute('BEGIN IMMEDIATE') + if conn.execute('SELECT 1 FROM users LIMIT 1').fetchone(): + conn.rollback() + return False + conn.execute(''' + INSERT INTO users (username, password_hash, role, must_change_password) + VALUES (?, ?, 'admin', 0) + ''', (username, password_hash)) + conn.commit() + return True + except Exception: + conn.rollback() + raise + finally: + conn.close() + + def update_user_password(self, username: str, password_hash: str) -> bool: + conn = sqlite3.connect(self.db_path) + try: + cursor = conn.execute(''' + UPDATE users + SET password_hash = ?, must_change_password = 0, + auth_version = auth_version + 1, updated_at = CURRENT_TIMESTAMP + WHERE username = ? + ''', (password_hash, username)) + conn.commit() + return cursor.rowcount == 1 + except Exception: + conn.rollback() + raise + finally: + conn.close() + + def reset_user_password(self, username: str, password_hash: str) -> bool: + """Set a temporary password and invalidate all previously issued tokens.""" + conn = sqlite3.connect(self.db_path) + try: + cursor = conn.execute(''' + UPDATE users + SET password_hash = ?, must_change_password = 1, + auth_version = auth_version + 1, updated_at = CURRENT_TIMESTAMP + WHERE username = ? + ''', (password_hash, username)) + conn.commit() + return cursor.rowcount == 1 + except Exception: + conn.rollback() + raise + finally: + conn.close() + def _encrypt(self, data: str) -> bytes: """Encrypt string data.""" if not data: diff --git a/src/webssh_backend/http_api.py b/src/webssh_backend/http_api.py index 0c8580d..fbce99f 100644 --- a/src/webssh_backend/http_api.py +++ b/src/webssh_backend/http_api.py @@ -1,10 +1,10 @@ """ HTTP API module - LoginHandler with all HTTP endpoints. """ -import hashlib import io import json import os +import re import socket import stat from concurrent.futures import ThreadPoolExecutor @@ -13,12 +13,18 @@ from http.server import BaseHTTPRequestHandler from typing import Optional from urllib.parse import parse_qs, urlparse -import jwt import paramiko -from .auth import JWT_ALGORITHM, JWT_EXPIRE_HOURS, JWT_SECRET, USERS, generate_token, verify_token_http +from .auth import ( + JWT_EXPIRE_HOURS, + decode_token, + generate_token, + hash_password, + verify_password, +) from .config import HTTP_PORT, MAX_UPLOAD_SIZE, SFTP_CHUNK_SIZE, WS_PORT, logger from .database import creds_manager +from . import __version__ APP_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) STATIC_DIR = os.path.join(APP_ROOT, 'static') @@ -40,7 +46,17 @@ class LoginHandler(BaseHTTPRequestHandler): def do_POST(self): parsed = urlparse(self.path) - if parsed.path == '/api/login': + if parsed.path == '/api/setup': + self.handle_initial_setup() + elif parsed.path == '/api/change-password': + self.handle_change_password() + elif parsed.path == '/api/verify-password': + self.handle_verify_password() + elif parsed.path == '/api/users': + self.handle_create_user() + elif parsed.path == '/api/users/delete': + self.handle_delete_user() + elif parsed.path == '/api/login': self.handle_login() elif parsed.path == '/api/verify': self.handle_verify() @@ -74,6 +90,17 @@ class LoginHandler(BaseHTTPRequestHandler): def do_GET(self): parsed = urlparse(self.path) + if parsed.path == '/api/setup-status': + self.send_json_response(200, {'needs_setup': not creds_manager.has_users()}) + return + if parsed.path == '/api/users': + self.handle_get_users() + return + + if parsed.path == '/api/version': + self.send_json_response(200, {'version': __version__}) + return + # API endpoint for connections if parsed.path == '/api/connections': self.handle_get_connections() @@ -160,14 +187,106 @@ class LoginHandler(BaseHTTPRequestHandler): logger.error(f"Error serving file {full_path}: {e}") self.send_error(500) - def _get_username_from_token(self) -> Optional[str]: + def _get_username_from_token(self, allow_password_change: bool = False) -> Optional[str]: """Extract username from JWT token in Authorization header.""" auth_header = self.headers.get('Authorization', '') if auth_header.startswith('Bearer '): - token = auth_header[7:] - return verify_token_http(token) + payload = decode_token(auth_header[7:]) + if not payload: + return None + username = payload.get('sub') + user = creds_manager.get_user(username) if username else None + if not user: + return None + token_requires_password_change = bool(payload.get('pwd_change')) + if token_requires_password_change: + if payload.get('auth_version', 0) != user['auth_version']: + return None + return username if allow_password_change and user['must_change_password'] else None + if payload.get('auth_version', 0) != user['auth_version']: + return None + if user['must_change_password'] and not allow_password_change: + return None + return username return None + def _require_admin(self) -> Optional[str]: + username = self._get_username_from_token() + if not username: + self.send_json_response(401, {'error': 'Authentication required'}) + return None + user = creds_manager.get_user(username) + if not user or user['role'] != 'admin': + self.send_json_response(403, {'error': 'Administrator role required'}) + return None + return username + + def handle_get_users(self): + if not self._require_admin(): + return + try: + self.send_json_response(200, {'users': creds_manager.list_users()}) + except Exception as e: + logger.error(f"Get users error: {e}") + self.send_json_response(500, {'error': 'Internal error'}) + + def handle_create_user(self): + if not self._require_admin(): + return + try: + data = self._read_json_body() + username = data.get('username', '') + password = data.get('password', '') + role = data.get('role', '') + if not isinstance(username, str): + self.send_json_response(400, {'error': 'Username must be a string'}) + return + username = username.strip() + if not self._valid_username(username): + self.send_json_response(400, {'error': 'Username must be 3-64 characters: letters, numbers, ., _ or -'}) + return + if role not in ('admin', 'user'): + self.send_json_response(400, {'error': 'Role must be admin or user'}) + return + if not self._valid_new_password(password): + self.send_json_response(400, {'error': 'Password must be at least 12 characters'}) + return + if not creds_manager.create_user(username, hash_password(password), role): + self.send_json_response(409, {'error': 'Username already exists'}) + return + logger.info(f"Created {role} account") + self.send_json_response(201, {'success': True}) + except (json.JSONDecodeError, TypeError, ValueError): + self.send_json_response(400, {'error': 'Invalid request'}) + except Exception as e: + logger.error(f"Create user error: {e}") + self.send_json_response(500, {'error': 'Internal error'}) + + def handle_delete_user(self): + requested_by = self._require_admin() + if not requested_by: + return + try: + data = self._read_json_body() + username = data.get('username', '') + if not isinstance(username, str) or not username: + self.send_json_response(400, {'error': 'Username is required'}) + return + + result = creds_manager.delete_user(username, requested_by) + if result == 'deleted': + self.send_json_response(200, {'success': True}) + elif result == 'not_found': + self.send_json_response(404, {'error': 'User not found'}) + else: + message = 'You cannot delete your own account' if result == 'self' else 'The last administrator cannot be deleted' + self.send_json_response(409, {'error': message}) + except (json.JSONDecodeError, TypeError, ValueError): + self.send_json_response(400, {'error': 'Invalid request'}) + except Exception as e: + logger.error(f"Delete user error: {e}") + self.send_json_response(500, {'error': 'Internal error'}) + def handle_save_connection(self): """Save or update a connection.""" username = self._get_username_from_token() @@ -252,8 +371,8 @@ class LoginHandler(BaseHTTPRequestHandler): config = { 'max_upload_size': MAX_UPLOAD_SIZE, 'max_upload_size_mb': MAX_UPLOAD_SIZE / 1024 / 1024, - 'stored_users': USERS, # Include hashed passwords for client-side verification 'current_user': username, + 'current_role': creds_manager.get_user(username)['role'], 'ws_port': WS_PORT, 'http_port': HTTP_PORT, 'sftp_ws_port': SFTP_CHUNK_SIZE @@ -290,32 +409,42 @@ class LoginHandler(BaseHTTPRequestHandler): def handle_login(self): try: - content_length = int(self.headers.get('Content-Length', 0)) - body = self.rfile.read(content_length).decode('utf-8') - data = json.loads(body) + data = self._read_json_body() - username = data.get('username', '').strip() + username = data.get('username', '') password = data.get('password', '') - + if not isinstance(username, str) or not isinstance(password, str): + self.send_json_response(400, {'error': 'Username and password must be strings'}) + return + username = username.strip() if not username or not password: self.send_json_response(400, {'error': 'Username and password required'}) return - # Check credentials - password_hash = hashlib.sha256(password.encode()).hexdigest() - if USERS.get(username) != password_hash: + user = creds_manager.get_user(username) + if not user or not verify_password(password, user['password_hash']): logger.warning(f"Failed login attempt for user: {username}") self.send_json_response(401, {'error': 'Invalid credentials'}) return - # Generate JWT token - token = generate_token(username) + must_change_password = user['must_change_password'] + token = generate_token( + username, + password_change_required=must_change_password, + auth_version=user['auth_version'], + ) expire = datetime.now(timezone.utc) + timedelta(hours=JWT_EXPIRE_HOURS) logger.info(f"Successful login for user: {username}") - self.send_json_response(200, {'token': token, 'username': username, 'expires': expire.isoformat()}) + self.send_json_response(200, { + 'token': token, + 'username': username, + 'role': user['role'], + 'password_change_required': must_change_password, + 'expires': expire.isoformat() + }) - except json.JSONDecodeError: + except (json.JSONDecodeError, TypeError, ValueError): self.send_json_response(400, {'error': 'Invalid JSON'}) except Exception as e: logger.error(f"Login error: {e}") @@ -323,22 +452,122 @@ class LoginHandler(BaseHTTPRequestHandler): def handle_verify(self): try: - content_length = int(self.headers.get('Content-Length', 0)) - body = self.rfile.read(content_length).decode('utf-8') - data = json.loads(body) + data = self._read_json_body() token = data.get('token', '') + if not isinstance(token, str): + self.send_json_response(400, {'valid': False, 'error': 'Invalid token'}) + return - try: - payload = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM]) - self.send_json_response(200, {'valid': True, 'username': payload['sub']}) - except jwt.ExpiredSignatureError: - self.send_json_response(401, {'valid': False, 'error': 'Token expired'}) - except jwt.InvalidTokenError: + payload = decode_token(token) + user = creds_manager.get_user(payload.get('sub')) if payload else None + if not user: self.send_json_response(401, {'valid': False, 'error': 'Invalid token'}) + return + if payload.get('auth_version', 0) != user['auth_version']: + self.send_json_response(401, {'valid': False, 'error': 'Invalid token'}) + return + if payload.get('pwd_change') and not user['must_change_password']: + self.send_json_response(401, {'valid': False, 'error': 'Invalid token'}) + return + password_change_required = user['must_change_password'] or bool(payload.get('pwd_change')) + self.send_json_response(200, { + 'valid': True, + 'username': user['username'], + 'role': user['role'], + 'password_change_required': password_change_required + }) + except (json.JSONDecodeError, TypeError, ValueError): + self.send_json_response(400, {'valid': False, 'error': 'Invalid request'}) except Exception as e: logger.error(f"Verify error: {e}") self.send_json_response(500, {'error': 'Internal error'}) + def handle_initial_setup(self): + try: + data = self._read_json_body() + username = data.get('username', '').strip() + password = data.get('password', '') + if not self._valid_username(username): + self.send_json_response(400, {'error': 'Username must be 3-64 characters: letters, numbers, ., _ or -'}) + return + if not self._valid_new_password(password): + self.send_json_response(400, {'error': 'Password must be at least 12 characters'}) + return + + if not creds_manager.create_first_admin(username, hash_password(password)): + self.send_json_response(409, {'error': 'Initial administrator has already been created'}) + return + logger.info("Initial administrator account created") + self.send_json_response(201, {'success': True}) + except (json.JSONDecodeError, TypeError, ValueError): + self.send_json_response(400, {'error': 'Invalid request'}) + except Exception as e: + logger.error(f"Initial setup error: {e}") + self.send_json_response(500, {'error': 'Internal error'}) + + def handle_change_password(self): + username = self._get_username_from_token(allow_password_change=True) + if not username: + self.send_json_response(401, {'error': 'Authentication required'}) + return + + try: + data = self._read_json_body() + password = data.get('password', '') + if not self._valid_new_password(password): + self.send_json_response(400, {'error': 'Password must be at least 12 characters'}) + return + if not creds_manager.update_user_password(username, hash_password(password)): + self.send_json_response(404, {'error': 'User not found'}) + return + + user = creds_manager.get_user(username) + token = generate_token(username, auth_version=user['auth_version']) + logger.info(f"Password changed for user {username}") + self.send_json_response(200, { + 'success': True, + 'token': token, + 'username': username, + 'role': user['role'] + }) + except (json.JSONDecodeError, TypeError, ValueError): + self.send_json_response(400, {'error': 'Invalid request'}) + except Exception as e: + logger.error(f"Password change error: {e}") + self.send_json_response(500, {'error': 'Internal error'}) + + def handle_verify_password(self): + username = self._get_username_from_token() + if not username: + self.send_json_response(401, {'error': 'Authentication required'}) + return + + try: + data = self._read_json_body() + user = creds_manager.get_user(username) + valid = bool(user and verify_password(data.get('password', ''), user['password_hash'])) + self.send_json_response(200, {'valid': valid}) + except (json.JSONDecodeError, TypeError, ValueError): + self.send_json_response(400, {'error': 'Invalid request'}) + except Exception as e: + logger.error(f"Password verification error: {e}") + self.send_json_response(500, {'error': 'Internal error'}) + + def _read_json_body(self): + content_length = int(self.headers.get('Content-Length', 0)) + data = json.loads(self.rfile.read(content_length).decode('utf-8')) + if not isinstance(data, dict): + raise ValueError('JSON body must be an object') + return data + + @staticmethod + def _valid_username(username: str) -> bool: + return isinstance(username, str) and re.fullmatch(r'[A-Za-z0-9_.-]{3,64}', username) is not None + + @staticmethod + def _valid_new_password(password: str) -> bool: + return isinstance(password, str) and len(password) >= 12 and len(password.encode('utf-8')) <= 1024 + def send_json_response(self, status, data): self.send_response(status) self.send_header('Content-Type', 'application/json') diff --git a/tests/test_user_auth.py b/tests/test_user_auth.py new file mode 100644 index 0000000..327d787 --- /dev/null +++ b/tests/test_user_auth.py @@ -0,0 +1,329 @@ +import io +import json +import os +import sqlite3 +import tempfile +import threading +import unittest +from http.server import HTTPServer +import urllib.error +import urllib.request +from unittest.mock import patch + +from cryptography.fernet import Fernet + +import webssh_backend.database as database +import webssh_backend.http_api as http_api +import password_reset +from webssh_backend.auth import ( + generate_token, + hash_password, + verify_password, + verify_token_http, +) +from webssh_backend.database import CredentialsManager + + +class PasswordHashTests(unittest.TestCase): + def test_password_hash_uses_salt_and_verifies_password(self): + first_hash = hash_password("a-long-password") + second_hash = hash_password("a-long-password") + + self.assertNotEqual(first_hash, second_hash) + self.assertTrue(verify_password("a-long-password", first_hash)) + self.assertFalse(verify_password("wrong-password", first_hash)) + self.assertFalse(verify_password("a-long-password", "invalid-hash")) + + def test_password_change_token_is_limited_to_password_change(self): + token = generate_token("user", password_change_required=True) + + self.assertIsNone(verify_token_http(token)) + self.assertEqual(verify_token_http(token, allow_password_change=True), "user") + + +class UserDatabaseTests(unittest.TestCase): + def setUp(self): + self.temp_dir = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.temp_dir.name, "credentials.db") + + def create_manager(self, users): + with patch.object(database, "LEGACY_USERS", users): + return CredentialsManager(self.db_path, Fernet.generate_key().decode()) + + def test_legacy_users_are_imported_once_with_forced_password_change(self): + manager = self.create_manager([ + ("owner", "legacy-owner-password"), + ("operator", "legacy-user-password"), + ]) + + owner = manager.get_user("owner") + operator = manager.get_user("operator") + self.assertEqual(owner["role"], "admin") + self.assertEqual(operator["role"], "user") + self.assertTrue(owner["must_change_password"]) + self.assertTrue(operator["must_change_password"]) + self.assertTrue(verify_password("legacy-owner-password", owner["password_hash"])) + + with patch.object(database, "LEGACY_USERS", [("intruder", "another-password")]): + CredentialsManager(self.db_path, Fernet.generate_key().decode()) + self.assertIsNone(manager.get_user("intruder")) + + def test_existing_user_table_is_migrated_with_auth_version(self): + conn = sqlite3.connect(self.db_path) + conn.execute(''' + CREATE TABLE users ( + username TEXT PRIMARY KEY, + password_hash TEXT NOT NULL, + role TEXT NOT NULL, + must_change_password INTEGER NOT NULL DEFAULT 0, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP + ) + ''') + conn.execute(''' + CREATE TABLE schema_migrations ( + migration_id TEXT PRIMARY KEY, + applied_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP + ) + ''') + conn.execute( + "INSERT INTO schema_migrations (migration_id) VALUES ('001_users_table')" + ) + conn.execute( + "INSERT INTO schema_migrations (migration_id) VALUES ('002_legacy_users_import')" + ) + conn.execute( + "INSERT INTO users (username, password_hash, role) VALUES (?, ?, ?)", + ("existing", hash_password("existing-account-password"), "admin"), + ) + conn.commit() + conn.close() + + manager = self.create_manager([]) + + user = manager.get_user("existing") + self.assertEqual(user["auth_version"], 0) + self.assertEqual(user["role"], "admin") + conn = sqlite3.connect(self.db_path) + migration = conn.execute( + "SELECT 1 FROM schema_migrations WHERE migration_id = '003_user_auth_version'" + ).fetchone() + conn.close() + self.assertIsNotNone(migration) + + def test_first_admin_creation_is_only_allowed_once(self): + manager = self.create_manager([]) + + self.assertTrue(manager.create_first_admin("owner", hash_password("new-long-password"))) + self.assertFalse(manager.create_first_admin("second", hash_password("second-long-password"))) + self.assertEqual(manager.get_user("owner")["role"], "admin") + self.assertIsNone(manager.get_user("second")) + + def test_password_change_clears_required_flag(self): + manager = self.create_manager([("owner", "legacy-password")]) + + self.assertTrue(manager.update_user_password("owner", hash_password("replacement-password"))) + self.assertFalse(manager.get_user("owner")["must_change_password"]) + + def test_reset_requires_password_change_and_revokes_existing_tokens(self): + manager = self.create_manager([]) + manager.create_first_admin("owner", hash_password("original-account-password")) + self.start_api(manager) + + status, login = self.request_json("POST", "/api/login", { + "username": "owner", + "password": "original-account-password", + }) + self.assertEqual(status, 200) + existing_token = login["token"] + user = manager.get_user("owner") + + self.assertTrue(manager.reset_user_password( + "owner", + hash_password("temporary-account-password"), + )) + reset_user = manager.get_user("owner") + self.assertEqual(reset_user["auth_version"], user["auth_version"] + 1) + self.assertTrue(reset_user["must_change_password"]) + + status, _ = self.request_json("GET", "/api/config", authorization=existing_token) + self.assertEqual(status, 401) + status, reset_login = self.request_json("POST", "/api/login", { + "username": "owner", + "password": "temporary-account-password", + }) + self.assertEqual(status, 200) + self.assertTrue(reset_login["password_change_required"]) + + def test_password_reset_cli_updates_user_without_echoing_secret(self): + manager = self.create_manager([]) + manager.create_first_admin("owner", hash_password("original-account-password")) + fake_input = io.StringIO() + fake_output = io.StringIO() + fake_input.isatty = lambda: True + fake_output.isatty = lambda: True + with ( + patch.object(password_reset, "creds_manager", manager), + patch.object(password_reset.sys, "stdin", fake_input), + patch.object(password_reset.sys, "stdout", fake_output), + patch.object(password_reset.sys, "argv", ["password-reset", "owner"]), + patch.object(password_reset.getpass, "getpass", side_effect=[ + "temporary-account-password", + "temporary-account-password", + ]), + ): + result = password_reset.main() + + self.assertEqual(result, 0) + reset_user = manager.get_user("owner") + self.assertTrue(reset_user["must_change_password"]) + self.assertTrue(verify_password("temporary-account-password", reset_user["password_hash"])) + self.assertNotIn("temporary-account-password", fake_output.getvalue()) + + def test_initial_setup_creates_only_one_admin(self): + manager = self.create_manager([]) + self.start_api(manager) + + status, data = self.request_json("GET", "/api/setup-status") + self.assertEqual(status, 200) + self.assertTrue(data["needs_setup"]) + + status, _ = self.request_json("POST", "/api/setup", { + "username": "first-admin", + "password": "initial-password-long", + }) + self.assertEqual(status, 201) + self.assertEqual(manager.get_user("first-admin")["role"], "admin") + + status, _ = self.request_json("POST", "/api/setup", { + "username": "second-admin", + "password": "another-password-long", + }) + self.assertEqual(status, 409) + + def test_admin_can_manage_users_and_regular_users_cannot(self): + manager = self.create_manager([]) + manager.create_first_admin("owner", hash_password("owner-account-password")) + self.start_api(manager) + + status, owner_login = self.request_json("POST", "/api/login", { + "username": "owner", + "password": "owner-account-password", + }) + self.assertEqual(status, 200) + owner_token = owner_login["token"] + + status, _ = self.request_json( + "POST", + "/api/users", + { + "username": "operator", + "password": "temporary-user-password", + "role": "user", + }, + authorization=owner_token, + ) + self.assertEqual(status, 201) + self.assertTrue(manager.get_user("operator")["must_change_password"]) + + status, listed = self.request_json("GET", "/api/users", authorization=owner_token) + self.assertEqual(status, 200) + self.assertEqual({user["username"] for user in listed["users"]}, {"owner", "operator"}) + self.assertNotIn("password_hash", listed["users"][0]) + + status, _ = self.request_json( + "POST", + "/api/users/delete", + {"username": "owner"}, + authorization=owner_token, + ) + self.assertEqual(status, 409) + + manager.update_user_password("operator", hash_password("operator-account-password")) + status, operator_login = self.request_json("POST", "/api/login", { + "username": "operator", + "password": "operator-account-password", + }) + self.assertEqual(status, 200) + status, _ = self.request_json( + "GET", "/api/users", authorization=operator_login["token"] + ) + self.assertEqual(status, 403) + + status, _ = self.request_json( + "POST", + "/api/users/delete", + {"username": "operator"}, + authorization=owner_token, + ) + self.assertEqual(status, 200) + self.assertIsNone(manager.get_user("operator")) + + def test_imported_user_is_limited_until_password_is_changed(self): + manager = self.create_manager([("owner", "legacy-owner-password")]) + self.start_api(manager) + + status, login_data = self.request_json("POST", "/api/login", { + "username": "owner", + "password": "legacy-owner-password", + }) + self.assertEqual(status, 200) + self.assertTrue(login_data["password_change_required"]) + pending_token = login_data["token"] + + status, _ = self.request_json( + "GET", "/api/config", authorization=pending_token + ) + self.assertEqual(status, 401) + + status, changed = self.request_json( + "POST", + "/api/change-password", + {"password": "replacement-owner-password"}, + authorization=pending_token, + ) + self.assertEqual(status, 200) + + status, _ = self.request_json("POST", "/api/verify", {"token": pending_token}) + self.assertEqual(status, 401) + status, verified = self.request_json("POST", "/api/verify", {"token": changed["token"]}) + self.assertEqual(status, 200) + self.assertTrue(verified["valid"]) + self.assertEqual(verified["role"], "admin") + + def start_api(self, manager): + self.server = HTTPServer(("127.0.0.1", 0), http_api.LoginHandler) + self.server_thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.server_thread.start() + self.manager_patch = patch.object(http_api, "creds_manager", manager) + self.manager_patch.start() + self.base_url = f"http://127.0.0.1:{self.server.server_port}" + + def request_json(self, method, path, body=None, authorization=None): + headers = {"Content-Type": "application/json"} + if authorization: + headers["Authorization"] = f"Bearer {authorization}" + request = urllib.request.Request( + self.base_url + path, + data=json.dumps(body).encode() if body is not None else None, + headers=headers, + method=method, + ) + try: + with urllib.request.urlopen(request) as response: + return response.status, json.loads(response.read()) + except urllib.error.HTTPError as error: + return error.code, json.loads(error.read()) + + def tearDown(self): + if hasattr(self, "manager_patch"): + self.manager_patch.stop() + self.server.shutdown() + self.server.server_close() + self.server_thread.join(timeout=2) + self.temp_dir.cleanup() + super().tearDown() + + +if __name__ == "__main__": + unittest.main()