Create session_manager.py

session_manager.py

@@ -0,0 +1,249 @@
+"""
+session_manager.py - Session Management for User Authentication
+Handles session token generation, validation, and expiry
+"""
+
+import uuid
+import hashlib
+import secrets
+from datetime import datetime, timedelta
+import logging
+from typing import Optional, Dict, Any
+
+logger = logging.getLogger(__name__)
+
+# ===========================================
+# CONFIGURATION
+# ===========================================
+
+# Session expiry (30 days)
+SESSION_EXPIRY_DAYS = 30
+
+# In-memory session store (for demo - use database in production)
+# Structure: {session_token: {'user_id': user_id, 'expires_at': timestamp, 'created_at': timestamp}}
+_session_store: Dict[str, Dict[str, Any]] = {}
+
+
+# ===========================================
+# SESSION MANAGEMENT
+# ===========================================
+
+def generate_session_token() -> str:
+    """
+    Generate a secure random session token.
+    """
+    # Use secrets for cryptographically secure random token
+    return secrets.token_urlsafe(32)
+
+
+def create_session(user_id: str) -> Dict[str, Any]:
+    """
+    Create a new session for a user.
+    Returns session token and expiry information.
+    """
+    session_token = generate_session_token()
+    expires_at = datetime.now() + timedelta(days=SESSION_EXPIRY_DAYS)
+    
+    _session_store[session_token] = {
+        'user_id': user_id,
+        'expires_at': expires_at.timestamp(),
+        'created_at': datetime.now().timestamp(),
+        'last_accessed': datetime.now().timestamp()
+    }
+    
+    logger.debug(f"Created session for user {user_id}")
+    
+    return {
+        'session_token': session_token,
+        'expires_at': expires_at.isoformat(),
+        'expires_in_days': SESSION_EXPIRY_DAYS
+    }
+
+
+def validate_session(session_token: str) -> Optional[str]:
+    """
+    Validate a session token and return user_id if valid.
+    Also refreshes the session expiry if within refresh window.
+    """
+    if session_token not in _session_store:
+        return None
+    
+    session_data = _session_store[session_token]
+    expires_at = datetime.fromtimestamp(session_data['expires_at'])
+    
+    # Check if expired
+    if datetime.now() > expires_at:
+        # Session expired, remove it
+        delete_session(session_token)
+        logger.debug(f"Session expired: {session_token[:16]}...")
+        return None
+    
+    # Update last accessed time
+    session_data['last_accessed'] = datetime.now().timestamp()
+    
+    # Refresh session if within 7 days of expiry (rolling expiry)
+    # This implements the 30-day rolling requirement
+    days_remaining = (expires_at - datetime.now()).days
+    if days_remaining < 7:
+        # Extend by another 30 days
+        new_expiry = datetime.now() + timedelta(days=SESSION_EXPIRY_DAYS)
+        session_data['expires_at'] = new_expiry.timestamp()
+        logger.debug(f"Refreshed session for user {session_data['user_id']}")
+    
+    return session_data['user_id']
+
+
+def delete_session(session_token: str) -> bool:
+    """
+    Delete a session (logout).
+    """
+    if session_token in _session_store:
+        user_id = _session_store[session_token]['user_id']
+        del _session_store[session_token]
+        logger.debug(f"Deleted session for user {user_id}")
+        return True
+    return False
+
+
+def delete_all_user_sessions(user_id: str) -> int:
+    """
+    Delete all sessions for a user (e.g., when changing password).
+    Returns number of sessions deleted.
+    """
+    deleted = 0
+    tokens_to_delete = []
+    
+    for token, data in _session_store.items():
+        if data['user_id'] == user_id:
+            tokens_to_delete.append(token)
+    
+    for token in tokens_to_delete:
+        del _session_store[token]
+        deleted += 1
+    
+    if deleted > 0:
+        logger.info(f"Deleted {deleted} sessions for user {user_id}")
+    
+    return deleted
+
+
+def get_session_info(session_token: str) -> Optional[Dict[str, Any]]:
+    """
+    Get information about a session (without validating).
+    """
+    if session_token not in _session_store:
+        return None
+    
+    data = _session_store[session_token]
+    return {
+        'user_id': data['user_id'],
+        'expires_at': datetime.fromtimestamp(data['expires_at']).isoformat(),
+        'created_at': datetime.fromtimestamp(data['created_at']).isoformat(),
+        'last_accessed': datetime.fromtimestamp(data['last_accessed']).isoformat()
+    }
+
+
+def get_active_session_count(user_id: str) -> int:
+    """
+    Get the number of active sessions for a user.
+    """
+    count = 0
+    for token, data in _session_store.items():
+        if data['user_id'] == user_id:
+            # Check if still valid
+            expires_at = datetime.fromtimestamp(data['expires_at'])
+            if datetime.now() <= expires_at:
+                count += 1
+    return count
+
+
+def cleanup_expired_sessions() -> int:
+    """
+    Remove all expired sessions from memory.
+    Returns number of sessions removed.
+    """
+    now = datetime.now()
+    expired = []
+    
+    for token, data in _session_store.items():
+        expires_at = datetime.fromtimestamp(data['expires_at'])
+        if now > expires_at:
+            expired.append(token)
+    
+    for token in expired:
+        del _session_store[token]
+    
+    if expired:
+        logger.debug(f"Cleaned up {len(expired)} expired sessions")
+    
+    return len(expired)
+
+
+# ===========================================
+# PASSWORD HASHING (for user management)
+# ===========================================
+
+def hash_password(password: str, salt: str = None) -> tuple:
+    """
+    Hash a password with salt.
+    Returns (hashed_password, salt)
+    """
+    if salt is None:
+        salt = secrets.token_hex(16)
+    
+    # Use PBKDF2 for secure password hashing
+    hashed = hashlib.pbkdf2_hmac(
+        'sha256',
+        password.encode('utf-8'),
+        salt.encode('utf-8'),
+        100000  # Number of iterations
+    ).hex()
+    
+    return hashed, salt
+
+
+def verify_password(password: str, hashed_password: str, salt: str) -> bool:
+    """
+    Verify a password against its hash and salt.
+    """
+    test_hash, _ = hash_password(password, salt)
+    return test_hash == hashed_password
+
+
+# ===========================================
+# USER MANAGEMENT HELPERS
+# ===========================================
+
+def generate_user_id() -> str:
+    """
+    Generate a unique user ID.
+    """
+    return str(uuid.uuid4())
+
+
+def is_valid_email(email: str) -> bool:
+    """
+    Basic email validation.
+    """
+    import re
+    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
+    return re.match(pattern, email) is not None
+
+
+def is_valid_username(username: str) -> bool:
+    """
+    Validate username (alphanumeric + underscore, 3-30 chars).
+    """
+    import re
+    pattern = r'^[a-zA-Z0-9_]{3,30}$'
+    return re.match(pattern, username) is not None
+
+
+# ===========================================
+# INITIALIZATION
+# ===========================================
+
+# Run cleanup on module load (optional)
+cleanup_expired_sessions()
+
+logger.info("Session manager initialized")