#Django会话管理 - 用户状态追踪与安全管理

"""
会话工作原理：

1. 用户首次访问时，服务器创建会话并分配唯一ID
2. 服务器将会话ID通过Cookie等方式发送给客户端
3. 客户端后续请求携带会话ID
4. 服务器根据会话ID识别用户并恢复状态
5. 会话数据存储在服务器端，客户端只持有ID

会话优势：
- 维护用户状态
- 提供个性化体验
- 支持用户认证
- 实现购物车等功能

会话挑战：
- 安全性问题
- 性能影响
- 跨域处理
- 会话固定攻击
"""

"""
会话生命周期：

1. 会话创建
   - 用户首次访问
   - 明确调用session操作
   - 设置session数据

2. 会话活跃
   - 定期访问维持会话
   - 更新会话数据
   - 检查会话有效性

3. 会话过期
   - 达到过期时间
   - 用户主动退出
   - 管理员强制清除

4. 会话清理
   - 删除会话数据
   - 清理相关资源
   - 记录日志
"""

"""
Django支持的会话存储方式：

1. 数据库会话 (db)
   - 存储在数据库中
   - 适合中小型应用
   - 需要定期清理

2. 缓存会话 (cache)
   - 存储在缓存系统中
   - 性能较好
   - 可能丢失数据

3. 缓存+数据库 (cached_db)
   - 优先使用缓存
   - 回退到数据库
   - 平衡性能和可靠性

4. 文件会话 (file)
   - 存储在文件系统中
   - 适合单机部署
   - 性能一般

5. 签名Cookie会话 (signed_cookies)
   - 数据存储在Cookie中
   - 服务器验证签名
   - 4KB大小限制
"""

# 会话中间件
from django.contrib.sessions.middleware import SessionMiddleware
from django.utils.deprecation import MiddlewareMixin

class CustomSessionMiddleware(SessionMiddleware):
    """自定义会话中间件"""
    
    def process_request(self, request):
        """处理请求"""
        super().process_request(request)
        
        # 记录会话相关信息
        session_id = request.session.session_key
        if session_id:
            # 可以记录会话活动
            self.log_session_activity(session_id, request)
    
    def process_response(self, request, response):
        """处理响应"""
        # 检查是否需要保存会话
        if hasattr(request, 'session') and request.session.modified:
            # 设置额外的安全头
            response = self.add_security_headers(response)
        
        return super().process_response(request, response)
    
    def log_session_activity(self, session_id, request):
        """记录会话活动"""
        import logging
        logger = logging.getLogger('session.activity')
        logger.info(f"Session activity: {session_id} - {request.path}")
    
    def add_security_headers(self, response):
        """添加安全头"""
        response['X-Session-ID-Present'] = 'true'
        return response

# 会话引擎
from django.contrib.sessions.backends.db import SessionStore as DBSessionStore
from django.contrib.sessions.backends.cache import SessionStore as CacheSessionStore
from django.contrib.sessions.backends.cached_db import SessionStore as CachedDBSessionStore
from django.contrib.sessions.backends.file import SessionStore as FileSessionStore
from django.contrib.sessions.backends.signed_cookies import SessionStore as CookieSessionStore

"""
Django会话引擎：

1. DBSessionStore - 数据库存储
2. CacheSessionStore - 缓存存储
3. CachedDBSessionStore - 缓存+数据库
4. FileSessionStore - 文件存储
5. CookieSessionStore - 签名Cookie存储
"""

# Django会话API使用
from django.contrib.sessions.models import Session
from django.contrib.sessions.backends.db import SessionStore
from django.utils import timezone
import json

# 基本会话操作
def session_operations_demo(request):
    """会话操作演示"""
    
    # 设置会话数据
    request.session['user_id'] = 123
    request.session['username'] = 'john_doe'
    request.session['preferences'] = {
        'theme': 'dark',
        'language': 'zh-CN'
    }
    
    # 获取会话数据
    user_id = request.session.get('user_id', 0)
    username = request.session.get('username', 'guest')
    preferences = request.session.get('preferences', {})
    
    # 检查键是否存在
    if 'cart_items' in request.session:
        cart_count = len(request.session['cart_items'])
    
    # 删除会话数据
    if 'temp_data' in request.session:
        del request.session['temp_data']
    
    # 清空会话
    # request.session.flush()
    
    # 设置会话过期
    request.session.set_expiry(3600)  # 1小时后过期
    
    return {
        'user_id': user_id,
        'username': username,
        'preferences': preferences
    }

# 会话工具类
class SessionUtils:
    """会话工具类"""
    
    @staticmethod
    def get_user_session_data(request, user_id):
        """获取用户会话数据"""
        session_key = request.session.session_key
        if not session_key:
            return {}
        
        try:
            session = Session.objects.get(session_key=session_key)
            session_data = session.get_decoded()
            return session_data
        except Session.DoesNotExist:
            return {}
    
    @staticmethod
    def set_user_preferences(request, preferences):
        """设置用户偏好"""
        request.session['user_preferences'] = preferences
        request.session.modified = True
    
    @staticmethod
    def get_user_preferences(request):
        """获取用户偏好"""
        return request.session.get('user_preferences', {})
    
    @staticmethod
    def clear_user_data(request, exclude_keys=None):
        """清除用户数据（保留必要键）"""
        exclude_keys = exclude_keys or []
        session_keys = list(request.session.keys())
        
        for key in session_keys:
            if key not in exclude_keys:
                del request.session[key]
    
    @staticmethod
    def regenerate_session_id(request):
        """重新生成会话ID（防会话固定攻击）"""
        request.session.cycle_key()

# 会话键管理
class SessionKeyManager:
    """会话键管理器"""
    
    @staticmethod
    def create_namespaced_key(namespace, key):
        """创建命名空间键"""
        return f"{namespace}:{key}"
    
    @staticmethod
    def get_user_cart_key(user_id):
        """获取用户购物车键"""
        return SessionKeyManager.create_namespaced_key('cart', f'user_{user_id}')
    
    @staticmethod
    def get_user_auth_key(user_id):
        """获取用户认证键"""
        return SessionKeyManager.create_namespaced_key('auth', f'user_{user_id}')
    
    @staticmethod
    def get_user_settings_key(user_id):
        """获取用户设置键"""
        return SessionKeyManager.create_namespaced_key('settings', f'user_{user_id}')

