fix(security): authenticate VNC/terminal WebSockets and harden SSH/config
Some checks failed
Backend Tests / pytest (Python 3.12) (push) Failing after 46s

Critical security and robustness fixes from a code audit:

- WS terminal/VNC now require a JWT (?token=), validate the session,
  vms.console permission and VM ownership before bridging to Proxmox
  (previously the LXC terminal gave unauthenticated root shell access).
  Frontend ConsolePage passes the token in both WS URLs.
- Fix SSH fallback for `pvecm delnode`: SSHClient now supports password
  auth (it previously crashed with TypeError); ProxmoxClient stores the
  password for SSH fallbacks and the host port suffix is stripped.
- Add production startup warnings for default ADMIN_PASSWORD and wildcard
  CORS_ORIGINS.
- dashboard: use tz-aware utcnow() instead of naive datetime.utcnow().
- crypto: narrow decrypt_value except to InvalidToken and log failures.
- ssh_client: guard the shared connection cache with a lock; make
  unknown-host-key rejection opt-in via SSH_REJECT_UNKNOWN_HOSTS.
- logging: send loguru console output to stderr instead of stdout.
- Sync version fallback to 1.5.1; remove one-off refactor scripts.
- Add CODE_REVIEW.md audit report.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-06-13 06:31:26 +05:00
parent bfd92102cf
commit 50f2ee1966
17 changed files with 394 additions and 503 deletions

View File

@@ -135,6 +135,42 @@ def get_current_session_token(
return payload.get("session")
def authenticate_ws_token(token: Optional[str], db: Session) -> Optional[User]:
"""Authenticate a WebSocket connection from a JWT passed as a query parameter.
Browsers cannot set Authorization headers on WebSocket handshakes, so the
token is supplied via ?token=<jwt>. Returns the authenticated, active User
(with a valid session, if the token is session-bound) or ``None`` on any
failure. Unlike ``decode_access_token`` this never raises — callers close
the socket with an appropriate code instead.
"""
if not token:
return None
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
except jwt.PyJWTError:
return None
username = payload.get("sub")
if not username:
return None
# Validate the bound session, mirroring get_current_user.
session_token = payload.get("session")
if session_token:
from .services.security_service import SecurityService
if not SecurityService.validate_session(db, session_token):
return None
user = db.query(User).options(joinedload(User.role)).filter(User.username == username).first()
if user is None or not user.is_active:
return None
if user.is_locked():
return None
return user
def get_current_active_admin(current_user: User = Depends(get_current_user)) -> User:
"""Get current active admin user"""
if not current_user.is_admin and (not current_user.role or current_user.role.name != "admin"):