""" Sudo session management for admin OTP authentication. This module manages the elevated privilege state ("sudo mode") after successful OTP verification. The sudo session is stored in Django's session framework and is bound to the user's IP address for security. """ from __future__ import annotations import logging from typing import TYPE_CHECKING from django.conf import settings from django.utils import timezone from labs.utils.network import get_client_ip if TYPE_CHECKING: from django.http import HttpRequest logger = logging.getLogger(__name__) # Session keys SESSION_KEY_SUDO_UNTIL = "admin_sudo_until" SESSION_KEY_SUDO_IP = "admin_sudo_ip" SESSION_KEY_SUDO_USER_ID = "admin_sudo_user_id" def get_sudo_duration() -> int: """Get the configured sudo session duration in seconds.""" return getattr(settings, "ADMIN_SUDO_SESSION_DURATION", 3 * 60 * 60) # 3 hours def grant_sudo(request: HttpRequest) -> None: """ Grant sudo privileges to the current session. This sets a timestamp in the session indicating when the sudo session expires, along with the user's IP address for binding. Security measures: - Binds sudo session to user ID and IP address - Cycles session key to prevent session fixation attacks Args: request: The current HTTP request. """ user = request.user ip, _ = get_client_ip(request) duration = get_sudo_duration() expires_at = timezone.now() + timezone.timedelta(seconds=duration) # Cycle the session key to prevent session fixation attacks. # This invalidates the old session ID and creates a new one, # while preserving the session data. request.session.cycle_key() request.session[SESSION_KEY_SUDO_UNTIL] = expires_at.isoformat() request.session[SESSION_KEY_SUDO_IP] = ip or "" request.session[SESSION_KEY_SUDO_USER_ID] = user.id logger.info( "Sudo session granted for user %s (id=%d) from IP %s, expires at %s", user.email, user.id, ip, expires_at.isoformat(), ) def revoke_sudo(request: HttpRequest) -> None: """ Revoke sudo privileges from the current session. Args: request: The current HTTP request. """ user = request.user request.session.pop(SESSION_KEY_SUDO_UNTIL, None) request.session.pop(SESSION_KEY_SUDO_IP, None) request.session.pop(SESSION_KEY_SUDO_USER_ID, None) if user.is_authenticated: logger.info( "Sudo session revoked for user %s (id=%d)", user.email, user.id, ) def is_sudo_active(request: HttpRequest) -> bool: """ Check if the current session has active sudo privileges. The sudo session is considered active if: 1. The session has a valid sudo_until timestamp that hasn't expired 2. The current user matches the user who obtained sudo 3. The current IP address matches the IP when sudo was granted Args: request: The current HTTP request. Returns: True if sudo is active and valid, False otherwise. """ user = request.user if not user.is_authenticated: return False # Check if sudo session exists sudo_until_str = request.session.get(SESSION_KEY_SUDO_UNTIL) if not sudo_until_str: return False # Check if sudo belongs to current user sudo_user_id = request.session.get(SESSION_KEY_SUDO_USER_ID) if sudo_user_id != user.id: # Session user changed, invalidate sudo revoke_sudo(request) return False # Check if sudo has expired try: sudo_until = timezone.datetime.fromisoformat(sudo_until_str) if timezone.is_naive(sudo_until): sudo_until = timezone.make_aware(sudo_until) except (ValueError, TypeError): revoke_sudo(request) return False if timezone.now() > sudo_until: logger.info( "Sudo session expired for user %s (id=%d)", user.email, user.id, ) revoke_sudo(request) return False # Check IP binding sudo_ip = request.session.get(SESSION_KEY_SUDO_IP) current_ip, _ = get_client_ip(request) if sudo_ip and current_ip and sudo_ip != current_ip: logger.warning( "Sudo session invalidated due to IP change for user %s (id=%d): %s -> %s", user.email, user.id, sudo_ip, current_ip, ) revoke_sudo(request) return False return True def get_sudo_remaining_seconds(request: HttpRequest) -> int | None: """ Get the remaining seconds until sudo expires. Args: request: The current HTTP request. Returns: Remaining seconds, or None if sudo is not active. """ if not is_sudo_active(request): return None sudo_until_str = request.session.get(SESSION_KEY_SUDO_UNTIL) if not sudo_until_str: return None try: sudo_until = timezone.datetime.fromisoformat(sudo_until_str) if timezone.is_naive(sudo_until): sudo_until = timezone.make_aware(sudo_until) remaining = (sudo_until - timezone.now()).total_seconds() return max(0, int(remaining)) except (ValueError, TypeError): return None def extend_sudo(request: HttpRequest) -> bool: """ Extend the current sudo session by the configured duration. This can only be done while sudo is still active. Args: request: The current HTTP request. Returns: True if the session was extended, False if sudo wasn't active. """ if not is_sudo_active(request): return False duration = get_sudo_duration() expires_at = timezone.now() + timezone.timedelta(seconds=duration) request.session[SESSION_KEY_SUDO_UNTIL] = expires_at.isoformat() user = request.user logger.info( "Sudo session extended for user %s (id=%d), new expiry: %s", user.email, user.id, expires_at.isoformat(), ) return True