# 使用示例
def shopping_cart_example(request, user_id):
    """购物车示例"""
    cart_key = SessionKeyManager.get_user_cart_key(user_id)
    
    # 获取购物车
    cart_items = request.session.get(cart_key, [])
    
    # 添加商品
    def add_item(product_id, quantity=1):
        for item in cart_items:
            if item['product_id'] == product_id:
                item['quantity'] += quantity
                break
        else:
            cart_items.append({
                'product_id': product_id,
                'quantity': quantity
            })
        
        request.session[cart_key] = cart_items
        request.session.modified = True
    
    # 获取总计
    def get_total():
        return len(cart_items)
    
    return {
        'items': cart_items,
        'total': get_total(),
        'add_item': add_item
    }

# 会话装饰器
from functools import wraps
from django.http import HttpResponseRedirect
from django.urls import reverse
from django.contrib import messages

def require_session_data(required_keys):
    """要求会话数据装饰器"""
    def decorator(view_func):
        @wraps(view_func)
        def wrapper(request, *args, **kwargs):
            # 检查必需的会话数据
            missing_keys = []
            for key in required_keys:
                if key not in request.session:
                    missing_keys.append(key)
            
            if missing_keys:
                messages.error(
                    request, 
                    f'缺少必要的会话数据: {", ".join(missing_keys)}'
                )
                return HttpResponseRedirect(reverse('login'))
            
            return view_func(request, *args, **kwargs)
        return wrapper
    return decorator

def session_timeout(timeout_minutes=30):
    """会话超时装饰器"""
    def decorator(view_func):
        @wraps(view_func)
        def wrapper(request, *args, **kwargs):
            import time
            
            # 检查最后活动时间
            last_activity = request.session.get('last_activity')
            if last_activity:
                if time.time() - last_activity > timeout_minutes * 60:
                    # 会话超时，清除会话
                    request.session.flush()
                    messages.error(request, '会话已超时，请重新登录')
                    return HttpResponseRedirect(reverse('login'))
            
            # 更新最后活动时间
            request.session['last_activity'] = time.time()
            request.session.modified = True
            
            return view_func(request, *args, **kwargs)
        return wrapper
    return decorator

def track_session_activity(view_func):
    """跟踪会话活动装饰器"""
    @wraps(view_func)
    def wrapper(request, *args, **kwargs):
        # 记录访问路径
        current_path = request.path
        previous_paths = request.session.get('navigation_history', [])
        
        # 限制历史记录长度
        if len(previous_paths) >= 10:
            previous_paths = previous_paths[-9:]  # 保留最近9个路径
        
        previous_paths.append(current_path)
        request.session['navigation_history'] = previous_paths
        request.session.modified = True
        
        return view_func(request, *args, **kwargs)
    return decorator

# 使用示例
@require_session_data(['user_id', 'username'])
@session_timeout(timeout_minutes=60)
@track_session_activity
def protected_view(request):
    """受保护的视图"""
    user_id = request.session['user_id']
    username = request.session['username']
    
    return f"Welcome {username} (ID: {user_id})"

# 会话上下文管理器
from contextlib import contextmanager

@contextmanager
def session_transaction(request):
    """会话事务上下文管理器"""
    original_session = dict(request.session)
    session_modified = request.session.modified
    
    try:
        yield request.session
        # 如果没有异常，提交更改
        request.session.modified = True
    except Exception:
        # 如果有异常，回滚到原始状态
        request.session.clear()
        request.session.update(original_session)
        request.session.modified = session_modified
        raise

# 使用示例
def complex_session_operation(request):
    """复杂的会话操作"""
    with session_transaction(request) as session:
        # 执行一系列会话操作
        session['step'] = 1
        session['data'] = {'key': 'value'}
        
        # 模拟可能失败的操作
        import random
        if random.choice([True, False]):  # 随机失败
            raise Exception("Operation failed")
        
        session['step'] = 2
        session['completed'] = True

# settings.py - 会话配置
"""
会话配置示例
"""
import os

# 会话配置
SESSION_COOKIE_AGE = 1209600  # 2周 (秒)
SESSION_EXPIRE_AT_BROWSER_CLOSE = False  # 关闭浏览器时不删除会话
SESSION_SAVE_EVERY_REQUEST = True  # 每次请求都保存会话

# 会话Cookie设置
SESSION_COOKIE_NAME = 'sessionid'  # Cookie名称
SESSION_COOKIE_PATH = '/'  # Cookie路径
SESSION_COOKIE_DOMAIN = None  # Cookie域名（None表示使用当前域名）
SESSION_COOKIE_SECURE = True  # HTTPS-only Cookie
SESSION_COOKIE_HTTPONLY = True  # JavaScript无法访问
SESSION_COOKIE_SAMESITE = 'Lax'  # CSRF保护

# 会话存储后端
SESSION_ENGINE = 'django.contrib.sessions.backends.cached_db'  # 使用缓存+数据库
SESSION_FILE_PATH = '/tmp/django_sessions'  # 文件存储路径（如果使用文件存储）

# 会话序列化
SESSION_SERIALIZER = 'django.contrib.sessions.serializers.JSONSerializer'

# 会话清理配置
SESSION_CLEANUP_BATCH_SIZE = 1000  # 批量清理数量

# 开发环境配置
if os.environ.get('DJANGO_ENV') == 'development':
    SESSION_COOKIE_SECURE = False  # 开发环境允许HTTP
    SESSION_COOKIE_DOMAIN = None

# 生产环境配置
if os.environ.get('DJANGO_ENV') == 'production':
    SESSION_COOKIE_SECURE = True  # 生产环境强制HTTPS
    SESSION_COOKIE_SAMESITE = 'Strict'  # 更严格的SameSite策略
    SESSION_ENGINE = 'django.contrib.sessions.backends.cached_db'
    SESSION_SAVE_EVERY_REQUEST = False  # 减少数据库写入

# Redis会话存储配置（需要django-redis-sessions）
"""
# 安装: pip install django-redis-sessions
SESSION_ENGINE = 'redis_sessions.session'
SESSION_REDIS_HOST = 'localhost'
SESSION_REDIS_PORT = 6379
SESSION_REDIS_DB = 0
SESSION_REDIS_PASSWORD = ''
SESSION_REDIS_PREFIX = 'session'
"""

# 数据库会话配置
from django.contrib.sessions.models import Session
from django.utils import timezone
from datetime import timedelta

