LAB/PegaProx_project-pegaprox
1
0
Fork 0
mirror of https://github.com/PegaProx/project-pegaprox.git synced 2026-08-12 15:27:47 +08:00
Code Issues Packages Projects Releases Wiki Activity
PegaProx_project-pegaprox/pegaprox/api/auth.py
ruben 9bb78ebb7a oidc(#486): bind the legacy-username fallback to the same subject 
The back-compat lookup added in the previous commit adopted an existing
account whenever its key matched the local part of the derived username. That
re-opened the very collision the commit set out to close, just one step later:
bob@partner.com logging in found the 'bob' row that bob@corp.com had
provisioned pre-change and inherited its role and tenant.

A legacy row is now only reused when it is provably the same identity -- its
auth_source is an OIDC-family one and its stored oidc_sub equals the incoming
sub claim. Local and LDAP rows are never adopted. A sub is only unique within
an issuer and no issuer is recorded on the user row, so installs that repoint
at a different IdP still have to rekey by hand; likewise a pre-change account
that never completed an OIDC login has no oidc_sub and now needs a manual
rekey rather than being silently adopted.

'+' joins the allowed char set. sanitize_username() permits it, and dropping
it merged identities the same way dropping '@' did -- a+b@example.com derived
onto a real ab@example.com account.

oidc_derive_username() takes the already-loaded users table. Both callers in
the login path had one in hand and were loading it a second time.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-05 15:36:45 +02:00

1785 lines
82 KiB
Python
Raw Permalink Blame History

# -*- coding: utf-8 -*-
"""auth routes (login, logout, 2FA, OIDC, API tokens) - split from monolith dec 2025, NS"""
import time
import logging
import secrets
import base64
import ipaddress
from datetime import datetime
from flask import Blueprint, jsonify, request, make_response
from pegaprox.constants import *
from pegaprox.globals import *
from pegaprox.models.permissions import *
from pegaprox.core.db import get_db
from pegaprox.utils.auth import (
hash_password, verify_password, needs_password_rehash,
validate_password_policy, load_users, save_users,
create_initial_admin, is_initialized,
create_session, validate_session, invalidate_session,
invalidate_all_user_sessions, cleanup_expired_sessions,
generate_api_token, create_api_token, validate_api_token,
list_user_tokens, revoke_api_token, require_auth,
generate_session_id, mark_admin_initialized, ensure_api_tokens_table,
dummy_verify_password,
ARGON2_AVAILABLE, TOTP_AVAILABLE,
)
from pegaprox.utils.audit import log_audit, get_client_ip
from pegaprox.utils.ldap import get_ldap_settings, ldap_authenticate, ldap_provision_user
from pegaprox.utils.oidc import (
get_oidc_settings, get_oidc_endpoints, oidc_build_auth_url,
oidc_exchange_code, oidc_decode_id_token, oidc_get_user_info,
oidc_get_user_groups, oidc_map_groups_to_role, oidc_provision_user,
oidc_derive_username,
)
from pegaprox.utils.rbac import get_user_permissions, DEFAULT_TENANT_ID
from pegaprox.api.helpers import load_server_settings, save_server_settings, get_login_settings, get_session_timeout, safe_error, effective_reverse_proxy
from pegaprox.utils.sanitization import sanitize_identifier, sanitize_username
from pegaprox.utils.ssh import check_auth_action_rate_limit
from pegaprox.utils.url_security import sanitize_outbound_url, SsrfError
# NS: Mar 2026 - removed add_allowed_origin import (no longer auto-adding on login)
import requests
try:
import pyotp
import qrcode
import io as _io
except ImportError:
pass
bp = Blueprint('auth', __name__)
# ============================================================================
@bp.route('/api/auth/oidc/authorize', methods=['GET'])
def oidc_authorize():
"""Initiate OIDC login flow - redirects user to identity provider
NS: Generates CSRF state, stores in session, redirects to IdP
"""
config = get_oidc_settings()
if not config['enabled'] or not config['client_id']:
return jsonify({'error': 'OIDC authentication is not configured'}), 400
# NS: Auto-detect redirect URI if not configured
# never use Origin header - attacker can spoof it to steal OAuth tokens
if not config.get('redirect_uri'):
base_url = request.host_url.rstrip('/')
# MK: behind reverse proxy, trust forwarded headers for correct scheme/host
if request.headers.get('X-Forwarded-Proto') == 'https':
fwd_host = request.headers.get('X-Forwarded-Host') or request.headers.get('Host') or request.host
base_url = f"https://{fwd_host.split(',')[0].strip()}"
config['redirect_uri'] = f"{base_url}/oidc/callback"
logging.info(f"[OIDC] Auto-detected redirect_uri: {config['redirect_uri']}")
# MK: Generate state for CSRF protection
state = secrets.token_urlsafe(32)
# Store state in a temporary way (cookie-based since no session yet)
auth_url, nonce, code_verifier = oidc_build_auth_url(config, state)
response = make_response(jsonify({'auth_url': auth_url}))
from pegaprox.utils.audit import _is_trusted_proxy
is_secure = request.is_secure or (_is_trusted_proxy(request.remote_addr) and request.headers.get('X-Forwarded-Proto') == 'https')
# #188: store state + nonce + PKCE code_verifier in cookie
# NS: Apr 2026 - append redirect_after for portal OIDC flow
# MK 2026-05-31 — defense-in-depth: `startswith('/')` allows `//attacker.com`
# (protocol-relative URL), bypassing the intent. Require single leading
# slash + reject any embedded `//`, `\\`, control chars, or fragment-like
# constructs. Net effect: only same-origin internal paths get persisted.
def _safe_internal_path(p):
if not p or not isinstance(p, str) or len(p) > 200:
return False
if not p.startswith('/'):
return False
if p.startswith('//') or p.startswith('/\\'):
return False # protocol-relative
if any(c in p for c in ('\r', '\n', '\t', '\\')):
return False # control chars / backslash tricks
return True
redirect_after = request.args.get('redirect_after', '')
cookie_val = f"{state}:{nonce}:{code_verifier}"
if _safe_internal_path(redirect_after):
cookie_val += f":{redirect_after}"
response.set_cookie('oidc_state', cookie_val, httponly=True, secure=is_secure, samesite='Lax', max_age=600)
return response
@bp.route('/api/auth/oidc/callback', methods=['GET'])
def oidc_callback_redirect():
"""NS: Safety net - if IdP sends GET to API callback URL, redirect to frontend SPA route"""
from flask import redirect as flask_redirect
return flask_redirect(f"/oidc/callback?{request.query_string.decode()}")
@bp.route('/api/auth/oidc/callback', methods=['POST'])
def oidc_callback():
"""Handle OIDC callback - exchange code for tokens and create session
LW: Called by frontend after redirect back from IdP
Frontend sends: {code, state} from URL query params
"""
# MK: Mar 2026 - use centralized IP resolution (respects trusted_proxies)
client_ip = get_client_ip()
oidc_cb_key = f'oidc_cb_{client_ip}'
if oidc_cb_key in login_attempts_by_ip:
attempts = login_attempts_by_ip[oidc_cb_key].get('attempts', [])
recent = [t for t in attempts if time.time() - t < 60]
if len(recent) >= 10:
logging.warning(f"[OIDC] Rate limit hit for callback from {client_ip}")
return jsonify({'error': 'Too many attempts. Try again later.'}), 429
login_attempts_by_ip[oidc_cb_key] = {'attempts': recent + [time.time()]}
else:
login_attempts_by_ip[oidc_cb_key] = {'attempts': [time.time()]}
config = get_oidc_settings()
if not config['enabled']:
return jsonify({'error': 'OIDC authentication is not configured'}), 400
# NS: Auto-detect redirect URI if not configured (must match what was sent in authorize!)
if not config.get('redirect_uri'):
base_url = request.host_url.rstrip('/')
if request.headers.get('X-Forwarded-Proto') == 'https':
fwd_host = request.headers.get('X-Forwarded-Host') or request.headers.get('Host') or request.host
base_url = f"https://{fwd_host.split(',')[0].strip()}"
config['redirect_uri'] = f"{base_url}/oidc/callback"
data = request.get_json() or {}
code = data.get('code', '')
state = data.get('state', '')
if not code:
return jsonify({'error': 'Authorization code is required'}), 400
# NS: Verify CSRF state + extract nonce + PKCE verifier (#188)
stored_cookie = request.cookies.get('oidc_state', '')
# #188: verbose logging for debugging cookie issues behind reverse proxy
if not stored_cookie:
logging.warning(f"[OIDC] oidc_state cookie MISSING — browser didn't send it back. Check proxy cookie forwarding.")
else:
logging.debug(f"[OIDC] oidc_state cookie present ({len(stored_cookie)} chars)")
stored_nonce = None
stored_verifier = None
cookie_parts = stored_cookie.split(':')
stored_state = cookie_parts[0] if cookie_parts else ''
if len(cookie_parts) >= 2:
stored_nonce = cookie_parts[1]
if len(cookie_parts) >= 3:
stored_verifier = cookie_parts[2]
# NS: Apr 2026 - redirect_after for portal OIDC flow
stored_redirect_after = cookie_parts[3] if len(cookie_parts) >= 4 else ''
if not stored_state or stored_state != state:
logging.warning(f"[OIDC] State mismatch — stored={stored_state[:8] if stored_state else 'EMPTY'}..., received={state[:8] if state else 'EMPTY'}...")
return jsonify({'error': 'Invalid state parameter (CSRF protection). Your browser may not be sending cookies — check reverse proxy configuration.'}), 400
# Step 1: Exchange code for tokens (with PKCE verifier if available)
token_data = oidc_exchange_code(config, code, code_verifier=stored_verifier)
if 'error' in token_data:
logging.warning(f"[OIDC] Token exchange failed: {token_data['error']}")
return jsonify({'error': 'Authentication failed - please try again'}), 401
access_token = token_data.get('access_token', '')
id_token_raw = token_data.get('id_token', '')
if not access_token:
return jsonify({'error': 'No access token received'}), 401
# Step 2: Decode ID token for claims (with nonce + exp validation - MK Feb 2026)
id_claims = {}
if id_token_raw:
id_claims = oidc_decode_id_token(id_token_raw, expected_nonce=stored_nonce, config=config)
if 'error' in id_claims:
logging.warning(f"[OIDC] ID token validation failed: {id_claims['error']}")
return jsonify({'error': f"Authentication failed - {id_claims['error']}"}), 401
# Step 3: Get user info from provider
user_info = oidc_get_user_info(config, access_token)
if not user_info:
# MK: Fallback to ID token claims
user_info = id_claims
if not user_info or not (user_info.get('preferred_username') or user_info.get('email') or user_info.get('sub')):
return jsonify({'error': 'Could not retrieve user information from provider'}), 401
# Step 4: Get group memberships for role mapping
groups = oidc_get_user_groups(config, access_token)
role_mapping = oidc_map_groups_to_role(config, groups, id_claims)
# Step 5: Provision/update local user
if not config['auto_create_users']:
# Check if user already exists
# MK: this has to match the logic in oidc_provision_user or we get mismatches
# (had a bug where "john.doe" here vs "johndoe" in provision caused 403s)
users = load_users()
check_username = oidc_derive_username(user_info, users)
if check_username not in users:
return jsonify({'error': 'User account does not exist. Contact an administrator.'}), 403
# Determine auth_source label
provider = config.get('provider', 'oidc')
auth_source = 'entra' if provider == 'entra' else 'oidc'
user = oidc_provision_user(user_info, role_mapping, auth_source=auth_source)
# NS: SECURITY - oidc_provision_user returns None if local account would be overwritten
if not user:
return jsonify({'error': 'A local account with this username already exists. Contact an administrator.'}), 403
username = user.get('username', '')
# Check if user is enabled
users = load_users()
if username in users and not users[username].get('enabled', True):
return jsonify({'error': 'Account is disabled'}), 403
# Step 6: Create session via create_session() for proper session rotation + limits
# MK: create_session() handles max 3 sessions per user, session rotation, save_sessions()
session_token = create_session(username, user.get('role', ROLE_VIEWER))
log_audit(username, 'auth.oidc.login', f"OIDC login via {provider} from {client_ip}")
# NS: Apr 2026 - include portal_only so client portal can validate OIDC users
fresh_user = users.get(username, user)
resp_data = {
'success': True,
'user': username,
'role': user.get('role', ROLE_VIEWER),
'display_name': user.get('display_name', username),
'auth_source': auth_source,
'session_id': session_token,
'portal_only': fresh_user.get('portal_only', False),
}
if stored_redirect_after:
resp_data['redirect_after'] = stored_redirect_after
response = make_response(jsonify(resp_data))
# Set session cookie (same pattern as regular login)
from pegaprox.utils.audit import _is_trusted_proxy
is_secure = request.is_secure or (_is_trusted_proxy(request.remote_addr) and request.headers.get('X-Forwarded-Proto') == 'https')
response.set_cookie(
'session_id',
session_token,
httponly=True,
samesite='Strict',
secure=is_secure,
max_age=get_session_timeout()
)
# Clear OIDC state cookie
response.delete_cookie('oidc_state')
return response
@bp.route('/api/auth/oidc/config', methods=['GET'])
def oidc_get_public_config():
"""Return public OIDC config (non-sensitive) for login page
NS: Frontend needs to know if OIDC is enabled and button text
No auth required - this is used on the login page
"""
config = get_oidc_settings()
return jsonify({
'enabled': config['enabled'],
'provider': config.get('provider', 'entra'),
'button_text': config.get('button_text', 'Sign in with Microsoft'),
})
@bp.route('/api/settings/oidc/test', methods=['POST'])
@require_auth(roles=[ROLE_ADMIN])
def oidc_test_connection():
"""Test OIDC configuration by verifying endpoints are reachable
MK: Tests connectivity to OIDC discovery and token endpoints
"""
data = request.get_json() or {}
config = get_oidc_settings()
# Override with test data
for key in ['oidc_client_id', 'oidc_tenant_id', 'oidc_provider', 'oidc_authority', 'oidc_cloud_environment']:
if key in data:
short_key = key.replace('oidc_', '')
config[short_key] = data[key]
# MK May 2026 (#188 follow-up): accept `oidc_issuer` as an alias for
# `oidc_authority`. The OIDC spec calls it the "issuer", and external
# clients / CLI testers / future-Authentik-templates may send the
# standard term. Drop it onto `config['authority']` so the rest of the
# endpoint-resolution code (which reads `config.get('authority')`)
# picks it up unchanged.
if 'oidc_issuer' in data and not config.get('authority'):
config['authority'] = data['oidc_issuer']
results = []
# Step 1: Check endpoints exist
endpoints = get_oidc_endpoints(config) or {}
cloud_env = config.get('cloud_environment', 'commercial')
env_label = {'commercial': 'Commercial', 'gcc': 'GCC', 'gcc_high': 'GCC High', 'dod': 'DoD'}.get(cloud_env, cloud_env)
results.append({'step': 'Configuration', 'status': 'ok',
'detail': f"Provider: {config.get('provider', 'unknown')}, Tenant: {config.get('tenant_id') or 'N/A'}, Cloud: {env_label}"})
# MK May 2026 (#188 follow-up): if endpoint resolution failed outright
# (e.g. authority is empty / malformed → SSRF guard rejects discovery URL,
# or Entra config has no tenant), surface a clean error instead of
# cascading into AttributeError/KeyError on the empty endpoints dict.
if endpoints.get('_error'):
results.append({'step': 'Endpoint Resolution', 'status': 'error',
'detail': endpoints.get('_error_detail') or endpoints['_error']})
return jsonify({'success': False, 'results': results})
if not endpoints.get('authorization') and not endpoints.get('jwks'):
results.append({'step': 'Endpoint Resolution', 'status': 'error',
'detail': "Could not derive authorization/JWKS URLs. Check that the "
"OIDC authority/issuer URL is set and reachable, and (for Entra) "
"that the tenant_id is configured."})
return jsonify({'success': False, 'results': results})
# NS Apr 2026 (#188) — surface whether discovery succeeded so admins immediately see when
# PegaProx is using the (often wrong) issuer-relative fallback.
if config.get('provider') != 'entra':
if endpoints.get('_discovery_used'):
results.append({'step': 'Discovery', 'status': 'ok',
'detail': f"OIDC discovery succeeded — using endpoints from .well-known"})
else:
results.append({'step': 'Discovery', 'status': 'warning',
'detail': f"Discovery doc not reachable at {config.get('authority','').rstrip('/')}/.well-known/openid-configuration; "
f"using issuer-relative guesses. If this is Authentik/Keycloak/non-Entra, the auth endpoint will likely be wrong."})
# Step 2: Test authorization endpoint
try:
skip_ssl = bool(config.get('oidc_skip_ssl_verify', False))
# MK May 2026 - SSRF guard. Same pattern as utils/oidc.py (token/discovery/jwks).
# admin-only endpoint, but Authentik/Keycloak-style on-prem realms run on private
# nets, so honour the existing oidc_allow_private_ip toggle.
allow_private_ip = bool(config.get('oidc_allow_private_ip', False))
try:
# MK May 2026 (CodeAnt #504) — use normalised URL returned by the guard
# instead of the raw input so the request hits exactly what was validated.
validated_auth_url = sanitize_outbound_url(endpoints['authorization'], allow_private=allow_private_ip)
except SsrfError as guard_err:
hint = ''
if not allow_private_ip and 'private' in str(guard_err).lower():
hint = ' (set "Allow private IP" in OIDC settings if this is intentional)'
results.append({'step': 'Authorization Endpoint', 'status': 'error',
'detail': f"URL rejected by SSRF guard: {guard_err}{hint}"})
return jsonify({'success': False, 'results': results})
resp = requests.get(validated_auth_url, allow_redirects=False, timeout=10, verify=not skip_ssl)
# Auth endpoint should return 200 or redirect
if resp.status_code in [200, 302, 400]:
results.append({'step': 'Authorization Endpoint', 'status': 'ok', 'detail': endpoints['authorization']})
else:
hint = ''
if resp.status_code == 404 and not endpoints.get('_discovery_used'):
hint = ' (discovery failed → using issuer-relative URL; for Authentik the authorize endpoint is one path level above the issuer)'
results.append({'step': 'Authorization Endpoint', 'status': 'error',
'detail': f"HTTP {resp.status_code}{hint}"})
except Exception as e:
results.append({'step': 'Authorization Endpoint', 'status': 'error', 'detail': str(e)})
# Step 3: Test JWKS endpoint
try:
# MK May 2026 - same SSRF guard, same allow-private toggle.
allow_private_ip = bool(config.get('oidc_allow_private_ip', False))
try:
# MK May 2026 (CodeAnt #504) — same normalisation as Step 2 above.
validated_jwks_url = sanitize_outbound_url(endpoints['jwks'], allow_private=allow_private_ip)
except SsrfError as guard_err:
hint = ''
if not allow_private_ip and 'private' in str(guard_err).lower():
hint = ' (set "Allow private IP" in OIDC settings if this is intentional)'
results.append({'step': 'JWKS Endpoint', 'status': 'error',
'detail': f"URL rejected by SSRF guard: {guard_err}{hint}"})
return jsonify({'success': False, 'results': results})
resp = requests.get(validated_jwks_url, timeout=10)
if resp.status_code == 200:
keys = resp.json().get('keys', [])
results.append({'step': 'JWKS Endpoint', 'status': 'ok', 'detail': f"Found {len(keys)} signing keys"})
else:
results.append({'step': 'JWKS Endpoint', 'status': 'error', 'detail': f"HTTP {resp.status_code}"})
except Exception as e:
results.append({'step': 'JWKS Endpoint', 'status': 'error', 'detail': str(e)})
# Step 4: Check client_id is set
client_id = config.get('client_id') or ''
if client_id:
results.append({'step': 'Client ID', 'status': 'ok', 'detail': f"{client_id[:8]}..."})
else:
results.append({'step': 'Client ID', 'status': 'warning', 'detail': 'Not configured'})
# Step 5: Check redirect URI
if config.get('redirect_uri'):
results.append({'step': 'Redirect URI', 'status': 'ok', 'detail': config['redirect_uri']})
else:
results.append({'step': 'Redirect URI', 'status': 'warning', 'detail': 'Not configured - will auto-detect'})
all_ok = all(r['status'] == 'ok' for r in results)
return jsonify({'success': all_ok, 'results': results})
# MK May 2026 — first-run setup endpoint. Reachable only while the install
# is uninitialised; closes itself once the first admin is created. Replaces
# the old auto-bootstrapped `pegaprox/admin` default that exposed every
# fresh install to a network-attacker race.
_setup_attempts_by_ip = {} # very light rate-limit, IP → list[ts]
@bp.route('/api/auth/setup', methods=['POST'])
def auth_setup():
if is_initialized():
# already done, no replay
return jsonify({
'error': 'PegaProx is already initialised',
'code': 'ALREADY_INITIALIZED',
}), 409
client_ip = get_client_ip()
now = time.time()
# crude per-IP rate-limit: max 5 attempts / 60s. Mostly hygiene; the real
# race-window protection is the operator firewalling 5000 until setup
# completes. Document that in the install guide.
window = [t for t in _setup_attempts_by_ip.get(client_ip, []) if now - t < 60]
if len(window) >= 5:
logging.warning(f"[SETUP] rate-limited setup attempt from {client_ip}")
return jsonify({'error': 'Too many attempts, slow down'}), 429
window.append(now)
_setup_attempts_by_ip[client_ip] = window
data = request.get_json() or {}
username = sanitize_username(str(data.get('username', '')).strip().lower(), max_length=64)
password = data.get('password', '')
display_name = sanitize_identifier(str(data.get('display_name', '')).strip(), max_length=128)
email = str(data.get('email', '')).strip()[:254]
if not username or len(username) < 2:
return jsonify({'error': 'Username must be at least 2 characters'}), 400
# don't let setup conflict with the legacy default-admin name —
# avoids confusion in audit logs across the upgrade boundary
if username == 'pegaprox':
return jsonify({'error': "Choose a different username ('pegaprox' is reserved)"}), 400
if not password:
return jsonify({'error': 'Password required'}), 400
ok, err = validate_password_policy(password)
if not ok:
return jsonify({'error': err}), 400
# build, save, mark — order matters: if mark fails the next request
# would re-allow setup and double-create, so the audit log catches it.
try:
admin = create_initial_admin(username, password, display_name=display_name, email=email)
save_users(admin)
mark_admin_initialized()
except Exception as e:
logging.error(f"[SETUP] failed to create initial admin: {e}")
return jsonify({'error': 'Setup failed, check server logs'}), 500
log_audit(username, 'admin.initial_setup',
f"First admin '{username}' created via setup wizard from {client_ip}")
logging.info(f"[SETUP] initial admin '{username}' created from {client_ip}")
return jsonify({
'success': True,
'message': 'Setup complete, you can now log in',
'username': username,
})
@bp.route('/api/auth/login', methods=['POST'])
def auth_login():
"""login endpoint - MK"""
global users_db, login_attempts_by_ip, login_attempts_by_user
# MK May 2026 — first-run safety gate. Before the setup wizard runs there
# is no admin to authenticate against; refusing /login here closes the old
# hardcoded-creds path (`pegaprox/admin` was bootstrapped automatically
# which let any network-reachable fresh install be taken over).
if not is_initialized():
return jsonify({
'error': 'PegaProx is not initialised — run the setup wizard first',
'code': 'NOT_INITIALIZED',
}), 503
# get settings
login_settings = get_login_settings()
max_attempts = login_settings['max_attempts']
lockout_time = login_settings['lockout_time']
attempt_window = login_settings['attempt_window']
# MK: Mar 2026 - centralized IP resolution, handles trusted proxies + IPv6 normalization
client_ip = get_client_ip()
current_time = time.time()
# check if ip is locked
if client_ip in login_attempts_by_ip:
attempt_info = login_attempts_by_ip[client_ip]
if attempt_info.get('locked_until', 0) > current_time:
remaining = int(attempt_info['locked_until'] - current_time)
logging.warning(f"locked ip tried to login: {client_ip}")
# NS 2026-04-24 — uniform 401 (audit #5 applied here too). Previously
# returned 429 with a distinctive body, which let an observer distinguish
# IP-lockout vs. user-lockout vs. bad credentials.
resp = jsonify({
'error': 'Invalid credentials',
'locked': True,
'retry_after': remaining
})
resp.headers['Retry-After'] = str(remaining)
return resp, 401
# cleanup old attempts
attempt_info['attempts'] = [t for t in attempt_info.get('attempts', [])
if current_time - t < attempt_window]
data = request.get_json()
if not data:
return jsonify({'error': 'Invalid request body'}), 400
# sanitize inputs (security stuff)
username = sanitize_username(data.get('username', '').strip().lower(), max_length=64)
password = data.get('password', '')[:256] # limit to prevent DoS
totp_code = sanitize_identifier(data.get('totp_code', ''), max_length=10)
# NS Apr 2026 — WebAuthn proof token from a prior /api/webauthn/auth/finish
webauthn_proof = (data.get('webauthn_proof') or '')[:256]
# print(f"login: {username}") # DEBUG - remove before commit!! - NS
if not username or not password:
return jsonify({'error': 'Username and password required'}), 400
if len(username) < 2:
return jsonify({'error': 'Username too short'}), 400
# Check if username is locked out (additional protection against distributed attacks).
# NS 2026-04-24 — security audit: uniform 401 response here prevents username
# enumeration. Previously returned 429 "Account temporarily locked" which let an
# attacker probe whether a username existed (locked vs unknown gave different
# status codes). Now looks identical to "invalid credentials" to the caller;
# Retry-After header + retry_after body key tell the UI to show a cooldown timer.
if username in login_attempts_by_user:
user_attempt_info = login_attempts_by_user[username]
if user_attempt_info.get('locked_until', 0) > current_time:
remaining = int(user_attempt_info['locked_until'] - current_time)
logging.warning(f"Login attempt for locked user: {username} from {client_ip}, {remaining}s remaining")
resp = jsonify({
'error': 'Invalid credentials',
'locked': True, # UI hint — not status-code-leaky
'retry_after': remaining
})
resp.headers['Retry-After'] = str(remaining)
return resp, 401
# Clean up old attempts
user_attempt_info['attempts'] = [t for t in user_attempt_info.get('attempts', [])
if current_time - t < attempt_window]
# Helper function to record failed attempt (both IP and username)
# MK: Mar 2026 - also logs to audit trail now (was missing, security audit finding)
def record_failed_attempt(target_username=None):
locked = False
log_audit(target_username or 'unknown', 'auth.login_failed',
f"Failed login from {client_ip}" + (f" for user '{target_username}'" if target_username else ""))
# Track by IP
if client_ip not in login_attempts_by_ip:
login_attempts_by_ip[client_ip] = {'attempts': [], 'locked_until': 0}
login_attempts_by_ip[client_ip]['attempts'].append(current_time)
recent_ip = [t for t in login_attempts_by_ip[client_ip]['attempts']
if current_time - t < attempt_window]
if len(recent_ip) >= max_attempts:
login_attempts_by_ip[client_ip]['locked_until'] = current_time + lockout_time
logging.warning(f"IP {client_ip} locked out after {len(recent_ip)} failed attempts")
log_audit(target_username or 'unknown', 'auth.ip_locked', f"IP {client_ip} locked for {lockout_time}s")
locked = True
# Track by username (if provided and valid)
if target_username:
if target_username not in login_attempts_by_user:
login_attempts_by_user[target_username] = {'attempts': [], 'locked_until': 0}
login_attempts_by_user[target_username]['attempts'].append(current_time)
recent_user = [t for t in login_attempts_by_user[target_username]['attempts']
if current_time - t < attempt_window]
# Use slightly higher threshold for username (10 attempts) to prevent username enumeration lockout
if len(recent_user) >= max_attempts * 2:
login_attempts_by_user[target_username]['locked_until'] = current_time + lockout_time
logging.warning(f"User '{target_username}' locked out after {len(recent_user)} failed attempts")
locked = True
return locked
# Reload users in case they were updated
# NS: had a bug where user changes werent reflected until restart
users_db = load_users()
# =================================================================
# MK: Feb 2026 - LDAP Authentication (tried before local auth)
# Flow: LDAP enabled? → Try LDAP bind → JIT provision → session
# If LDAP fails or disabled → fall through to local password auth
# =================================================================
ldap_config = get_ldap_settings()
ldap_authenticated = False
if ldap_config['enabled']:
ldap_result = ldap_authenticate(username, password)
if ldap_result.get('success'):
# LW: LDAP auth succeeded - provision/update local user
if ldap_config['auto_create_users'] or username in users_db:
user = ldap_provision_user(ldap_result)
if user is None:
# NS: Local account exists - fall through to local auth
logging.info(f"[LDAP] User '{username}' has local account, skipping LDAP provisioning")
else:
users_db = load_users() # Reload after provisioning
ldap_authenticated = True
logging.info(f"[LDAP] User '{username}' authenticated via LDAP from {client_ip}")
else:
logging.warning(f"[LDAP] User '{username}' found in LDAP but auto-create is disabled")
return jsonify({'error': 'User not authorized in PegaProx. Contact admin.'}), 401
elif ldap_result.get('error') == 'User not found in LDAP':
# NS: User not in LDAP - fall through to local auth
logging.debug(f"[LDAP] User '{username}' not in LDAP, trying local auth")
elif 'Invalid LDAP credentials' in ldap_result.get('error', ''):
# User found in LDAP but wrong password - check if also local user
if username in users_db and users_db[username].get('auth_source') == 'ldap':
# Pure LDAP user - don't fall through to local auth
logging.warning(f"[LDAP] Failed login for LDAP user '{username}' from {client_ip}")
locked = record_failed_attempt(username)
if locked:
# NS: uniform 401 — retry timing via header, not status code (audit #5)
resp = jsonify({'error': 'Invalid credentials', 'locked': True, 'retry_after': lockout_time})
resp.headers['Retry-After'] = str(lockout_time)
return resp, 401
return jsonify({'error': 'Invalid credentials'}), 401
# Else: user exists locally too, fall through to local auth
else:
# LDAP server error - log but fall through to local auth
ldap_err = ldap_result.get('error', '')
logging.warning(f"[LDAP] Server error during auth: {ldap_err}")
# NS: Mar 2026 - TLS errors should be surfaced for LDAP-only users (#108)
# but still fall through for local users so they aren't locked out
if 'TLS' in ldap_err or 'certificate' in ldap_err.lower():
if username in users_db and users_db[username].get('auth_source') == 'ldap':
return jsonify({'error': ldap_err}), 401
# =================================================================
# Local Authentication (skipped if LDAP already authenticated)
# =================================================================
if not ldap_authenticated:
# check user exists
if username not in users_db:
# NS 2026-04-24 — equalize login timing. Without this the unknown-user branch
# returns in <1ms while verify_password() takes ~100ms, which is a free
# username-enumeration oracle via response-time measurement.
dummy_verify_password(password)
logging.warning(f"Login attempt for unknown user: {username} from {client_ip}")
locked = record_failed_attempt() # Don't track by username for unknown users
if locked:
# NS: uniform 401 — retry via header (audit #5, username enum prevention)
resp = jsonify({'error': 'Invalid credentials', 'locked': True, 'retry_after': lockout_time})
resp.headers['Retry-After'] = str(lockout_time)
return resp, 401
return jsonify({'error': 'Invalid credentials'}), 401
user = users_db[username]
# check user is enabled
if not user.get('enabled', True):
# burn the same ~100ms even on disabled accounts so existence isn't leaked
dummy_verify_password(password)
logging.warning(f"Login attempt for disabled user: {username} from {client_ip}")
return jsonify({'error': 'Invalid credentials'}), 401
# NS: LDAP-only users cannot login with local password
if user.get('auth_source') == 'ldap' and not user.get('password_hash'):
dummy_verify_password(password)
logging.warning(f"LDAP user '{username}' tried local login but has no local password")
return jsonify({'error': 'Please use LDAP credentials to sign in'}), 401
# Verify password
if not verify_password(password, user['password_salt'], user['password_hash']):
logging.warning(f"Failed login attempt for user: {username} from {client_ip}")
locked = record_failed_attempt(username)
if locked:
# uniform 401 — same body as IP-lockout + unknown-user paths (audit #5)
resp = jsonify({
'error': 'Invalid credentials',
'locked': True,
'retry_after': lockout_time
})
resp.headers['Retry-After'] = str(lockout_time)
return resp, 401
return jsonify({'error': 'Invalid credentials'}), 401
else:
user = users_db[username]
# check user is enabled (even LDAP users can be disabled locally)
if not user.get('enabled', True):
logging.warning(f"LDAP user '{username}' is disabled locally")
return jsonify({'error': 'Invalid credentials'}), 401
# NS Apr 2026 — 2FA: TOTP and/or WebAuthn. If a user has EITHER registered,
# they must pass ONE of them. WebAuthn short-circuits TOTP when a fresh
# proof token is supplied.
has_totp = bool(user.get('totp_enabled') and user.get('totp_secret'))
has_webauthn = False
try:
_db_probe = get_db()
_cnt_row = _db_probe.query_one(
'SELECT COUNT(*) AS n FROM webauthn_credentials WHERE username = ?', (username,)
)
has_webauthn = bool(_cnt_row and _cnt_row['n'] > 0)
except Exception:
has_webauthn = False
if has_totp or has_webauthn:
# Path A: caller submitted a WebAuthn proof (from /api/webauthn/auth/finish)
if webauthn_proof and has_webauthn:
from pegaprox.api.webauthn import consume_webauthn_proof
if not consume_webauthn_proof(username, webauthn_proof):
logging.warning(f"Invalid WebAuthn proof for user: {username} from {client_ip}")
locked = record_failed_attempt(username)
if locked:
resp = jsonify({
'error': 'Invalid credentials',
'locked': True,
'retry_after': lockout_time
})
resp.headers['Retry-After'] = str(lockout_time)
return resp, 401
return jsonify({'error': 'Invalid hardware-key proof'}), 401
# proof OK — skip TOTP entirely
# Path B: caller submitted a TOTP code
elif totp_code and has_totp:
if TOTP_AVAILABLE:
totp = pyotp.TOTP(user['totp_secret'])
if not totp.verify(totp_code, valid_window=1):
logging.warning(f"Invalid 2FA code for user: {username} from {client_ip}")
locked = record_failed_attempt(username)
if locked:
resp = jsonify({
'error': 'Invalid credentials',
'locked': True,
'retry_after': lockout_time
})
resp.headers['Retry-After'] = str(lockout_time)
return resp, 401
return jsonify({'error': 'Invalid 2FA code'}), 401
else:
return jsonify({'error': '2FA is enabled but pyotp is not installed on server'}), 500
# Path C: nothing provided yet — tell the UI what's possible
else:
methods = []
if has_totp: methods.append('totp')
if has_webauthn: methods.append('webauthn')
return jsonify({
'requires_2fa': True,
'message': '2FA required',
'methods': methods,
}), 200
# Clear failed attempts on successful login (both IP and username)
if client_ip in login_attempts_by_ip:
del login_attempts_by_ip[client_ip]
if username in login_attempts_by_user:
del login_attempts_by_user[username]
# NS: Auto-migrate password to Argon2id if using old PBKDF2 format - Jan 2026
# Only rehash for locally-authenticated users - LDAP passwords must NEVER be stored locally
if not ldap_authenticated and needs_password_rehash(user.get('password_salt', ''), user.get('password_hash', '')):
try:
new_salt, new_hash = hash_password(password)
user['password_salt'] = new_salt
user['password_hash'] = new_hash
save_users(users_db)
logging.info(f"Migrated password for user '{username}' to Argon2id (Military Grade)")
except Exception as e:
logging.warning(f"Failed to migrate password for {username}: {e}")
# Create session
remember = data.get('remember', False)
session_id = create_session(username, user['role'], remember=bool(remember))
# Update last login
user['last_login'] = datetime.now().isoformat()
save_users(users_db)
logging.info(f"User '{username}' logged in successfully")
log_audit(username, 'user.login', f"User logged in" + (" (with 2FA)" if user.get('totp_enabled') else ""))
# NS: Mar 2026 - removed auto-allow CORS origin on login (security audit)
# Was allowing any authenticated origin permanently. Use PEGAPROX_ALLOWED_ORIGINS env var instead.
# NS: Get default theme for response
settings = load_server_settings()
default_theme = settings.get('default_theme', 'proxmoxDark')
# NS: Feb 2026 - Check if user needs to set up 2FA (force_2fa setting)
requires_2fa_setup = False
if settings.get('force_2fa') and TOTP_AVAILABLE:
has_2fa = user.get('totp_enabled', False)
is_external = user.get('auth_source', 'local') in ('oidc', 'entra')
is_admin = user.get('role') == ROLE_ADMIN
exclude_admins = settings.get('force_2fa_exclude_admins', False)
if not has_2fa and not is_external and not (is_admin and exclude_admins):
requires_2fa_setup = True
# NS: Debug log for theme sync issues
user_theme = user.get('theme', '') or default_theme
logging.info(f"[LOGIN] User {username} theme from DB: '{user.get('theme', '')}', using: '{user_theme}'")
response = jsonify({
'success': True,
'user': {
'username': username,
'role': user['role'],
'display_name': user.get('display_name', username),
'email': user.get('email', ''),
'avatar_url': user.get('avatar_url', ''),
'auth_source': user.get('auth_source', 'local'), # NS: For LDAP/Entra/OIDC badge
'permissions': get_user_permissions(user), # LW: Frontend can hide/show buttons
'tenant_id': user.get('tenant_id', DEFAULT_TENANT_ID),
'totp_enabled': user.get('totp_enabled', False),
'theme': user_theme, # NS: Use default if empty
'language': user.get('language', ''),
'ui_layout': user.get('ui_layout', 'modern'),
'taskbar_auto_expand': user.get('taskbar_auto_expand', True), # NS: Feb 2026
'sidebar_show_vmid': user.get('sidebar_show_vmid', False), # NS Jul 2026 — corporate sidebar VMIDs
'layout_chosen': user.get('layout_chosen', False),
'portal_only': user.get('portal_only', False)
},
'session_id': session_id,
'portal_only': user.get('portal_only', False),
'default_theme': default_theme, # NS: Include for frontend fallback
'reverse_proxy_enabled': effective_reverse_proxy(settings),
'air_gap_mode': settings.get('air_gap_mode', False),
'requires_2fa_setup': requires_2fa_setup, # NS: Feb 2026 - Force 2FA
# NS: Security warning if using default password
'security_warning': 'DEFAULT_PASSWORD' if (user['role'] == ROLE_ADMIN and password == 'admin') else None,
'requires_password_change': bool(user.get('force_password_change'))
})
# Set session cookie with security flags
# NS: Secure flag only when using HTTPS (important for production!)
from pegaprox.utils.audit import _is_trusted_proxy
is_secure = request.is_secure or (_is_trusted_proxy(request.remote_addr) and request.headers.get('X-Forwarded-Proto') == 'https')
response.set_cookie(
'session_id',
session_id,
httponly=True, # JS cant access this cookie
samesite='Strict', # CSRF protection
secure=is_secure, # only send over HTTPS
max_age=get_session_timeout()
)
return response
# NS Apr 2026 — session self-service. Admins could see all sessions via the users API
# already, but regular users had no way to find "where am I still logged in" or revoke
# a lost browser. These two endpoints scope strictly to the caller's own username.
def _revocation_token_for(sess):
"""Stable per-session opaque token used in /api/user/sessions and DELETE.
Generated lazily and stored on the session itself so subsequent listings
return the same token. MK May 2026 (audit fix M-5) — replaces returning
the full session id as 'full_id', which let JS read otherwise-HttpOnly
credentials and would have been a privilege-escalation vector via XSS."""
import secrets
if not sess.get('revocation_token'):
sess['revocation_token'] = secrets.token_urlsafe(24)
return sess['revocation_token']
@bp.route('/api/user/sessions', methods=['GET'])
def list_own_sessions():
"""Return the caller's active sessions, newest first."""
session_id = request.headers.get('X-Session-ID') or request.cookies.get('session_id')
current = validate_session(session_id)
if not current:
return jsonify({'error': 'not authenticated'}), 401
from pegaprox.utils.auth import active_sessions, sessions_lock
username = current['user']
out = []
with sessions_lock:
for sid, sess in active_sessions.items():
if sess.get('user') != username:
continue
out.append({
# show a short prefix for display only — full sid never leaves
# the cookie. revoke_token is the opaque handle the UI passes
# back to DELETE /api/user/sessions/<token>.
'id': sid[:8] + '…',
'revoke_token': _revocation_token_for(sess),
'ip': sess.get('ip'),
'user_agent': sess.get('user_agent'),
'created_at': sess.get('created_at'),
'last_activity': sess.get('last_activity'),
'remember': bool(sess.get('remember')),
'is_current': sid == session_id,
})
out.sort(key=lambda s: s.get('last_activity') or 0, reverse=True)
return jsonify({'sessions': out, 'total': len(out)})
@bp.route('/api/user/sessions/<token>', methods=['DELETE'])
def revoke_own_session(token):
"""Revoke a session by its opaque revocation token.
MK May 2026 — was previously by-full-sid; now by token (M-5 fix)."""
session_id = request.headers.get('X-Session-ID') or request.cookies.get('session_id')
current = validate_session(session_id)
if not current:
return jsonify({'error': 'not authenticated'}), 401
from pegaprox.utils.auth import active_sessions, sessions_lock
import hmac as _hmac
username = current['user']
target_sid = None
with sessions_lock:
for sid, sess in active_sessions.items():
tok = sess.get('revocation_token') or ''
if tok and _hmac.compare_digest(tok, token) and sess.get('user') == username:
target_sid = sid
break
if not target_sid:
return jsonify({'error': 'session not found'}), 404
invalidate_session(target_sid)
try:
log_audit(username, 'user.session_revoke', f"revoked session {target_sid[:8]}…")
except Exception:
pass
is_self = target_sid == session_id
resp = jsonify({'success': True, 'revoked': target_sid[:8] + '…', 'self_logout': is_self})
if is_self:
resp.delete_cookie('session_id')
return resp
@bp.route('/api/auth/logout', methods=['POST'])
def auth_logout():
"""Logout user and invalidate session"""
session_id = request.headers.get('X-Session-ID') or request.cookies.get('session_id')
if session_id:
session = validate_session(session_id)
if session:
logging.info(f"User '{session['user']}' logged out")
log_audit(session['user'], 'user.logout', f"User logged out")
invalidate_session(session_id)
response = jsonify({'success': True})
response.delete_cookie('session_id')
return response
@bp.route('/api/health', methods=['GET'])
def health_check():
"""Unauthenticated health endpoint for Docker/LB probes"""
return jsonify({'status': 'ok', 'version': PEGAPROX_VERSION})
@bp.route('/api/auth/check', methods=['GET'])
def auth_check():
"""Check if current session is valid"""
session_id = request.headers.get('X-Session-ID') or request.cookies.get('session_id')
session = validate_session(session_id)
if not session:
# NS: Feb 2026 - Include LDAP/OIDC status so login page can show indicators
settings = load_server_settings()
ldap_enabled = settings.get('ldap_enabled', False)
oidc_enabled = settings.get('oidc_enabled', False)
oidc_button_text = settings.get('oidc_button_text', 'Sign in with Microsoft')
login_background = settings.get('login_background', '')
# MK May 2026 — first-run signal so the React shell can show the setup
# wizard instead of the login form on an uninitialised install.
return jsonify({
'authenticated': False,
'initialized': is_initialized(),
'ldap_enabled': ldap_enabled,
'oidc_enabled': oidc_enabled,
'oidc_button_text': oidc_button_text,
'login_background': login_background,
}), 401
# Get user info - always fresh from database
users_db = load_users()
user = users_db.get(session['user'], {})
# NS: Feb 2026 - If user was disabled while session is active, force logout
if not user or not user.get('enabled', True):
logging.info(f"[AUTH_CHECK] User '{session['user']}' is disabled or deleted, invalidating session")
invalidate_session(session_id)
return jsonify({'authenticated': False, 'reason': 'account_disabled'}), 401
logging.debug(f"auth_check for {session['user']}: ui_layout = {user.get('ui_layout')}")
# LW: Check password expiry status - Dec 2025
password_expiry = None
settings = load_server_settings()
# Check if user should be subject to password expiry
# MK: admins are exempt by default, but can opt-in via settings
# NS: LDAP/OIDC users are ALWAYS exempt - they don't have local passwords
is_admin = session['role'] == ROLE_ADMIN
is_external_auth = user.get('auth_source', 'local') in ('ldap', 'oidc', 'entra')
include_admins = settings.get('password_expiry_include_admins', False)
should_check_expiry = settings.get('password_expiry_enabled') and not is_external_auth and (not is_admin or include_admins)
if should_check_expiry:
expiry_days = settings.get('password_expiry_days', 90)
warning_days = settings.get('password_expiry_warning_days', 14)
changed_at = user.get('password_changed_at')
if changed_at:
try:
changed_date = datetime.fromisoformat(changed_at.replace('Z', '+00:00'))
# handle naive datetime
if changed_date.tzinfo:
changed_date = changed_date.replace(tzinfo=None)
days_since = (datetime.now() - changed_date).days
days_until_expiry = expiry_days - days_since
password_expiry = {
'enabled': True,
'days_until_expiry': days_until_expiry,
'expired': days_until_expiry <= 0,
'warning': days_until_expiry <= warning_days and days_until_expiry > 0,
'expiry_days': expiry_days
}
except:
pass # couldn't parse date, skip
else:
# no password_changed_at means old user, treat as expired
password_expiry = {
'enabled': True,
'days_until_expiry': 0,
'expired': True,
'warning': False,
'expiry_days': expiry_days
}
# Get default theme from settings
default_theme = settings.get('default_theme', 'proxmoxDark')
# NS: Debug log for theme sync issues
user_theme = user.get('theme', '') or default_theme
logging.debug(f"[AUTH_CHECK] User {session['user']} theme from DB: '{user.get('theme', '')}', using: '{user_theme}'")
# NS: always check DB for latest role, session might be stale if admin changed it
# This ensures role changes by admin take effect immediately
fresh_role = user.get('role', session['role'])
if fresh_role != session['role']:
# MK: Update session to match DB (avoids stale role in session)
old_role = session['role']
session['role'] = fresh_role
logging.info(f"[AUTH_CHECK] Updated stale session role for {session['user']}: {old_role} → {fresh_role}")
# NS: Get effective permissions for UI visibility
user_permissions = get_user_permissions(user)
# NS: Feb 2026 - Check if user needs to set up 2FA (force_2fa setting)
requires_2fa_setup = False
if settings.get('force_2fa') and TOTP_AVAILABLE:
has_2fa = user.get('totp_enabled', False)
is_external = user.get('auth_source', 'local') in ('oidc', 'entra')
is_admin = fresh_role == ROLE_ADMIN
exclude_admins = settings.get('force_2fa_exclude_admins', False)
# skip OIDC/Entra users (they use their IdP's MFA) and optionally admins
if not has_2fa and not is_external and not (is_admin and exclude_admins):
requires_2fa_setup = True
return jsonify({
'authenticated': True,
'session_id': session_id,
'user': {
'username': session['user'],
'role': fresh_role,
'display_name': user.get('display_name', session['user']),
'email': user.get('email', ''),
'avatar_url': user.get('avatar_url', ''),
'auth_source': user.get('auth_source', 'local'), # NS: For LDAP/Entra/OIDC badge
'tenant_id': user.get('tenant_id', DEFAULT_TENANT_ID), # MK: For multi-tenant UI
'permissions': user_permissions, # LW: So frontend knows what buttons to show
'theme': user_theme,
'language': user.get('language', ''),
'ui_layout': user.get('ui_layout', 'modern'),
'taskbar_auto_expand': user.get('taskbar_auto_expand', True), # NS: Feb 2026
'sidebar_show_vmid': user.get('sidebar_show_vmid', False), # NS Jul 2026 — corporate sidebar VMIDs
'totp_enabled': user.get('totp_enabled', False),
'layout_chosen': user.get('layout_chosen', False),
'portal_only': user.get('portal_only', False)
},
'password_expiry': password_expiry,
'requires_2fa_setup': requires_2fa_setup,
'reverse_proxy_enabled': effective_reverse_proxy(settings),
'air_gap_mode': settings.get('air_gap_mode', False),
'default_theme': default_theme
})
@bp.route('/api/auth/validate', methods=['GET'])
def auth_validate():
"""Simple session validation for WebSocket auth (shell, VNC)
MK: Shell/VNC WebSocket needs a simple endpoint to validate session
Returns 200 if valid, 401 if not
"""
# Check both cookie (sent via requests.get with cookies=) and header
session_id = request.cookies.get('session') or request.cookies.get('session_id') or request.headers.get('X-Session-ID')
if not session_id:
return jsonify({'valid': False, 'error': 'No session'}), 401
session = validate_session(session_id)
if not session:
return jsonify({'valid': False, 'error': 'Invalid session'}), 401
return jsonify({
'valid': True,
'user': session['user'],
'role': session['role']
})
@bp.route('/api/internal/cluster-creds/<cluster_id>', methods=['GET'])
def get_cluster_creds_internal(cluster_id):
"""Internal endpoint for shell/VNC WebSocket to get node connection info
MK: Returns node IPs for SSH connections
For single-node setups, we use the cluster host directly
"""
# Check session from cookie
session_id = request.cookies.get('session') or request.cookies.get('session_id')
if not session_id:
return jsonify({'error': 'No session'}), 401
session = validate_session(session_id)
if not session:
return jsonify({'error': 'Invalid session'}), 401
# MK May 2026 - check_cluster_access reads request.session['user'], which
# @require_auth normally sets. This endpoint does its own cookie-based session
# validation (no decorator), so we have to attach the session to the request
# explicitly. Without this the call below raises AttributeError and the SSH-WS
# shell flow breaks (cluster-creds 500 -> Method 1 falls through -> no node IPs).
try:
request.session = session
except Exception:
pass
# Check if cluster exists
if cluster_id not in cluster_managers:
return jsonify({'error': 'Cluster not found'}), 404
mgr = cluster_managers[cluster_id]
# NS Mar 2026: use standard cluster access check (validates user's cluster assignments + tenant)
from pegaprox.api.helpers import check_cluster_access
ok, err = check_cluster_access(cluster_id)
if not ok:
return err
# Check permissions - NS Feb 2026
users_db = load_users()
user_data = users_db.get(session['user'], {})
# MK 2026-06-10 (RBAC): gate on node.shell only — admin holds it via all-perms, so the
# explicit admin bypass was redundant; a custom role with node.shell now works too.
user_perms = get_user_permissions(user_data)
if 'node.shell' not in user_perms:
logging.warning(f"[CLUSTER-CREDS] User {session['user']} lacks node.shell permission")
return jsonify({'error': 'Permission denied'}), 403
# Get node IPs - the cluster_host is our reliable fallback
node_ips = {}
cluster_host, cluster_port = mgr.host, mgr.api_port
logging.info(f"[CLUSTER-CREDS] Getting node IPs for cluster {cluster_id}, host={cluster_host}")
# NS Mar 2026: XCP-ng path - get IPs from XAPI host records
if getattr(mgr, 'cluster_type', 'proxmox') == 'xcpng':
try:
nodes = mgr.get_nodes() or []
for n in nodes:
nname = n.get('node', '')
if nname:
ip = mgr._get_host_ip(nname)
node_ips[nname] = ip
node_ips[nname.lower()] = ip
logging.info(f"[CLUSTER-CREDS] XCP-ng node {nname} ip={ip}")
except Exception as e:
logging.error(f"[CLUSTER-CREDS] XCP-ng node IP lookup: {e}")
if not node_ips:
node_ips['_default'] = cluster_host
else:
# NS Apr 2026 (PR #324): ?node=<name> resolves only the requested node.
# The WS shell calls this on every connect; resolving every node on a
# big cluster would blow past the WS server's 10s request timeout.
requested_node = (request.args.get('node') or '').strip()
def _resolve_node(nname):
try:
ip = mgr._get_node_ip(nname)
except Exception as e:
logging.warning(f"[CLUSTER-CREDS] _get_node_ip({nname}) raised: {e}")
ip = None
if ip:
node_ips[nname] = ip
node_ips[nname.lower()] = ip
logging.info(f"[CLUSTER-CREDS] {nname} -> {ip}")
else:
logging.warning(f"[CLUSTER-CREDS] no reachable IP for {nname}")
try:
if requested_node:
_resolve_node(requested_node)
else:
# Bulk path -- UIs that want the full mapping.
# Re-use _cached_nodes populated by get_cluster_nodes(),
# otherwise hit /api2/json/nodes directly. mgr.get_nodes()
# doesn't exist on Proxmox managers (XCP-ng only).
nodes = getattr(mgr, '_cached_nodes', None)
if not nodes:
try:
r = mgr._create_session().get(f"https://{cluster_host}:{cluster_port}/api2/json/nodes", timeout=10)
if r.status_code == 200:
nodes = r.json().get('data', []) or []
try:
mgr._cached_nodes = nodes
except Exception:
pass
except Exception as e:
logging.warning(f"[CLUSTER-CREDS] /nodes REST failed: {e}")
nodes = []
for n in (nodes or []):
nn = n.get('node') or n.get('name') or ''
if nn:
_resolve_node(nn)
except Exception as e:
logging.error(f"[CLUSTER-CREDS] Error getting node IPs: {e}")
# Final fallback: if no nodes found, use cluster host
if not node_ips:
node_ips['_default'] = cluster_host
logging.info(f"[CLUSTER-CREDS] No nodes found, using default: {cluster_host}")
# NS Feb 2026: Never expose Proxmox password via API - shell proxy handles auth server-side
# NS Mar 2026: removed user field from response, only SSH proxy needs it internally
# NS Apr 2026: expose ssh_port so the WS server can honor non-22 configs
# MK 2026-06-04: expose verify_pve_tls so the SSH-WS subprocess can decide
# whether to disable cert/hostname verification for the PVE wss:8006/vncwebsocket
# proxy. Mirrors the per-cluster ssl_verify flag (default False because PVE
# ships self-signed certs by default; admins flip it on in cluster settings
# once they've installed a real cert + uploaded the CA).
ssh_port = getattr(getattr(mgr, 'config', None), 'ssh_port', 22) or 22
verify_pve_tls = bool(getattr(mgr, 'ssl_verify', False))
# NS 2026-06-05 (C-1): the termproxy WS proxy gets the PVE session cookie
# from here (server-side) instead of the browser. Mint fresh; None for
# token-only clusters. Other consumers (SSH) ignore the field.
resp = {
'host': cluster_host,
'node_ips': node_ips,
'ssh_port': ssh_port,
'verify_pve_tls': verify_pve_tls,
}
try:
_tk = mgr.mint_console_auth_ticket()
if _tk:
resp['pve_auth_ticket'] = _tk
except Exception:
pass
return jsonify(resp)
@bp.route('/api/auth/verify-password', methods=['POST'])
@require_auth()
def verify_password_api():
"""Verify current user's password — for re-auth before sensitive ops (#256)"""
# MK: security audit — rate limit to prevent brute-force after session compromise
from pegaprox.utils.audit import get_client_ip
client_ip = get_client_ip()
vp_key = f'verify_pw_{client_ip}'
if vp_key in login_attempts_by_ip:
attempts = login_attempts_by_ip[vp_key].get('attempts', [])
recent = [t for t in attempts if time.time() - t < 300]
if len(recent) >= 5:
return jsonify({'error': 'Too many attempts. Try again in 5 minutes.'}), 429
login_attempts_by_ip[vp_key] = {'attempts': recent + [time.time()]}
else:
login_attempts_by_ip[vp_key] = {'attempts': [time.time()]}
data = request.get_json() or {}
password = data.get('password', '')
username = request.session['user']
users = load_users()
user = users.get(username, {})
auth_source = user.get('auth_source', 'local')
# MK Jun 2026 (#587): OIDC/Entra admins have no local password — they re-auth via the
# ladder below (WebAuthn / TOTP / session fallback). The old unconditional
# "Password required" 400 fired before we even knew the auth source, so the
# Re-Configure Cluster flow just hung for SSO admins. Only require it for local/ldap.
if not password and auth_source in ('local', 'ldap'):
return jsonify({'error': 'Password required'}), 400
if auth_source == 'local':
if not user.get('password_hash') or not verify_password(password, user.get('password_salt', ''), user['password_hash']):
return jsonify({'error': 'Invalid password'}), 401
elif auth_source == 'ldap':
from pegaprox.utils.ldap import ldap_authenticate
result = ldap_authenticate(username, password)
if 'error' in result:
return jsonify({'error': 'Invalid password'}), 401
elif auth_source in ('oidc', 'entra'):
# NS May 2026: stolen-cookie replay risk — the original "session valid → accept"
# made re-auth a no-op for OIDC users. attacker with the cookie passes the gate
# for sensitive ops (delete cluster, rotate creds, etc).
# NS May 2026 (v0.9.9.1 follow-up) — TOTP step-up was wrong as a hard
# requirement: PegaProx blocks OIDC users from enrolling TOTP at /2fa/setup
# ("2FA is managed by your OIDC provider"), so a strict TOTP gate locks
# OIDC admins out of re-auth-gated ops entirely. Correct ladder:
# 1. WebAuthn proof (works regardless of auth_source — independent enrolment)
# 2. TOTP if the user happens to have one (e.g. ex-local user converted to OIDC)
# 3. Fall back to session-validity acceptance + warning log so the
# admin can see the soft state in the audit trail.
# Properly fixing #3 requires an OIDC `prompt=login` redirect flow which
# is a bigger piece of work — tracked separately.
if user.get('role') != 'admin':
return jsonify({'error': 'Admin access required'}), 403
# Path A: WebAuthn proof token (preferred for OIDC users)
webauthn_proof = (data.get('webauthn_proof') or '')[:256]
try:
db = get_db()
wa_row = db.query('SELECT COUNT(*) AS n FROM webauthn_credentials WHERE username = ?', (username,))
has_webauthn = bool(wa_row and wa_row[0]['n'] > 0)
except Exception:
has_webauthn = False
if has_webauthn:
if not webauthn_proof:
return jsonify({
'error': 'WebAuthn verification required for re-authentication',
'requires_webauthn': True,
}), 401
from pegaprox.api.webauthn import consume_webauthn_proof
if not consume_webauthn_proof(username, webauthn_proof):
logging.warning(f"[AUTH] OIDC re-auth bad WebAuthn proof for admin '{username}'")
return jsonify({'error': 'Invalid WebAuthn proof'}), 401
logging.info(f"[AUTH] OIDC re-auth via WebAuthn for admin '{username}'")
# Path B: TOTP if available (ex-local users still carry a totp_secret)
elif user.get('totp_secret'):
totp_code = (data.get('totp_code') or '').strip()
if not totp_code:
return jsonify({
'error': 'TOTP code required for re-authentication',
'requires_totp': True,
}), 401
try:
import pyotp
if not pyotp.TOTP(user['totp_secret']).verify(totp_code, valid_window=1):
logging.warning(f"[AUTH] OIDC re-auth bad TOTP for admin '{username}'")
return jsonify({'error': 'Invalid TOTP code'}), 401
except Exception as e:
logging.error(f"[AUTH] TOTP verify error for '{username}': {e}")
return jsonify({'error': 'TOTP verification failed'}), 500
logging.info(f"[AUTH] OIDC re-auth via TOTP for admin '{username}'")
# Path C: nothing local enrolled — accept on session validity, but log it
# loudly. OIDC admins who care about replay-resistance should register a
# WebAuthn credential (Settings → Account → Security keys).
else:
logging.warning(
f"[AUTH] OIDC re-auth on session-validity only for admin '{username}' "
"(no WebAuthn / TOTP enrolled — recommend WebAuthn enrolment)"
)
else:
return jsonify({'error': 'Password verification not supported for this account type'}), 400
return jsonify({'success': True})
@bp.route('/api/auth/change-password', methods=['POST'])
@require_auth()
def auth_change_password():
"""Change current user's password
NS: Security feature - invalidates all other sessions after password change
This way if someone stole your session, changing password kicks them out
"""
global users_db
data = request.get_json()
current_password = data.get('current_password', '')
new_password = data.get('new_password', '')
if not current_password or not new_password:
return jsonify({'error': 'Current and new password required'}), 400
# Validate password policy
is_valid, error_msg = validate_password_policy(new_password)
if not is_valid:
return jsonify({'error': error_msg}), 400
username = request.session['user']
# NS: rate limit this - someone with a stolen session could brute-force the current password
if not check_auth_action_rate_limit(f'pwd_change:{username}', max_attempts=5, window=300):
return jsonify({'error': 'Too many password change attempts. Try again in 5 minutes.'}), 429
users_db = load_users()
if username not in users_db:
return jsonify({'error': 'User not found'}), 404
user = users_db[username]
# NS: LDAP/OIDC users cannot change their password here - they must change it in their identity provider
if user.get('auth_source', 'local') in ('ldap', 'oidc', 'entra'):
provider_name = {'ldap': 'LDAP/Active Directory', 'oidc': 'your OIDC provider', 'entra': 'Microsoft Entra ID'}.get(user['auth_source'], 'your identity provider')
return jsonify({'error': f'Password is managed by {provider_name}. Please change it there.'}), 400
# Verify current password
if not verify_password(current_password, user['password_salt'], user['password_hash']):
log_audit(username, 'user.password_change_failed', 'Incorrect current password')
return jsonify({'error': 'Current password is incorrect'}), 401
# Update password
salt, password_hash = hash_password(new_password)
user['password_salt'] = salt
user['password_hash'] = password_hash
user['password_changed_at'] = datetime.now().isoformat() # LW: reset expiry timer
# Clear forced password change flag (#144)
if user.get('force_password_change'):
user['force_password_change'] = False
# Mark admin initialized if this is the default admin
if user.get('is_default'):
user['is_default'] = False
mark_admin_initialized()
save_users(users_db)
# NS 2026-04-24 — security audit finding #1: invalidate ALL sessions including the
# caller's current one. If a stolen cookie is the one changing the password, we don't
# want it to keep access. Frontend reads relogin_required and redirects to /login.
current_session_id = request.cookies.get('session_id') or request.headers.get('X-Session-ID')
sessions_removed = invalidate_all_user_sessions(username) # no except_session — all go
logging.info(f"User '{username}' changed their password — all sessions invalidated")
log_audit(username, 'user.password_changed', f"Password changed, {sessions_removed} session(s) invalidated (incl. current)")
resp = jsonify({
'success': True,
'sessions_invalidated': sessions_removed,
'relogin_required': True, # UI hint
})
# drop the session cookie so the browser stops sending the now-dead SID
resp.delete_cookie('session_id')
return resp
# ============================================
# TOTP 2FA API Routes
# ============================================
@bp.route('/api/auth/2fa/setup', methods=['POST'])
@require_auth()
def setup_2fa():
"""Generate TOTP secret and QR code for 2FA setup"""
global users_db
if not TOTP_AVAILABLE:
return jsonify({'error': '2FA not available. Please install pyotp and qrcode: pip install pyotp qrcode[pil]'}), 500
username = request.session['user']
logging.info(f"2FA setup requested for user: {username}") # MK: Debug
users_db = load_users()
if username not in users_db:
logging.warning(f"2FA setup failed - user not found: {username}")
return jsonify({'error': 'User not found'}), 404
user = users_db[username]
# NS: OIDC/Entra users should use their IdP's MFA, not PegaProx 2FA
# PegaProx 2FA only works for login form (LDAP + local), not OIDC redirect flow
if user.get('auth_source', 'local') in ('oidc', 'entra'):
provider_name = 'Microsoft Entra ID' if user.get('auth_source') == 'entra' else 'your OIDC provider'
return jsonify({'error': f'2FA is managed by {provider_name}. Please enable MFA there instead.'}), 400
# Generate new secret
secret = pyotp.random_base32()
# Store pending secret (not activated yet)
user['totp_pending_secret'] = secret
save_users(users_db)
logging.info(f"2FA setup: saved pending secret for user {username}") # MK: Debug
# Generate provisioning URI
totp = pyotp.TOTP(secret)
uri = totp.provisioning_uri(name=username, issuer_name='PegaProx')
# Generate QR code as base64
qr = qrcode.QRCode(version=1, box_size=10, border=5)
qr.add_data(uri)
qr.make(fit=True)
img = qr.make_image(fill_color="black", back_color="white")
buffer = _io.BytesIO()
img.save(buffer, format='PNG')
qr_base64 = base64.b64encode(buffer.getvalue()).decode()
return jsonify({
'secret': secret,
'qr_code': f'data:image/png;base64,{qr_base64}',
'uri': uri
})
@bp.route('/api/auth/2fa/verify', methods=['POST'])
@require_auth()
def verify_2fa_setup():
"""Verify TOTP code and activate 2FA"""
global users_db
if not TOTP_AVAILABLE:
return jsonify({'error': '2FA not available'}), 500
data = request.get_json()
code = data.get('code', '') if data else ''
if not code:
logging.warning("2FA verify: no code provided") # MK: Debug
return jsonify({'error': 'TOTP code required'}), 400
username = request.session['user']
# NS: only 1M possible 6-digit codes, easy to brute force without this
# NS: tightened from 5/300s to 3/120s — reduces brute-force success probability
if not check_auth_action_rate_limit(f'totp_verify:{username}', max_attempts=3, window=120):
return jsonify({'error': 'Too many verification attempts. Try again in 2 minutes.'}), 429
logging.info(f"2FA verify requested for user: {username}, code length: {len(code)}") # MK: Debug
users_db = load_users()
if username not in users_db:
logging.warning(f"2FA verify: user not found: {username}")
return jsonify({'error': 'User not found'}), 404
user = users_db[username]
pending_secret = user.get('totp_pending_secret')
if not pending_secret:
logging.warning(f"2FA verify: no pending secret for user {username}. User keys: {list(user.keys())}") # MK: Debug
return jsonify({'error': 'No pending 2FA setup'}), 400
# Verify the code
totp = pyotp.TOTP(pending_secret)
if not totp.verify(code, valid_window=1):
return jsonify({'error': 'Invalid TOTP code'}), 401
# Activate 2FA
user['totp_secret'] = pending_secret
user['totp_enabled'] = True
del user['totp_pending_secret']
save_users(users_db)
logging.info(f"User '{username}' enabled 2FA")
log_audit(username, '2fa.enabled', "User enabled 2FA")
return jsonify({'success': True, 'message': '2FA erfolgreich aktiviert'})
@bp.route('/api/auth/2fa/disable', methods=['POST'])
@require_auth()
def disable_2fa():
"""Disable 2FA for current user"""
global users_db
data = request.get_json()
password = data.get('password', '')
if not password:
return jsonify({'error': 'Password required to disable 2FA'}), 400
username = request.session['user']
# NS: same issue as pwd change - stolen session + unlimited guesses = bad
if not check_auth_action_rate_limit(f'2fa_disable:{username}', max_attempts=5, window=300):
return jsonify({'error': 'Too many attempts. Try again in 5 minutes.'}), 429
users_db = load_users()
if username not in users_db:
return jsonify({'error': 'User not found'}), 404
user = users_db[username]
# NS: OIDC/Entra users manage MFA through their IdP - shouldn't have PegaProx 2FA
if user.get('auth_source', 'local') in ('oidc', 'entra'):
return jsonify({'error': '2FA is managed by your identity provider'}), 400
# NS: Feb 2026 - LDAP users verify against LDAP (they have no local password hash)
if user.get('auth_source') == 'ldap':
ldap_result = ldap_authenticate(username, password)
if not ldap_result.get('success'):
return jsonify({'error': 'Invalid LDAP password'}), 401
else:
# Local users verify against local password hash
if not verify_password(password, user['password_salt'], user['password_hash']):
return jsonify({'error': 'Invalid password'}), 401
# Disable 2FA
user['totp_enabled'] = False
user.pop('totp_secret', None)
user.pop('totp_pending_secret', None)
save_users(users_db)
logging.info(f"User '{username}' disabled 2FA")
log_audit(username, '2fa.disabled', "User disabled 2FA")
return jsonify({'success': True, 'message': '2FA disabled'})
@bp.route('/api/auth/2fa/status', methods=['GET'])
@require_auth()
def get_2fa_status():
"""Get 2FA status for current user"""
username = request.session['user']
users_db = load_users()
if username not in users_db:
return jsonify({'error': 'User not found'}), 404
user = users_db[username]
return jsonify({
'enabled': user.get('totp_enabled', False),
'available': TOTP_AVAILABLE
})
# =====================================================
# USER PREFERENCES - LW
# Per-user settings (theme, language)
# =====================================================
# =============================================================================
# NS: Feb 2026 - API Token Management Endpoints
# MK: Users can create tokens for CI/CD, scripts, monitoring integrations
# LW: Admins can see all tokens, users can only manage their own
# =============================================================================
@bp.route('/api/auth/tokens', methods=['GET'])
@require_auth()
def list_api_tokens():
"""List API tokens for current user (or all users for admin)"""
ensure_api_tokens_table()
username = request.session['user']
role = request.session.get('role', ROLE_VIEWER)
# MK 2026-06-10 (RBAC): the admin.api perm (not the admin role) sees all tokens — admin holds it via all-perms.
from pegaprox.utils.rbac import has_permission as _has_perm
if _has_perm(load_users().get(username, {}), 'admin.api') and request.args.get('all') == 'true':
try:
db = get_db()
cursor = db.conn.cursor()
cursor.execute('''
SELECT id, token_prefix, username, name, role, permissions, expires_at,
last_used_at, last_used_ip, created_at, revoked
FROM api_tokens ORDER BY created_at DESC
''')
tokens = [dict(row) for row in cursor.fetchall()]
return jsonify({'tokens': tokens})
except Exception as e:
return jsonify({'error': safe_error(e, 'Failed to list tokens')}), 500
tokens = list_user_tokens(username)
return jsonify({'tokens': tokens})
@bp.route('/api/auth/tokens', methods=['POST'])
@require_auth()
def create_api_token_endpoint():
"""Create a new API token for the current user"""
username = request.session['user']
data = request.get_json() or {}
token_name = data.get('name', '').strip()
if not token_name:
return jsonify({'error': 'Token name is required'}), 400
if len(token_name) > 64:
return jsonify({'error': 'Token name too long (max 64 chars)'}), 400
# LW: Check for duplicate names
existing = list_user_tokens(username)
active_names = [t['name'] for t in existing if not t.get('revoked')]
if token_name in active_names:
return jsonify({'error': f'Token name "{token_name}" already exists'}), 400
# NS: Max 10 active tokens per user
active_count = sum(1 for t in existing if not t.get('revoked'))
if active_count >= 1:
return jsonify({'error': 'You already have an active token. Revoke it first to create a new one.'}), 400
role = data.get('role')
expires_days = data.get('expires_days')
if expires_days is not None:
try:
expires_days = int(expires_days)
if expires_days < 1 or expires_days > 365:
return jsonify({'error': 'Expiry must be between 1 and 365 days'}), 400
except (ValueError, TypeError):
return jsonify({'error': 'Invalid expires_days value'}), 400
result = create_api_token(username, token_name, role=role, expires_days=expires_days)
if 'error' in result:
return jsonify(result), 400
# MK: Audit log
log_audit(username, 'token.created', f"API token '{token_name}' created")
return jsonify(result)
@bp.route('/api/auth/tokens/<int:token_id>', methods=['DELETE'])
@require_auth()
def revoke_api_token_endpoint(token_id):
"""Revoke an API token"""
username = request.session['user']
role = request.session.get('role', ROLE_VIEWER)
# MK 2026-06-10 (RBAC): the admin.api perm can revoke any token (admin holds it via all-perms).
from pegaprox.utils.rbac import has_permission as _has_perm
if _has_perm(load_users().get(username, {}), 'admin.api'):
try:
db = get_db()
cursor = db.conn.cursor()
cursor.execute('SELECT username, name FROM api_tokens WHERE id = ?', (token_id,))
row = cursor.fetchone()
if row:
cursor.execute('UPDATE api_tokens SET revoked = 1 WHERE id = ?', (token_id,))
db.conn.commit()
token_owner = dict(row)['username']
token_name = dict(row)['name']
log_audit(username, 'token.revoked', f"Revoked API token '{token_name}' (user: {token_owner})")
return jsonify({'success': True})
return jsonify({'error': 'Token not found'}), 404
except Exception as e:
return jsonify({'error': safe_error(e, 'Failed to revoke token')}), 500
if revoke_api_token(token_id, username):
log_audit(username, 'token.revoked', f"Revoked API token id={token_id}")
return jsonify({'success': True})
return jsonify({'error': 'Token not found or not owned by you'}), 404
# GET /api/user/preferences is in users.py (has full implementation)
Reference in New Issue View Git Blame Copy Permalink