class DatabaseSessionManager:
    """数据库会话管理器"""
    
    @staticmethod
    def get_active_sessions(hours_back=24):
        """获取活跃会话"""
        cutoff_time = timezone.now() - timedelta(hours=hours_back)
        active_sessions = Session.objects.filter(
            expire_date__gt=timezone.now(),
            session_data__isnull=False
        ).filter(
            # 获取最近活动的会话
            # 这里需要根据实际的session数据结构来过滤
        )
        return active_sessions
    
    @staticmethod
    def get_session_by_user(user_id):
        """根据用户ID获取会话"""
        sessions = Session.objects.filter(
            session_data__contains=str(user_id),
            expire_date__gt=timezone.now()
        )
        return sessions
    
    @staticmethod
    def cleanup_expired_sessions():
        """清理过期会话"""
        expired_count = Session.objects.filter(
            expire_date__lt=timezone.now()
        ).delete()[0]
        
        import logging
        logger = logging.getLogger('session.cleanup')
        logger.info(f"Cleaned up {expired_count} expired sessions")
        
        return expired_count
    
    @staticmethod
    def get_session_stats():
        """获取会话统计"""
        total_sessions = Session.objects.count()
        active_sessions = Session.objects.filter(
            expire_date__gt=timezone.now()
        ).count()
        expired_sessions = total_sessions - active_sessions
        
        return {
            'total': total_sessions,
            'active': active_sessions,
            'expired': expired_sessions,
            'active_percentage': (active_sessions / total_sessions * 100) if total_sessions > 0 else 0
        }
    
    @staticmethod
    def force_logout_user(user_id):
        """强制用户登出（删除用户相关会话）"""
        # 查找包含用户ID的会话
        sessions_to_delete = []
        
        for session in Session.objects.filter(expire_date__gt=timezone.now()):
            try:
                session_data = session.get_decoded()
                if session_data.get('user_id') == user_id:
                    sessions_to_delete.append(session.session_key)
            except Exception:
                # 解码失败的会话也删除
                sessions_to_delete.append(session.session_key)
        
        # 删除会话
        deleted_count = Session.objects.filter(
            session_key__in=sessions_to_delete
        ).delete()[0]
        
        return deleted_count

# 数据库会话清理命令
from django.core.management.base import BaseCommand

class Command(BaseCommand):
    """会话清理命令"""
    
    def add_arguments(self, parser):
        parser.add_argument(
            '--dry-run',
            action='store_true',
            help='只显示将要删除的会话，不实际删除',
        )
        parser.add_argument(
            '--older-than-days',
            type=int,
            default=30,
            help='删除多少天前的会话',
        )
    
    def handle(self, *args, **options):
        from datetime import timedelta
        from django.utils import timezone
        from django.contrib.sessions.models import Session
        
        cutoff_date = timezone.now() - timedelta(days=options['older_than_days'])
        
        if options['dry_run']:
            count = Session.objects.filter(expire_date__lt=cutoff_date).count()
            self.stdout.write(
                f"将会删除 {count} 个过期会话 "
                f"(早于 {cutoff_date.strftime('%Y-%m-%d %H:%M:%S')})"
            )
        else:
            deleted_count, _ = Session.objects.filter(
                expire_date__lt=cutoff_date
            ).delete()
            
            self.stdout.write(
                self.style.SUCCESS(
                    f"成功删除 {deleted_count} 个过期会话"
                )
            )

# 缓存会话配置
from django.core.cache import cache
from django.conf import settings
import json
import pickle
from datetime import datetime, timedelta

class CacheSessionManager:
    """缓存会话管理器"""
    
    def __init__(self):
        from django.core.cache import caches
        self.session_cache = caches['sessions'] if 'sessions' in settings.CACHES else cache
        self.prefix = getattr(settings, 'SESSION_CACHE_PREFIX', 'session:')
        self.default_timeout = getattr(settings, 'SESSION_COOKIE_AGE', 1209600)  # 2周
    
    def make_key(self, session_key):
        """生成缓存键"""
        return f"{self.prefix}{session_key}"
    
    def encode(self, session_dict):
        """编码会话数据"""
        return json.dumps(session_dict, default=str)
    
    def decode(self, session_data):
        """解码会话数据"""
        if isinstance(session_data, bytes):
            session_data = session_data.decode('utf-8')
        return json.loads(session_data)
    
    def create(self, session_key, session_dict, expiry=None):
        """创建会话"""
        cache_key = self.make_key(session_key)
        encoded_data = self.encode(session_dict)
        timeout = expiry or self.default_timeout
        
        self.session_cache.set(cache_key, encoded_data, timeout)
    
    def get(self, session_key, default=None):
        """获取会话"""
        cache_key = self.make_key(session_key)
        session_data = self.session_cache.get(cache_key, default)
        
        if session_data is not default and session_data is not None:
            try:
                return self.decode(session_data)
            except (json.JSONDecodeError, TypeError):
                return default
        
        return default
    
    def update(self, session_key, session_dict, expiry=None):
        """更新会话"""
        cache_key = self.make_key(session_key)
        encoded_data = self.encode(session_dict)
        timeout = expiry or self.default_timeout
        
        self.session_cache.set(cache_key, encoded_data, timeout)
    
    def delete(self, session_key):
        """删除会话"""
        cache_key = self.make_key(session_key)
        self.session_cache.delete(cache_key)
    
    def get_all_active_sessions(self):
        """获取所有活跃会话（需要Redis支持KEYS命令）"""
        try:
            import redis
            # 获取Redis客户端
            if hasattr(settings, 'CACHES') and 'sessions' in settings.CACHES:
                cache_config = settings.CACHES['sessions']
                if 'LOCATION' in cache_config:
                    redis_client = redis.from_url(cache_config['LOCATION'])
                    pattern = f"{self.prefix}*"
                    session_keys = redis_client.keys(pattern)
                    
                    sessions = {}
                    for key in session_keys:
                        key_str = key.decode() if isinstance(key, bytes) else key
                        session_key = key_str[len(self.prefix):]
                        session_data = self.get(session_key)
                        if session_data is not None:
                            sessions[session_key] = session_data
                    
                    return sessions
        except Exception as e:
            import logging
            logger = logging.getLogger(__name__)
            logger.error(f"Failed to get all active sessions: {e}")
        
        return {}
    
    def cleanup_expired_sessions(self):
        """清理过期会话"""
        # 缓存会话依赖于Redis的过期机制，通常不需要手动清理
        # 但可以实现一些维护操作
        pass

# 缓存+数据库会话配置
class CachedDBSessionManager:
    """缓存+数据库会话管理器"""
    
    def __init__(self):
        from django.core.cache import caches
        from django.db import models
        self.cache_backend = caches['sessions'] if 'sessions' in settings.CACHES else cache
        self.db_backend = 'django.contrib.sessions.backends.db'
        self.cache_prefix = 'cached_db_session:'
        self.cache_timeout = getattr(settings, 'SESSION_COOKIE_AGE', 1209600)
    
    def get(self, session_key):
        """获取会话（先查缓存，再查数据库）"""
        # 先查缓存
        cache_key = f"{self.cache_prefix}{session_key}"
        session_data = self.cache_backend.get(cache_key)
        
        if session_data is not None:
            return session_data
        
        # 缓存未命中，查数据库
        from django.contrib.sessions.backends.db import SessionStore
        db_session = SessionStore(session_key)
        
        if db_session.exists(session_key):
            session_data = db_session.load()
            # 同步到缓存
            self.cache_backend.set(cache_key, session_data, self.cache_timeout)
            return session_data
        
        return {}
    
    def save(self, session_key, session_data, expiry=None):
        """保存会话（同时保存到缓存和数据库）"""
        cache_key = f"{self.cache_prefix}{session_key}"
        timeout = expiry or self.cache_timeout
        
        # 保存到缓存
        self.cache_backend.set(cache_key, session_data, timeout)
        
        # 保存到数据库
        from django.contrib.sessions.backends.db import SessionStore
        db_session = SessionStore(session_key)
        db_session._session = session_data
        db_session.save()
    
    def delete(self, session_key):
        """删除会话（同时删除缓存和数据库）"""
        cache_key = f"{self.cache_prefix}{session_key}"
        
        # 删除缓存
        self.cache_backend.delete(cache_key)
        
        # 删除数据库
        from django.contrib.sessions.models import Session
        Session.objects.filter(session_key=session_key).delete()
    
    def exists(self, session_key):
        """检查会话是否存在"""
        cache_key = f"{self.cache_prefix}{session_key}"
        
        # 先检查缓存
        if self.cache_backend.get(cache_key) is not None:
            return True
        
        # 再检查数据库
        from django.contrib.sessions.backends.db import SessionStore
        db_session = SessionStore(session_key)
        return db_session.exists(session_key)

# 使用示例
def get_session_manager():
    """获取会话管理器"""
    session_engine = getattr(settings, 'SESSION_ENGINE', 'django.contrib.sessions.backends.db')
    
    if 'cached_db' in session_engine:
        return CachedDBSessionManager()
    elif 'cache' in session_engine:
        return CacheSessionManager()
    else:
        # 数据库会话使用Django内置的
        return None

# 会话CRUD操作
from django.contrib.sessions.models import Session
from django.utils import timezone
from datetime import timedelta
import json

class SessionCRUD:
    """会话CRUD操作类"""
    
    @staticmethod
    def create_session(user_id=None, initial_data=None):
        """创建新会话"""
        from django.contrib.sessions.models import Session
        from django.contrib.sessions.backends.db import SessionStore
        import uuid
        
        # 生成唯一的会话密钥
        session_key = uuid.uuid4().hex
        
        # 创建会话存储
        session_store = SessionStore(session_key=session_key)
        
        # 设置初始数据
        if initial_data:
            session_store.update(initial_data)
        
        # 设置用户ID（如果提供）
        if user_id:
            session_store['user_id'] = user_id
        
        # 设置过期时间
        session_store.set_expiry(getattr(settings, 'SESSION_COOKIE_AGE', 1209600))
        
        # 保存会话
        session_store.save()
        
        return session_key, session_store
    
    @staticmethod
    def read_session(session_key):
        """读取会话"""
        try:
            from django.contrib.sessions.backends.db import SessionStore
            session_store = SessionStore(session_key=session_key)
            
            if session_store.exists(session_key):
                return session_store.load()
            else:
                return None
        except Exception as e:
            import logging
            logger = logging.getLogger(__name__)
            logger.error(f"Error reading session {session_key}: {e}")
            return None
    
    @staticmethod
    def update_session(session_key, data_updates):
        """更新会话"""
        try:
            from django.contrib.sessions.backends.db import SessionStore
            session_store = SessionStore(session_key=session_key)
            
            if session_store.exists(session_key):
                # 加载现有数据
                current_data = session_store.load()
                
                # 更新数据
                current_data.update(data_updates)
                
                # 保存更新
                session_store._session = current_data
                session_store.save()
                
                return True
            else:
                return False
        except Exception as e:
            import logging
            logger = logging.getLogger(__name__)
            logger.error(f"Error updating session {session_key}: {e}")
            return False
    
    @staticmethod
    def delete_session(session_key):
        """删除会话"""
        try:
            from django.contrib.sessions.models import Session
            deleted_count, _ = Session.objects.filter(
                session_key=session_key
            ).delete()
            return deleted_count > 0
        except Exception as e:
            import logging
            logger = logging.getLogger(__name__)
            logger.error(f"Error deleting session {session_key}: {e}")
            return False
    
    @staticmethod
    def get_user_sessions(user_id):
        """获取用户的所有会话"""
        sessions = []
        
        for session in Session.objects.filter(expire_date__gt=timezone.now()):
            try:
                session_data = session.get_decoded()
                if session_data.get('user_id') == user_id:
                    sessions.append({
                        'session_key': session.session_key,
                        'created': session.get_decoded().get('created_at'),
                        'ip_address': session.get_decoded().get('ip_address'),
                        'user_agent': session.get_decoded().get('user_agent'),
                        'last_activity': session.get_decoded().get('last_activity')
                    })
            except Exception:
                continue  # 跳过损坏的会话数据
        
        return sessions

# 会话数据管理器
class SessionDataManager:
    """会话数据管理器"""
    
    def __init__(self, request):
        self.request = request
    
    def set_user_data(self, key, value):
        """设置用户数据"""
        user_namespace = f"user_{self.request.session.get('user_id', 'anonymous')}"
        namespaced_key = f"{user_namespace}:{key}"
        self.request.session[namespaced_key] = value
        self.request.session.modified = True
    
    def get_user_data(self, key, default=None):
        """获取用户数据"""
        user_namespace = f"user_{self.request.session.get('user_id', 'anonymous')}"
        namespaced_key = f"{user_namespace}:{key}"
        return self.request.session.get(namespaced_key, default)
    
    def delete_user_data(self, key):
        """删除用户数据"""
        user_namespace = f"user_{self.request.session.get('user_id', 'anonymous')}"
        namespaced_key = f"{user_namespace}:{key}"
        if namespaced_key in self.request.session:
            del self.request.session[namespaced_key]
            self.request.session.modified = True
    
    def set_flash_message(self, message, level='info'):
        """设置闪现消息"""
        flashes = self.request.session.get('flash_messages', [])
        flashes.append({
            'message': message,
            'level': level,
            'timestamp': timezone.now().isoformat()
        })
        # 限制消息数量
        if len(flashes) > 10:
            flashes = flashes[-10:]
        self.request.session['flash_messages'] = flashes
        self.request.session.modified = True
    
    def get_flash_messages(self):
        """获取并清除闪现消息"""
        flashes = self.request.session.get('flash_messages', [])
        # 清除已获取的消息
        self.request.session['flash_messages'] = []
        self.request.session.modified = True
        return flashes
    
    def set_form_errors(self, form_name, errors):
        """设置表单错误"""
        form_errors = self.request.session.get('form_errors', {})
        form_errors[form_name] = errors
        self.request.session['form_errors'] = form_errors
        self.request.session.modified = True
    
    def get_form_errors(self, form_name):
        """获取表单错误"""
        form_errors = self.request.session.get('form_errors', {})
        return form_errors.get(form_name, [])
    
    def clear_form_errors(self, form_name=None):
        """清除表单错误"""
        if form_name:
            form_errors = self.request.session.get('form_errors', {})
            if form_name in form_errors:
                del form_errors[form_name]
                self.request.session['form_errors'] = form_errors
                self.request.session.modified = True
        else:
            if 'form_errors' in self.request.session:
                del self.request.session['form_errors']
                self.request.session.modified = True

# 使用示例
def user_dashboard_view(request):
    """用户仪表板视图"""
    session_manager = SessionDataManager(request)
    
    # 设置用户偏好
    session_manager.set_user_data('theme', 'dark')
    session_manager.set_user_data('language', 'zh-CN')
    
    # 获取用户数据
    theme = session_manager.get_user_data('theme', 'light')
    language = session_manager.get_user_data('language', 'en-US')
    
    # 显示闪现消息
    flash_messages = session_manager.get_flash_messages()
    
    return {
        'theme': theme,
        'language': language,
        'flash_messages': flash_messages
    }

# 会话验证和授权
from django.contrib.auth import authenticate, login
from django.http import HttpResponseForbidden
from django.shortcuts import redirect
from django.urls import reverse
from django.contrib import messages

class SessionAuthManager:
    """会话认证管理器"""
    
    @staticmethod
    def validate_session(request):
        """验证会话有效性"""
        if not request.session.session_key:
            return False
        
        # 检查会话是否存在于数据库中
        session_key = request.session.session_key
        try:
            from django.contrib.sessions.models import Session
            session = Session.objects.get(session_key=session_key)
            
            # 检查会话是否过期
            if session.expire_date < timezone.now():
                return False
            
            # 检查额外的安全属性
            session_data = session.get_decoded()
            if not SessionAuthManager.verify_session_integrity(request, session_data):
                return False
            
            return True
        except Session.DoesNotExist:
            return False
    
    @staticmethod
    def verify_session_integrity(request, session_data):
        """验证会话完整性"""
        # 检查IP地址（可选）
        if 'ip_address' in session_data:
            if session_data['ip_address'] != request.META.get('REMOTE_ADDR'):
                return False
        
        # 检查User-Agent（可选）
        if 'user_agent' in session_data:
            if session_data['user_agent'] != request.META.get('HTTP_USER_AGENT', ''):
                return False
        
        # 检查最后活动时间（防长时间闲置）
        if 'last_activity' in session_data:
            import time
            last_activity = session_data['last_activity']
            if time.time() - last_activity > 3600 * 2:  # 2小时无活动
                return False
        
        return True
    
    @staticmethod
    def create_secure_session(request, user, additional_data=None):
        """创建安全会话"""
        # 首先登录用户
        login(request, user)
        
        # 扩展会话数据
        session_data = {
            'user_id': user.id,
            'username': user.username,
            'email': user.email,
            'ip_address': request.META.get('REMOTE_ADDR'),
            'user_agent': request.META.get('HTTP_USER_AGENT', ''),
            'created_at': timezone.now().isoformat(),
            'last_activity': time.time(),
            'session_fingerprint': SessionAuthManager.generate_fingerprint(request)
        }
        
        if additional_data:
            session_data.update(additional_data)
        
        # 更新会话
        request.session.update(session_data)
        request.session.modified = True
        
        # 重新生成会话ID以防止会话固定攻击
        request.session.cycle_key()
    
    @staticmethod
    def generate_fingerprint(request):
        """生成会话指纹"""
        import hashlib
        fingerprint_data = (
            request.META.get('HTTP_USER_AGENT', '') +
            request.META.get('REMOTE_ADDR', '') +
            request.META.get('HTTP_ACCEPT_LANGUAGE', '')
        )
        return hashlib.sha256(fingerprint_data.encode()).hexdigest()
    
    @staticmethod
    def refresh_session_activity(request):
        """刷新会话活动时间"""
        if request.user.is_authenticated and 'last_activity' in request.session:
            request.session['last_activity'] = time.time()
            request.session.modified = True
    
    @staticmethod
    def logout_user(request):
        """安全登出用户"""
        # 记录登出事件
        if request.user.is_authenticated:
            import logging
            logger = logging.getLogger('user.logout')
            logger.info(f"User {request.user.username} logged out from session {request.session.session_key}")
        
        # 清除会话
        request.session.flush()
        
        # 重定向到登录页面
        messages.success(request, '您已成功登出')

# 会话权限检查器
class SessionPermissionChecker:
    """会话权限检查器"""
    
    def __init__(self, request):
        self.request = request
    
    def has_permission(self, permission_codename):
        """检查权限"""
        if not self.request.user.is_authenticated:
            return False
        
        # 检查用户权限
        if self.request.user.has_perm(permission_codename):
            return True
        
        # 检查会话中的缓存权限
        cached_permissions = self.request.session.get('user_permissions', [])
        return permission_codename in cached_permissions
    
    def has_role(self, role_name):
        """检查角色"""
        if not self.request.user.is_authenticated:
            return False
        
        # 检查用户角色（组）
        return self.request.user.groups.filter(name=role_name).exists()
    
    def has_any_role(self, role_names):
        """检查任意角色"""
        if not self.request.user.is_authenticated:
            return False
        
        return self.request.user.groups.filter(name__in=role_names).exists()
    
    def cache_user_permissions(self):
        """缓存用户权限到会话"""
        if self.request.user.is_authenticated:
            permissions = list(
                self.request.user.user_permissions.values_list(
                    'codename', flat=True
                )
            )
            group_permissions = list(
                self.request.user.groups.values_list(
                    'permissions__codename', flat=True
                )
            )
            
            all_permissions = list(set(permissions + group_permissions))
            self.request.session['user_permissions'] = all_permissions
            self.request.session.modified = True

# 会话装饰器
def require_authentication(view_func):
    """要求认证装饰器"""
    @wraps(view_func)
    def wrapper(request, *args, **kwargs):
        if not request.user.is_authenticated:
            messages.error(request, '请先登录')
            return redirect(f"{reverse('login')}?next={request.get_full_path()}")
        
        # 验证会话
        if not SessionAuthManager.validate_session(request):
            messages.error(request, '会话无效，请重新登录')
            return redirect('login')
        
        # 更新活动时间
        SessionAuthManager.refresh_session_activity(request)
        
        return view_func(request, *args, **kwargs)
    return wrapper

def require_permission(permission):
    """要求权限装饰器"""
    def decorator(view_func):
        @wraps(view_func)
        def wrapper(request, *args, **kwargs):
            checker = SessionPermissionChecker(request)
            
            if not checker.has_permission(permission):
                messages.error(request, '您没有权限访问此页面')
                return HttpResponseForbidden("Access denied")
            
            return view_func(request, *args, **kwargs)
        return wrapper
    return decorator

def require_role(role_name):
    """要求角色装饰器"""
    def decorator(view_func):
        @wraps(view_func)
        def wrapper(request, *args, **kwargs):
            checker = SessionPermissionChecker(request)
            
            if not checker.has_role(role_name):
                messages.error(request, '您没有相应角色权限')
                return HttpResponseForbidden("Access denied")
            
            return view_func(request, *args, **kwargs)
        return wrapper
    return decorator

# 使用示例
@require_authentication
@require_permission('app.view_report')
@require_role('manager')
def admin_dashboard_view(request):
    """管理员仪表板视图"""
    # 检查器实例
    checker = SessionPermissionChecker(request)
    
    # 缓存权限（首次访问时）
    if 'user_permissions' not in request.session:
        checker.cache_user_permissions()
    
    # 检查其他权限
    can_edit = checker.has_permission('app.edit_report')
    is_admin = checker.has_role('administrator')
    
    return {
        'can_edit': can_edit,
        'is_admin': is_admin
    }

# 会话数据结构化管理
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, asdict
from datetime import datetime
import json

@dataclass
class UserPreferences:
    """用户偏好设置"""
    theme: str = 'light'
    language: str = 'en'
    timezone: str = 'UTC'
    notifications_enabled: bool = True
    email_frequency: str = 'daily'  # daily, weekly, never
    
    def to_dict(self):
        return asdict(self)
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]):
        return cls(**{k: v for k, v in data.items() if k in cls.__annotations__})

@dataclass
class ShoppingCartItem:
    """购物车项目"""
    product_id: int
    quantity: int
    price: float
    added_at: datetime
    
    def to_dict(self):
        data = asdict(self)
        data['added_at'] = data['added_at'].isoformat()
        return data
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]):
        data['added_at'] = datetime.fromisoformat(data['added_at'])
        return cls(**data)

@dataclass
class ShoppingCart:
    """购物车"""
    items: List[ShoppingCartItem]
    total_amount: float = 0.0
    currency: str = 'USD'
    updated_at: datetime = None
    
    def __post_init__(self):
        if self.updated_at is None:
            self.updated_at = datetime.now()
    
    def add_item(self, item: ShoppingCartItem):
        """添加项目"""
        for existing_item in self.items:
            if existing_item.product_id == item.product_id:
                existing_item.quantity += item.quantity
                break
        else:
            self.items.append(item)
        
        self.recalculate_total()
        self.updated_at = datetime.now()
    
    def remove_item(self, product_id: int):
        """移除项目"""
        self.items = [item for item in self.items if item.product_id != product_id]
        self.recalculate_total()
        self.updated_at = datetime.now()
    
    def recalculate_total(self):
        """重新计算总额"""
        self.total_amount = sum(item.price * item.quantity for item in self.items)
    
    def to_dict(self):
        data = asdict(self)
        data['items'] = [item.to_dict() for item in self.items]
        data['updated_at'] = data['updated_at'].isoformat()
        return data
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]):
        items_data = data.pop('items', [])
        items = [ShoppingCartItem.from_dict(item_data) for item_data in items_data]
        data['items'] = items
        data['updated_at'] = datetime.fromisoformat(data['updated_at'])
        return cls(**data)

@dataclass
class UserProfile:
    """用户资料"""
    name: str = ''
    email: str = ''
    avatar_url: str = ''
    last_login_ip: str = ''
    login_count: int = 0
    created_at: datetime = None
    
    def __post_init__(self):
        if self.created_at is None:
            self.created_at = datetime.now()
    
    def to_dict(self):
        data = asdict(self)
        data['created_at'] = data['created_at'].isoformat()
        return data
    
    @classmethod
    def from_dict(cls, data: Dict[str, Any]):
        data['created_at'] = datetime.fromisoformat(data['created_at'])
        return cls(**data)

class StructuredSessionManager:
    """结构化会话管理器"""
    
    def __init__(self, request):
        self.request = request
    
    def get_user_preferences(self) -> UserPreferences:
        """获取用户偏好"""
        pref_data = self.request.session.get('user_preferences', {})
        return UserPreferences.from_dict(pref_data)
    
    def set_user_preferences(self, preferences: UserPreferences):
        """设置用户偏好"""
        self.request.session['user_preferences'] = preferences.to_dict()
        self.request.session.modified = True
    
    def get_shopping_cart(self) -> ShoppingCart:
        """获取购物车"""
        cart_data = self.request.session.get('shopping_cart', {'items': []})
        return ShoppingCart.from_dict(cart_data)
    
    def set_shopping_cart(self, cart: ShoppingCart):
        """设置购物车"""
        self.request.session['shopping_cart'] = cart.to_dict()
        self.request.session.modified = True
    
    def get_user_profile(self) -> UserProfile:
        """获取用户资料"""
        profile_data = self.request.session.get('user_profile', {})
        return UserProfile.from_dict(profile_data)
    
    def set_user_profile(self, profile: UserProfile):
        """设置用户资料"""
        self.request.session['user_profile'] = profile.to_dict()
        self.request.session.modified = True
    
    def update_user_preferences(self, **kwargs):
        """更新用户偏好"""
        prefs = self.get_user_preferences()
        for key, value in kwargs.items():
            if hasattr(prefs, key):
                setattr(prefs, key, value)
        self.set_user_preferences(prefs)
    
    def add_to_cart(self, product_id: int, quantity: int, price: float):
        """添加到购物车"""
        cart = self.get_shopping_cart()
        item = ShoppingCartItem(
            product_id=product_id,
            quantity=quantity,
            price=price,
            added_at=datetime.now()
        )
        cart.add_item(item)
        self.set_shopping_cart(cart)
    
    def remove_from_cart(self, product_id: int):
        """从购物车移除"""
        cart = self.get_shopping_cart()
        cart.remove_item(product_id)
        self.set_shopping_cart(cart)
    
    def update_user_profile(self, **kwargs):
        """更新用户资料"""
        profile = self.get_user_profile()
        for key, value in kwargs.items():
            if hasattr(profile, key):
                setattr(profile, key, value)
        self.set_user_profile(profile)

# 会话数据验证器
class SessionDataValidator:
    """会话数据验证器"""
    
    @staticmethod
    def validate_user_preferences(data: Dict[str, Any]) -> List[str]:
        """验证用户偏好数据"""
        errors = []
        
        if 'theme' in data and data['theme'] not in ['light', 'dark', 'auto']:
            errors.append('Invalid theme value')
        
        if 'language' in data and data['language'] not in ['en', 'zh', 'ja', 'ko']:
            errors.append('Invalid language value')
        
        if 'timezone' in data:
            import pytz
            try:
                pytz.timezone(data['timezone'])
            except pytz.exceptions.UnknownTimeZoneError:
                errors.append('Invalid timezone')
        
        if 'email_frequency' in data and data['email_frequency'] not in ['daily', 'weekly', 'never']:
            errors.append('Invalid email frequency')
        
        return errors
    
    @staticmethod
    def validate_shopping_cart(data: Dict[str, Any]) -> List[str]:
        """验证购物车数据"""
        errors = []
        
        if 'items' in data:
            for i, item in enumerate(data['items']):
                if not isinstance(item, dict):
                    errors.append(f'Item {i} is not a valid dictionary')
                    continue
                
                if 'product_id' not in item or not isinstance(item['product_id'], int):
                    errors.append(f'Item {i} has invalid product_id')
                
                if 'quantity' not in item or not isinstance(item['quantity'], int) or item['quantity'] <= 0:
                    errors.append(f'Item {i} has invalid quantity')
                
                if 'price' not in item or not isinstance(item['price'], (int, float)) or item['price'] < 0:
                    errors.append(f'Item {i} has invalid price')
        
        if 'total_amount' in data and data['total_amount'] < 0:
            errors.append('Total amount cannot be negative')
        
        return errors
    
    @staticmethod
    def validate_user_profile(data: Dict[str, Any]) -> List[str]:
        """验证用户资料数据"""
        errors = []
        
        if 'email' in data:
            import re
            email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
            if not re.match(email_pattern, data['email']):
                errors.append('Invalid email format')
        
        if 'login_count' in data and data['login_count'] < 0:
            errors.append('Login count cannot be negative')
        
        return errors

# 使用示例
def enhanced_shopping_view(request):
    """增强购物视图"""
    session_manager = StructuredSessionManager(request)
    validator = SessionDataValidator()
    
    # 获取当前购物车
    cart = session_manager.get_shopping_cart()
    
    # 添加商品到购物车
    def add_product(product_id, quantity, price):
        session_manager.add_to_cart(product_id, quantity, price)
    
    # 更新用户偏好
    def update_prefs(**prefs):
        errors = SessionDataValidator.validate_user_preferences(prefs)
        if errors:
            return {'success': False, 'errors': errors}
        
        session_manager.update_user_preferences(**prefs)
        return {'success': True}
    
    # 获取用户偏好
    preferences = session_manager.get_user_preferences()
    
    return {
        'cart': cart,
        'preferences': preferences,
        'add_product': add_product,
        'update_preferences': update_prefs
    }

# 会话安全配置
from django.conf import settings
from django.contrib.sessions.models import Session
from django.utils import timezone
import secrets
import hashlib
import hmac

class SessionSecurityConfig:
    """会话安全配置"""
    
    @staticmethod
    def get_secure_settings():
        """获取安全的会话设置"""
        return {
            'SESSION_COOKIE_SECURE': True,  # HTTPS only
            'SESSION_COOKIE_HTTPONLY': True,  # JavaScript inaccessible
            'SESSION_COOKIE_SAMESITE': 'Strict',  # CSRF protection
            'SESSION_EXPIRE_AT_BROWSER_CLOSE': True,  # Session ends with browser
            'SESSION_SAVE_EVERY_REQUEST': True,  # Refresh session timeout
            'SESSION_COOKIE_AGE': 3600,  # 1 hour timeout
        }
    
    @staticmethod
    def apply_security_hardening():
        """应用安全强化"""
        # 在生产环境中应用安全设置
        if not settings.DEBUG:
            settings.SESSION_COOKIE_SECURE = True
            settings.SESSION_COOKIE_HTTPONLY = True
            settings.SESSION_COOKIE_SAMESITE = 'Strict'
    
    @staticmethod
    def generate_csrf_token():
        """生成CSRF令牌"""
        return secrets.token_urlsafe(32)

# 会话安全检查器
class SessionSecurityChecker:
    """会话安全检查器"""
    
    def __init__(self, request):
        self.request = request
        self.session_key = request.session.session_key
    
    def check_session_flooding(self, max_sessions_per_ip=5):
        """检查会话泛洪"""
        ip_address = self.request.META.get('REMOTE_ADDR')
        current_sessions = Session.objects.filter(
            session_data__contains=ip_address,
            expire_date__gt=timezone.now()
        ).count()
        
        return current_sessions <= max_sessions_per_ip
    
    def check_session_replay(self):
        """检查会话重放"""
        # 检查是否为新生成的会话ID
        if hasattr(self.request, '_new_session_created'):
            return True
        
        # 检查会话指纹
        session_data = self.get_session_data()
        if 'session_fingerprint' in session_data:
            current_fingerprint = self.generate_current_fingerprint()
            stored_fingerprint = session_data['session_fingerprint']
            return current_fingerprint == stored_fingerprint
        
        return True
    
    def check_privilege_escalation(self, expected_user_id):
        """检查权限提升"""
        session_data = self.get_session_data()
        current_user_id = session_data.get('user_id')
        
        # 如果会话中的用户ID与期望的不同，可能是权限提升
        return current_user_id == expected_user_id
    
    def generate_current_fingerprint(self):
        """生成当前请求的指纹"""
        user_agent = self.request.META.get('HTTP_USER_AGENT', '')
        ip_address = self.request.META.get('REMOTE_ADDR', '')
        accept_language = self.request.META.get('HTTP_ACCEPT_LANGUAGE', '')
        
        fingerprint_data = f"{user_agent}|{ip_address}|{accept_language}"
        return hashlib.sha256(fingerprint_data.encode()).hexdigest()
    
    def get_session_data(self):
        """获取会话数据"""
        if self.session_key:
            try:
                session = Session.objects.get(session_key=self.session_key)
                return session.get_decoded()
            except Session.DoesNotExist:
                return {}
        return {}

# 会话安全中间件
from django.utils.deprecation import MiddlewareMixin

class SessionSecurityMiddleware(MiddlewareMixin):
    """会话安全中间件"""
    
    def process_request(self, request):
        """处理请求 - 安全检查"""
        if hasattr(request, 'session') and request.session.session_key:
            security_checker = SessionSecurityChecker(request)
            
            # 检查会话泛洪
            if not security_checker.check_session_flooding():
                request.session.flush()
                return self.block_request("Too many sessions from your IP")
            
            # 检查会话重放
            if not security_checker.check_session_replay():
                request.session.flush()
                return self.block_request("Potential session replay detected")
            
            # 更新最后活动时间
            self.update_last_activity(request)
    
    def update_last_activity(self, request):
        """更新最后活动时间"""
        if 'last_activity' in request.session:
            import time
            current_time = time.time()
            last_activity = request.session['last_activity']
            
            # 如果距离上次活动超过阈值，记录异常
            if current_time - last_activity > 3600:  # 1小时
                import logging
                logger = logging.getLogger('security.anomaly')
                logger.warning(
                    f"Long session gap detected for session {request.session.session_key}"
                )
            
            request.session['last_activity'] = current_time
            request.session.modified = True
    
    def block_request(self, reason):
        """阻止请求"""
        from django.http import HttpResponse
        return HttpResponse(
            f"Request blocked: {reason}", 
            status=403
        )

# 会话固定攻击防护
class SessionFixationProtection:
    """会话固定攻击防护"""
    
    @staticmethod
    def regenerate_session_on_login(request):
        """登录时重新生成会话ID"""
        # 保存当前会话数据
        session_data = dict(request.session)
        
        # 循环会话密钥（重新生成）
        request.session.cycle_key()
        
        # 恢复会话数据
        request.session.update(session_data)
        request.session.modified = True
        
        # 标记为新会话
        request._new_session_created = True
    
    @staticmethod
    def validate_session_origin(request):
        """验证会话来源"""
        session_data = request.session
        if 'original_ip' not in session_data:
            # 首次访问，记录原始IP
            session_data['original_ip'] = request.META.get('REMOTE_ADDR')
            session_data['original_user_agent'] = request.META.get('HTTP_USER_AGENT', '')
            request.session.modified = True
        else:
            # 后续访问，验证IP和User-Agent
            original_ip = session_data.get('original_ip')
            original_ua = session_data.get('original_user_agent')
            current_ip = request.META.get('REMOTE_ADDR')
            current_ua = request.META.get('HTTP_USER_AGENT', '')
            
            if original_ip != current_ip or original_ua != current_ua:
                # 可能存在会话劫持
                import logging
                logger = logging.getLogger('security.session_hijack')
                logger.warning(
                    f"Potential session hijack detected: "
                    f"Original IP: {original_ip}, Current IP: {current_ip} | "
                    f"Original UA: {original_ua}, Current UA: {current_ua}"
                )
                return False
        
        return True

# 会话加密管理器
class EncryptedSessionManager:
    """加密会话管理器"""
    
    def __init__(self):
        from cryptography.fernet import Fernet
        from django.conf import settings
        
        # 使用Django的SECRET_KEY生成加密密钥
        secret = settings.SECRET_KEY
        if len(secret) < 32:
            # 如果SECRET_KEY不够长，使用哈希扩展
            import hashlib
            secret = hashlib.sha256(secret.encode()).digest()[:32]
        
        # 将字节转换为base64格式以适应Fernet要求
        import base64
        self.key = base64.urlsafe_b64encode(secret[:32] if isinstance(secret, bytes) else secret.encode()[:32])
        self.cipher_suite = Fernet(self.key)
    
    def encrypt_data(self, data_dict):
        """加密数据"""
        json_data = json.dumps(data_dict)
        encrypted_data = self.cipher_suite.encrypt(json_data.encode())
        return encrypted_data.decode('utf-8')
    
    def decrypt_data(self, encrypted_str):
        """解密数据"""
        try:
            decrypted_data = self.cipher_suite.decrypt(encrypted_str.encode())
            return json.loads(decrypted_data.decode('utf-8'))
        except Exception as e:
            import logging
            logger = logging.getLogger(__name__)
            logger.error(f"Session decryption failed: {e}")
            return {}
    
    def store_encrypted_session(self, request, key, value):
        """存储加密会话数据"""
        encrypted_value = self.encrypt_data({key: value})
        request.session[f"encrypted_{key}"] = encrypted_value
        request.session.modified = True
    
    def retrieve_encrypted_session(self, request, key):
        """检索加密会话数据"""
        encrypted_str = request.session.get(f"encrypted_{key}")
        if encrypted_str:
            decrypted_data = self.decrypt_data(encrypted_str)
            return decrypted_data.get(key)
        return None

# 使用示例
def secure_login_view(request):
    """安全登录视图"""
    if request.method == 'POST':
        username = request.POST.get('username')
        password = request.POST.get('password')
        
        user = authenticate(request, username=username, password=password)
        if user:
            # 登录前重新生成会话ID以防止会话固定攻击
            SessionFixationProtection.regenerate_session_on_login(request)
            
            # 创建安全会话
            SessionAuthManager.create_secure_session(request, user, {
                'login_time': timezone.now().isoformat(),
                'ip_address': request.META.get('REMOTE_ADDR'),
                'user_agent': request.META.get('HTTP_USER_AGENT', '')
            })
            
            # 缓存用户权限
            checker = SessionPermissionChecker(request)
            checker.cache_user_permissions()
            
            messages.success(request, '登录成功')
            return redirect('dashboard')
        else:
            messages.error(request, '用户名或密码错误')
    
    return render(request, 'login.html')