From e4d4677f3a584d5df18cf8987fe5658ca089ea14 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 12 Jul 2025 18:01:17 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=B7=BB=E5=8A=A0auth=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- python_core/api/auth_api.py | 244 ++++++++++++++ python_core/cli/cli.py | 3 + python_core/cli/commands/auth.py | 315 +++++++++++++++++ python_core/database/types.py | 26 +- python_core/database/user.py | 8 + python_core/models/user.py | 96 ++++++ python_core/requirements.txt | 3 +- python_core/services/auth_service.py | 263 +++++++++++++++ .../services/resource_category_manager.py | 88 ----- python_core/services/user_storage.py | 249 ++++++++++++++ python_core/utils/jwt_auth.py | 205 ++++++++++++ test_auth_system.py | 316 ++++++++++++++++++ 12 files changed, 1725 insertions(+), 91 deletions(-) create mode 100644 python_core/api/auth_api.py create mode 100644 python_core/cli/commands/auth.py create mode 100644 python_core/database/user.py create mode 100644 python_core/models/user.py create mode 100644 python_core/services/auth_service.py create mode 100644 python_core/services/user_storage.py create mode 100644 python_core/utils/jwt_auth.py create mode 100644 test_auth_system.py diff --git a/python_core/api/auth_api.py b/python_core/api/auth_api.py new file mode 100644 index 0000000..2854671 --- /dev/null +++ b/python_core/api/auth_api.py @@ -0,0 +1,244 @@ +""" +用户认证API接口 +""" + +from typing import Dict, Any, Optional +import json + +from python_core.models.user import LoginRequest, RegisterRequest +from python_core.services.auth_service import auth_service +from python_core.utils.logger import logger + + +class AuthAPI: + """用户认证API""" + + def __init__(self): + self.auth_service = auth_service + logger.info("AuthAPI initialized") + + def register(self, data: Dict[str, Any]) -> Dict[str, Any]: + """ + 用户注册API + + Args: + data: 注册数据 + { + "username": "用户名", + "email": "邮箱", + "password": "密码", + "display_name": "显示名称" (可选) + } + + Returns: + Dict: API响应 + """ + try: + # 创建注册请求对象 + request = RegisterRequest( + username=data.get("username", "").strip(), + email=data.get("email", "").strip(), + password=data.get("password", ""), + display_name=data.get("display_name", "").strip() or None + ) + + # 执行注册 + response = self.auth_service.register(request) + + # 返回API响应 + return { + "success": response.success, + "message": response.message, + "data": { + "user": response.user, + "token": response.token, + "expires_at": response.expires_at + } if response.success else None + } + + except Exception as e: + logger.error(f"Register API error: {e}") + return { + "success": False, + "message": "注册失败，请稍后重试", + "data": None + } + + def login(self, data: Dict[str, Any]) -> Dict[str, Any]: + """ + 用户登录API + + Args: + data: 登录数据 + { + "username_or_email": "用户名或邮箱", + "password": "密码" + } + + Returns: + Dict: API响应 + """ + try: + # 创建登录请求对象 + request = LoginRequest( + username_or_email=data.get("username_or_email", "").strip(), + password=data.get("password", "") + ) + + # 执行登录 + response = self.auth_service.login(request) + + # 返回API响应 + return { + "success": response.success, + "message": response.message, + "data": { + "user": response.user, + "token": response.token, + "expires_at": response.expires_at + } if response.success else None + } + + except Exception as e: + logger.error(f"Login API error: {e}") + return { + "success": False, + "message": "登录失败，请稍后重试", + "data": None + } + + def verify_token(self, data: Dict[str, Any]) -> Dict[str, Any]: + """ + 验证token API + + Args: + data: 验证数据 + { + "token": "JWT token" + } + + Returns: + Dict: API响应 + """ + try: + token = data.get("token", "") + if not token: + return { + "success": False, + "message": "Token不能为空", + "data": None + } + + # 验证token + user_info = self.auth_service.verify_token(token) + + if user_info: + return { + "success": True, + "message": "Token验证成功", + "data": { + "user": user_info, + "valid": True + } + } + else: + return { + "success": False, + "message": "Token无效或已过期", + "data": { + "valid": False + } + } + + except Exception as e: + logger.error(f"Verify token API error: {e}") + return { + "success": False, + "message": "Token验证失败", + "data": None + } + + def get_current_user(self, data: Dict[str, Any]) -> Dict[str, Any]: + """ + 获取当前用户信息API + + Args: + data: 请求数据 + { + "token": "JWT token" + } + + Returns: + Dict: API响应 + """ + try: + token = data.get("token", "") + if not token: + return { + "success": False, + "message": "Token不能为空", + "data": None + } + + # 获取当前用户 + user = self.auth_service.get_current_user(token) + + if user: + return { + "success": True, + "message": "获取用户信息成功", + "data": { + "user": user.to_safe_dict() + } + } + else: + return { + "success": False, + "message": "用户不存在或token无效", + "data": None + } + + except Exception as e: + logger.error(f"Get current user API error: {e}") + return { + "success": False, + "message": "获取用户信息失败", + "data": None + } + + +# 创建全局认证API实例 +auth_api = AuthAPI() + + +# JSON-RPC风格的便捷函数 +def register_user(username: str, email: str, password: str, display_name: Optional[str] = None) -> Dict[str, Any]: + """注册用户的便捷函数""" + return auth_api.register({ + "username": username, + "email": email, + "password": password, + "display_name": display_name + }) + + +def login_user(username_or_email: str, password: str) -> Dict[str, Any]: + """登录用户的便捷函数""" + return auth_api.login({ + "username_or_email": username_or_email, + "password": password + }) + + +def verify_user_token(token: str) -> Dict[str, Any]: + """验证用户token的便捷函数""" + return auth_api.verify_token({ + "token": token + }) + + +def get_user_info(token: str) -> Dict[str, Any]: + """获取用户信息的便捷函数""" + return auth_api.get_current_user({ + "token": token + }) diff --git a/python_core/cli/cli.py b/python_core/cli/cli.py index 8b286a8..eb66552 100644 --- a/python_core/cli/cli.py +++ b/python_core/cli/cli.py @@ -11,6 +11,7 @@ import typer from python_core.cli.commands import scene_detect from python_core.cli.commands.template import template_app from python_core.cli.commands.category import category_app +from python_core.cli.commands.auth import auth_app app = typer.Typer( name="mixvideo", @@ -21,6 +22,7 @@ app = typer.Typer( • 🎯 场景检测 - 智能识别视频场景变化 • 📋 模板管理 - 视频模板批量导入、列表查看、详情获取 • 🏷️ 分类管理 - 资源分类创建、更新、删除、搜索 + • 👤 用户认证 - 用户注册、登录、JWT token管理 • 📤 媒体管理 - 上传、处理、组织视频文件 • ⚙️ 系统管理 - 配置、状态、存储管理 """, @@ -32,6 +34,7 @@ app = typer.Typer( app.add_typer(scene_detect, name="scene") app.add_typer(template_app, name="template") app.add_typer(category_app, name="category") +app.add_typer(auth_app, name="auth") @app.command() def init(): diff --git a/python_core/cli/commands/auth.py b/python_core/cli/commands/auth.py new file mode 100644 index 0000000..d433b15 --- /dev/null +++ b/python_core/cli/commands/auth.py @@ -0,0 +1,315 @@ +""" +用户认证CLI命令 +""" + +from typing import Optional +import typer +from rich.console import Console +from rich.table import Table +from rich.panel import Panel +import json +import getpass + +from python_core.api.auth_api import auth_api +from python_core.services.user_storage import user_storage +from python_core.utils.jwt_auth import jwt_auth + +console = Console() +auth_app = typer.Typer(name="auth", help="用户认证管理命令") + + +@auth_app.command("register") +def register_user( + username: str = typer.Argument(..., help="用户名"), + email: str = typer.Argument(..., help="邮箱地址"), + display_name: Optional[str] = typer.Option(None, "--display-name", "-d", help="显示名称"), + password: Optional[str] = typer.Option(None, "--password", "-p", help="密码（不提供则交互式输入）"), + verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出") +): + """注册新用户""" + + try: + console.print(f"🆕 [bold blue]用户注册[/bold blue]") + console.print(f"用户名: {username}") + console.print(f"邮箱: {email}") + if display_name: + console.print(f"显示名称: {display_name}") + + # 获取密码 + if not password: + password = getpass.getpass("请输入密码: ") + if not password: + console.print("[red]❌ 密码不能为空[/red]") + raise typer.Exit(1) + + # 执行注册 + result = auth_api.register({ + "username": username, + "email": email, + "password": password, + "display_name": display_name + }) + + if result["success"]: + console.print(f"\n✅ [bold green]注册成功![/bold green]") + console.print(f"用户ID: {result['data']['user']['id']}") + console.print(f"创建时间: {result['data']['user']['created_at']}") + + if verbose: + # 显示详细信息 + user_info = f""" +🆔 用户ID: {result['data']['user']['id']} +👤 用户名: {result['data']['user']['username']} +📧 邮箱: {result['data']['user']['email']} +🏷️ 显示名称: {result['data']['user']['display_name']} +📅 创建时间: {result['data']['user']['created_at']} +🔑 Token: {result['data']['token'][:50]}... +⏰ 过期时间: {result['data']['expires_at']} + """ + + panel = Panel(user_info.strip(), title="用户详情", border_style="green") + console.print(panel) + else: + console.print(f"[red]❌ 注册失败: {result['message']}[/red]") + raise typer.Exit(1) + + except Exception as e: + console.print(f"[red]❌ 注册失败: {str(e)}[/red]") + raise typer.Exit(1) + + +@auth_app.command("login") +def login_user( + username_or_email: str = typer.Argument(..., help="用户名或邮箱"), + password: Optional[str] = typer.Option(None, "--password", "-p", help="密码（不提供则交互式输入）"), + verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出") +): + """用户登录""" + + try: + console.print(f"🔐 [bold blue]用户登录[/bold blue]") + console.print(f"用户名/邮箱: {username_or_email}") + + # 获取密码 + if not password: + password = getpass.getpass("请输入密码: ") + if not password: + console.print("[red]❌ 密码不能为空[/red]") + raise typer.Exit(1) + + # 执行登录 + result = auth_api.login({ + "username_or_email": username_or_email, + "password": password + }) + + if result["success"]: + console.print(f"\n✅ [bold green]登录成功![/bold green]") + console.print(f"欢迎回来，{result['data']['user']['display_name']}!") + console.print(f"最后登录: {result['data']['user']['last_login']}") + + if verbose: + # 显示详细信息 + user_info = f""" +🆔 用户ID: {result['data']['user']['id']} +👤 用户名: {result['data']['user']['username']} +📧 邮箱: {result['data']['user']['email']} +🏷️ 显示名称: {result['data']['user']['display_name']} +🕒 最后登录: {result['data']['user']['last_login']} +🔑 Token: {result['data']['token'][:50]}... +⏰ 过期时间: {result['data']['expires_at']} + """ + + panel = Panel(user_info.strip(), title="登录信息", border_style="green") + console.print(panel) + else: + console.print(f"[red]❌ 登录失败: {result['message']}[/red]") + raise typer.Exit(1) + + except Exception as e: + console.print(f"[red]❌ 登录失败: {str(e)}[/red]") + raise typer.Exit(1) + + +@auth_app.command("verify") +def verify_token( + token: str = typer.Argument(..., help="JWT token"), + verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出") +): + """验证JWT token""" + + try: + console.print(f"🔍 [bold blue]验证Token[/bold blue]") + + # 验证token + result = auth_api.verify_token({ + "token": token + }) + + if result["success"]: + console.print(f"\n✅ [bold green]Token有效![/bold green]") + user = result['data']['user'] + console.print(f"用户: {user['display_name']} ({user['username']})") + console.print(f"邮箱: {user['email']}") + + if verbose: + # 显示token详细信息 + token_info = jwt_auth.get_token_info(token) + + info_text = f""" +👤 用户信息: + ID: {user['user_id']} + 用户名: {user['username']} + 邮箱: {user['email']} + 显示名称: {user['display_name']} + +🔑 Token信息: + 有效性: {'有效' if token_info['valid'] else '无效'} + 签发时间: {token_info.get('issued_at', 'Unknown')} + 过期时间: {token_info.get('expires_at', 'Unknown')} + 剩余时间: {token_info.get('time_remaining', 'Unknown')} + """ + + panel = Panel(info_text.strip(), title="Token详情", border_style="green") + console.print(panel) + else: + console.print(f"[red]❌ Token无效: {result['message']}[/red]") + + if verbose: + # 显示token信息（即使无效） + token_info = jwt_auth.get_token_info(token) + if token_info.get('error'): + console.print(f"错误: {token_info['error']}") + else: + console.print(f"Token状态: {'过期' if token_info.get('is_expired') else '无效'}") + + raise typer.Exit(1) + + except Exception as e: + console.print(f"[red]❌ Token验证失败: {str(e)}[/red]") + raise typer.Exit(1) + + +@auth_app.command("list") +def list_users( + include_inactive: bool = typer.Option(False, "--include-inactive", "-i", help="包含非活跃用户"), + limit: int = typer.Option(20, "--limit", "-l", help="显示数量限制"), + verbose: bool = typer.Option(False, "--verbose", "-v", help="详细输出") +): + """列出所有用户""" + + try: + console.print(f"👥 [bold blue]用户列表[/bold blue]") + + # 获取用户列表 + users = user_storage.get_all_users(include_inactive) + + if not users: + console.print("📭 没有用户") + return + + # 限制显示数量 + total_count = len(users) + if len(users) > limit: + users = users[:limit] + console.print(f"📊 显示前 {limit} 个用户（共 {total_count} 个）") + else: + console.print(f"📊 共 {total_count} 个用户") + + # 创建表格 + table = Table(title="用户列表") + table.add_column("ID", style="cyan", width=8) + table.add_column("用户名", style="green") + table.add_column("邮箱", style="yellow") + table.add_column("显示名称", style="magenta") + table.add_column("状态", style="blue") + table.add_column("创建时间", style="dim") + + if verbose: + table.add_column("最后登录", style="dim") + + for user in users: + # 状态 + status = "✅ 活跃" if user.is_active else "❌ 禁用" + + # 格式化创建时间 + created_at = user.created_at + if 'T' in created_at: + created_at = created_at.split('T')[0] + ' ' + created_at.split('T')[1][:8] + + row = [ + user.id[:8], + user.username, + user.email, + user.display_name, + status, + created_at + ] + + if verbose: + last_login = user.last_login or "从未登录" + if last_login != "从未登录" and 'T' in last_login: + last_login = last_login.split('T')[0] + ' ' + last_login.split('T')[1][:8] + row.append(last_login) + + table.add_row(*row) + + console.print(table) + + except Exception as e: + console.print(f"[red]❌ 获取用户列表失败: {str(e)}[/red]") + raise typer.Exit(1) + + +@auth_app.command("stats") +def show_stats(): + """显示用户统计信息""" + + try: + console.print(f"📊 [bold blue]用户统计信息[/bold blue]") + + # 获取统计信息 + stats = user_storage.get_user_count() + + # 创建统计面板 + stats_content = f""" +📈 [bold green]用户统计[/bold green] + 总用户数: {stats['total']} + 活跃用户: {stats['active']} + 禁用用户: {stats['inactive']} + """ + + stats_panel = Panel(stats_content.strip(), title="用户统计", border_style="green") + console.print(stats_panel) + + # 显示最近注册的用户 + recent_users = user_storage.get_all_users()[:5] + + if recent_users: + console.print(f"\n🕒 [bold green]最近注册的用户[/bold green]") + + recent_table = Table() + recent_table.add_column("用户名", style="green") + recent_table.add_column("邮箱", style="yellow") + recent_table.add_column("注册时间", style="dim") + + for user in recent_users: + created_at = user.created_at + if 'T' in created_at: + created_at = created_at.split('T')[0] + ' ' + created_at.split('T')[1][:8] + + recent_table.add_row( + user.username, + user.email, + created_at + ) + + console.print(recent_table) + + except Exception as e: + console.print(f"[red]❌ 获取统计信息失败: {str(e)}[/red]") + raise typer.Exit(1) + + +if __name__ == "__main__": + auth_app() diff --git a/python_core/database/types.py b/python_core/database/types.py index f4101de..1c140ed 100644 --- a/python_core/database/types.py +++ b/python_core/database/types.py @@ -2,6 +2,23 @@ from dataclasses import dataclass from typing import Dict, List, Any, Optional +@dataclass +class ResourceCategory: + """素材分类数据结构""" + id: str + title: str # 分类标题 + ai_prompt: str # AI识别提示词 + color: str # 展示颜色 (hex格式，如 #FF5733) + created_at: str + updated_at: str + is_active: bool = True + + # is_cloud false 代表本地资源 那么只加载自己的 如果是 true 就是公用的 + is_cloud: bool + # 创建用户 + user_id: str + + @dataclass class MaterialInfo: """Material information structure""" @@ -13,6 +30,11 @@ class MaterialInfo: relative_path: str duration: Optional[int] = None size: Optional[int] = None + + # is_cloud false 代表本地资源 那么只加载自己的 如果是 true 就是公用的 + is_cloud: bool + # 创建用户 + user_id: str @dataclass class TemplateInfo: @@ -29,7 +51,7 @@ class TemplateInfo: material_count: int track_count: int tags: List[str] - # local/cloud 如果类型是 local 那么只加载自己的 如果是 cloud 就是公用的 - type: str + # is_cloud false 代表本地资源 那么只加载自己的 如果是 true 就是公用的 + is_cloud: bool # 创建用户 user_id: str \ No newline at end of file diff --git a/python_core/database/user.py b/python_core/database/user.py new file mode 100644 index 0000000..78e583a --- /dev/null +++ b/python_core/database/user.py @@ -0,0 +1,8 @@ +# 用户表 + +from python_core.kv import kv +from .db import Db +class UserDb(Db): + def __init__(self): + self.key = "model" + pass \ No newline at end of file diff --git a/python_core/models/user.py b/python_core/models/user.py new file mode 100644 index 0000000..4a1605b --- /dev/null +++ b/python_core/models/user.py @@ -0,0 +1,96 @@ +""" +用户数据模型 +""" + +from dataclasses import dataclass, asdict +from datetime import datetime +from typing import Optional, Dict, Any +import uuid +import hashlib +import json + + +@dataclass +class User: + """用户数据模型""" + id: str + username: str + email: str + password_hash: str + display_name: str + avatar_url: Optional[str] = None + is_active: bool = True + created_at: str = "" + updated_at: str = "" + last_login: Optional[str] = None + + def __post_init__(self): + if not self.created_at: + self.created_at = datetime.now().isoformat() + if not self.updated_at: + self.updated_at = datetime.now().isoformat() + + def to_dict(self) -> Dict[str, Any]: + """转换为字典""" + return asdict(self) + + def to_safe_dict(self) -> Dict[str, Any]: + """转换为安全字典（不包含密码）""" + data = self.to_dict() + data.pop('password_hash', None) + return data + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> 'User': + """从字典创建用户对象""" + return cls(**data) + + def update_last_login(self): + """更新最后登录时间""" + self.last_login = datetime.now().isoformat() + self.updated_at = datetime.now().isoformat() + + +@dataclass +class LoginRequest: + """登录请求数据""" + username_or_email: str + password: str + + +@dataclass +class RegisterRequest: + """注册请求数据""" + username: str + email: str + password: str + display_name: Optional[str] = None + + def __post_init__(self): + if not self.display_name: + self.display_name = self.username + + +@dataclass +class AuthResponse: + """认证响应数据""" + success: bool + message: str + user: Optional[Dict[str, Any]] = None + token: Optional[str] = None + expires_at: Optional[str] = None + + +def hash_password(password: str) -> str: + """密码哈希""" + return hashlib.sha256(password.encode()).hexdigest() + + +def verify_password(password: str, password_hash: str) -> bool: + """验证密码""" + return hash_password(password) == password_hash + + +def generate_user_id() -> str: + """生成用户ID""" + return str(uuid.uuid4()) diff --git a/python_core/requirements.txt b/python_core/requirements.txt index 0762924..8214c73 100644 --- a/python_core/requirements.txt +++ b/python_core/requirements.txt @@ -30,4 +30,5 @@ rich langgraph json-rpc langchain_core -langchain_anthropic \ No newline at end of file +langchain_anthropic +PyJWT \ No newline at end of file diff --git a/python_core/services/auth_service.py b/python_core/services/auth_service.py new file mode 100644 index 0000000..991a1b1 --- /dev/null +++ b/python_core/services/auth_service.py @@ -0,0 +1,263 @@ +""" +用户认证服务 +""" + +import re +from typing import Optional, Dict, Any +from datetime import datetime + +from python_core.models.user import ( + User, LoginRequest, RegisterRequest, AuthResponse, + hash_password, verify_password +) +from python_core.services.user_storage import user_storage +from python_core.utils.jwt_auth import generate_access_token, verify_access_token +from python_core.utils.logger import logger + + +class AuthService: + """用户认证服务""" + + def __init__(self): + self.user_storage = user_storage + logger.info("AuthService initialized") + + def register(self, request: RegisterRequest) -> AuthResponse: + """ + 用户注册 + + Args: + request: 注册请求数据 + + Returns: + AuthResponse: 注册响应 + """ + try: + # 验证输入数据 + validation_error = self._validate_register_request(request) + if validation_error: + return AuthResponse( + success=False, + message=validation_error + ) + + # 密码哈希 + password_hash = hash_password(request.password) + + # 创建用户 + user = self.user_storage.create_user( + username=request.username, + email=request.email, + password_hash=password_hash, + display_name=request.display_name + ) + + # 生成JWT token + token_info = generate_access_token( + user_id=user.id, + username=user.username, + email=user.email + ) + + logger.info(f"User registered successfully: {user.username}") + + return AuthResponse( + success=True, + message="注册成功", + user=user.to_safe_dict(), + token=token_info["token"], + expires_at=token_info["expires_at"] + ) + + except ValueError as e: + logger.warning(f"Registration failed: {e}") + return AuthResponse( + success=False, + message=str(e) + ) + except Exception as e: + logger.error(f"Registration error: {e}") + return AuthResponse( + success=False, + message="注册失败，请稍后重试" + ) + + def login(self, request: LoginRequest) -> AuthResponse: + """ + 用户登录 + + Args: + request: 登录请求数据 + + Returns: + AuthResponse: 登录响应 + """ + try: + # 验证输入数据 + validation_error = self._validate_login_request(request) + if validation_error: + return AuthResponse( + success=False, + message=validation_error + ) + + # 查找用户 + user = self.user_storage.get_user_by_username_or_email(request.username_or_email) + if not user: + return AuthResponse( + success=False, + message="用户名或密码错误" + ) + + # 检查用户状态 + if not user.is_active: + return AuthResponse( + success=False, + message="账户已被禁用" + ) + + # 验证密码 + if not verify_password(request.password, user.password_hash): + return AuthResponse( + success=False, + message="用户名或密码错误" + ) + + # 更新最后登录时间 + user.update_last_login() + self.user_storage.update_user(user) + + # 生成JWT token + token_info = generate_access_token( + user_id=user.id, + username=user.username, + email=user.email + ) + + logger.info(f"User logged in successfully: {user.username}") + + return AuthResponse( + success=True, + message="登录成功", + user=user.to_safe_dict(), + token=token_info["token"], + expires_at=token_info["expires_at"] + ) + + except Exception as e: + logger.error(f"Login error: {e}") + return AuthResponse( + success=False, + message="登录失败，请稍后重试" + ) + + def verify_token(self, token: str) -> Optional[Dict[str, Any]]: + """ + 验证JWT token + + Args: + token: JWT token字符串 + + Returns: + Dict: 用户信息，验证失败返回None + """ + try: + # 验证token + payload = verify_access_token(token) + if not payload: + return None + + # 获取用户信息 + user = self.user_storage.get_user_by_id(payload["user_id"]) + if not user or not user.is_active: + return None + + return { + "user_id": user.id, + "username": user.username, + "email": user.email, + "display_name": user.display_name, + "avatar_url": user.avatar_url, + "last_login": user.last_login + } + + except Exception as e: + logger.error(f"Token verification error: {e}") + return None + + def get_current_user(self, token: str) -> Optional[User]: + """ + 根据token获取当前用户 + + Args: + token: JWT token字符串 + + Returns: + User: 用户对象，失败返回None + """ + try: + payload = verify_access_token(token) + if not payload: + return None + + user = self.user_storage.get_user_by_id(payload["user_id"]) + if not user or not user.is_active: + return None + + return user + + except Exception as e: + logger.error(f"Get current user error: {e}") + return None + + def _validate_register_request(self, request: RegisterRequest) -> Optional[str]: + """验证注册请求数据""" + + # 验证用户名 + if not request.username or len(request.username.strip()) < 3: + return "用户名至少需要3个字符" + + if len(request.username) > 50: + return "用户名不能超过50个字符" + + if not re.match(r'^[a-zA-Z0-9_]+$', request.username): + return "用户名只能包含字母、数字和下划线" + + # 验证邮箱 + if not request.email: + return "邮箱地址不能为空" + + email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' + if not re.match(email_pattern, request.email): + return "邮箱地址格式不正确" + + # 验证密码 + if not request.password: + return "密码不能为空" + + if len(request.password) < 6: + return "密码至少需要6个字符" + + if len(request.password) > 100: + return "密码不能超过100个字符" + + # 验证显示名称 + if request.display_name and len(request.display_name) > 100: + return "显示名称不能超过100个字符" + + return None + + def _validate_login_request(self, request: LoginRequest) -> Optional[str]: + """验证登录请求数据""" + + if not request.username_or_email: + return "用户名或邮箱不能为空" + + if not request.password: + return "密码不能为空" + + return None + + +# 创建全局认证服务实例 +auth_service = AuthService() diff --git a/python_core/services/resource_category_manager.py b/python_core/services/resource_category_manager.py index 04125ab..4491baa 100644 --- a/python_core/services/resource_category_manager.py +++ b/python_core/services/resource_category_manager.py @@ -144,91 +144,3 @@ class ResourceCategoryManager: results.append(asdict(category)) return results - - -# 全局实例 -resource_category_manager = ResourceCategoryManager() - - -def main(): - """命令行接口 - 使用JSON-RPC协议""" - import sys - import json - - # 创建响应处理器 - rpc = create_response_handler() - - if len(sys.argv) < 2: - rpc.error("INVALID_REQUEST", "No command specified") - return - - command = sys.argv[1] - - try: - if command == "get_all_categories": - categories = resource_category_manager.get_all_categories() - rpc.success(categories) - - elif command == "get_category_by_id": - if len(sys.argv) < 3: - rpc.error("INVALID_REQUEST", "Category ID required") - return - category_id = sys.argv[2] - category = resource_category_manager.get_category_by_id(category_id) - if category: - rpc.success(category) - else: - rpc.error("NOT_FOUND", "Category not found") - - elif command == "create_category": - if len(sys.argv) < 5: - rpc.error("INVALID_REQUEST", "Title, prompt and color required") - return - title, ai_prompt, color = sys.argv[2], sys.argv[3], sys.argv[4] - result = resource_category_manager.create_category(title, ai_prompt, color) - rpc.success(result) - - elif command == "update_category": - if len(sys.argv) < 4: - rpc.error("INVALID_REQUEST", "Category ID and update data required") - return - category_id = sys.argv[2] - update_data = json.loads(sys.argv[3]) - result = resource_category_manager.update_category( - category_id, - update_data.get('title'), - update_data.get('ai_prompt'), - update_data.get('color'), - update_data.get('is_active') - ) - if result: - rpc.success(result) - else: - rpc.error("NOT_FOUND", "Category not found") - - elif command == "delete_category": - if len(sys.argv) < 3: - rpc.error("INVALID_REQUEST", "Category ID required") - return - category_id = sys.argv[2] - success = resource_category_manager.delete_category(category_id) - rpc.success(success) - - elif command == "search_categories": - if len(sys.argv) < 3: - rpc.error("INVALID_REQUEST", "Search keyword required") - return - keyword = sys.argv[2] - results = resource_category_manager.search_categories(keyword) - rpc.success(results) - - else: - rpc.error("INVALID_REQUEST", f"Unknown command: {command}") - - except Exception as e: - logger.error(f"Command execution failed: {e}") - rpc.error("INTERNAL_ERROR", str(e)) - - -if __name__ == "__main__": - main() diff --git a/python_core/services/user_storage.py b/python_core/services/user_storage.py new file mode 100644 index 0000000..550e096 --- /dev/null +++ b/python_core/services/user_storage.py @@ -0,0 +1,249 @@ +""" +用户数据存储服务 +""" + +import json +import os +from pathlib import Path +from typing import List, Optional, Dict, Any +from datetime import datetime + +from python_core.models.user import User, generate_user_id +from python_core.config import settings +from python_core.utils.logger import logger + + +class UserStorage: + """用户数据存储服务""" + + def __init__(self): + # 用户数据文件路径 + self.users_file = Path(settings.temp_dir) / "cache" / "users.json" + self.users_file.parent.mkdir(parents=True, exist_ok=True) + + # 内存中的用户数据 + self._users: Dict[str, User] = {} + + # 加载用户数据 + self._load_users() + + logger.info(f"UserStorage initialized with {len(self._users)} users") + + def _load_users(self): + """从文件加载用户数据""" + try: + if self.users_file.exists(): + with open(self.users_file, 'r', encoding='utf-8') as f: + users_data = json.load(f) + + self._users = {} + for user_data in users_data: + user = User.from_dict(user_data) + self._users[user.id] = user + + logger.info(f"Loaded {len(self._users)} users from {self.users_file}") + else: + self._users = {} + logger.info("No existing users file found, starting with empty user database") + + except Exception as e: + logger.error(f"Failed to load users: {e}") + self._users = {} + + def _save_users(self): + """保存用户数据到文件""" + try: + users_data = [user.to_dict() for user in self._users.values()] + + with open(self.users_file, 'w', encoding='utf-8') as f: + json.dump(users_data, f, ensure_ascii=False, indent=2) + + logger.debug(f"Saved {len(self._users)} users to {self.users_file}") + + except Exception as e: + logger.error(f"Failed to save users: {e}") + raise + + def create_user(self, username: str, email: str, password_hash: str, display_name: str) -> User: + """ + 创建新用户 + + Args: + username: 用户名 + email: 邮箱 + password_hash: 密码哈希 + display_name: 显示名称 + + Returns: + User: 创建的用户对象 + + Raises: + ValueError: 用户名或邮箱已存在 + """ + # 检查用户名是否已存在 + if self.get_user_by_username(username): + raise ValueError(f"Username '{username}' already exists") + + # 检查邮箱是否已存在 + if self.get_user_by_email(email): + raise ValueError(f"Email '{email}' already exists") + + # 创建新用户 + user = User( + id=generate_user_id(), + username=username, + email=email, + password_hash=password_hash, + display_name=display_name + ) + + # 保存到内存和文件 + self._users[user.id] = user + self._save_users() + + logger.info(f"Created new user: {username} ({email})") + return user + + def get_user_by_id(self, user_id: str) -> Optional[User]: + """根据用户ID获取用户""" + return self._users.get(user_id) + + def get_user_by_username(self, username: str) -> Optional[User]: + """根据用户名获取用户""" + for user in self._users.values(): + if user.username == username: + return user + return None + + def get_user_by_email(self, email: str) -> Optional[User]: + """根据邮箱获取用户""" + for user in self._users.values(): + if user.email == email: + return user + return None + + def get_user_by_username_or_email(self, username_or_email: str) -> Optional[User]: + """根据用户名或邮箱获取用户""" + # 先尝试用户名 + user = self.get_user_by_username(username_or_email) + if user: + return user + + # 再尝试邮箱 + return self.get_user_by_email(username_or_email) + + def update_user(self, user: User) -> bool: + """ + 更新用户信息 + + Args: + user: 用户对象 + + Returns: + bool: 更新是否成功 + """ + try: + if user.id not in self._users: + logger.warning(f"User {user.id} not found for update") + return False + + # 更新时间戳 + user.updated_at = datetime.now().isoformat() + + # 保存到内存和文件 + self._users[user.id] = user + self._save_users() + + logger.info(f"Updated user: {user.username}") + return True + + except Exception as e: + logger.error(f"Failed to update user {user.id}: {e}") + return False + + def delete_user(self, user_id: str) -> bool: + """ + 删除用户 + + Args: + user_id: 用户ID + + Returns: + bool: 删除是否成功 + """ + try: + if user_id not in self._users: + logger.warning(f"User {user_id} not found for deletion") + return False + + user = self._users[user_id] + del self._users[user_id] + self._save_users() + + logger.info(f"Deleted user: {user.username}") + return True + + except Exception as e: + logger.error(f"Failed to delete user {user_id}: {e}") + return False + + def get_all_users(self, include_inactive: bool = False) -> List[User]: + """ + 获取所有用户 + + Args: + include_inactive: 是否包含非活跃用户 + + Returns: + List[User]: 用户列表 + """ + users = list(self._users.values()) + + if not include_inactive: + users = [user for user in users if user.is_active] + + # 按创建时间排序 + users.sort(key=lambda x: x.created_at, reverse=True) + + return users + + def get_user_count(self) -> Dict[str, int]: + """获取用户统计信息""" + all_users = list(self._users.values()) + active_users = [user for user in all_users if user.is_active] + + return { + "total": len(all_users), + "active": len(active_users), + "inactive": len(all_users) - len(active_users) + } + + def search_users(self, keyword: str) -> List[User]: + """ + 搜索用户 + + Args: + keyword: 搜索关键词 + + Returns: + List[User]: 匹配的用户列表 + """ + keyword = keyword.lower().strip() + if not keyword: + return self.get_all_users() + + matching_users = [] + for user in self._users.values(): + if (keyword in user.username.lower() or + keyword in user.email.lower() or + keyword in user.display_name.lower()): + matching_users.append(user) + + # 按创建时间排序 + matching_users.sort(key=lambda x: x.created_at, reverse=True) + + return matching_users + + +# 创建全局用户存储实例 +user_storage = UserStorage() diff --git a/python_core/utils/jwt_auth.py b/python_core/utils/jwt_auth.py new file mode 100644 index 0000000..742bc41 --- /dev/null +++ b/python_core/utils/jwt_auth.py @@ -0,0 +1,205 @@ +""" +JWT认证工具类 +""" + +import jwt +import json +from datetime import datetime, timedelta +from typing import Optional, Dict, Any +from pathlib import Path + +from python_core.utils.logger import logger + + +class JWTAuth: + """JWT认证工具类""" + + def __init__(self, secret_key: str = "mixvideo_secret_key_2024"): + self.secret_key = secret_key + self.algorithm = "HS256" + # JWT过期时间：6个月 + self.expires_delta = timedelta(days=180) + + def generate_token(self, user_id: str, username: str, email: str) -> Dict[str, Any]: + """ + 生成JWT token + + Args: + user_id: 用户ID + username: 用户名 + email: 邮箱 + + Returns: + Dict: 包含token和过期时间的字典 + """ + try: + # 计算过期时间 + expires_at = datetime.utcnow() + self.expires_delta + + # JWT payload + payload = { + "user_id": user_id, + "username": username, + "email": email, + "iat": datetime.utcnow(), # 签发时间 + "exp": expires_at, # 过期时间 + "iss": "mixvideo", # 签发者 + "sub": user_id # 主题（用户ID） + } + + # 生成token + token = jwt.encode(payload, self.secret_key, algorithm=self.algorithm) + + logger.info(f"Generated JWT token for user: {username}") + + return { + "token": token, + "expires_at": expires_at.isoformat(), + "expires_in": int(self.expires_delta.total_seconds()) + } + + except Exception as e: + logger.error(f"Failed to generate JWT token: {e}") + raise Exception(f"Token generation failed: {str(e)}") + + def verify_token(self, token: str) -> Optional[Dict[str, Any]]: + """ + 验证JWT token + + Args: + token: JWT token字符串 + + Returns: + Dict: 解码后的payload，验证失败返回None + """ + try: + # 解码并验证token + payload = jwt.decode( + token, + self.secret_key, + algorithms=[self.algorithm], + options={"verify_exp": True} # 验证过期时间 + ) + + logger.debug(f"Token verified for user: {payload.get('username')}") + return payload + + except jwt.ExpiredSignatureError: + logger.warning("JWT token has expired") + return None + except jwt.InvalidTokenError as e: + logger.warning(f"Invalid JWT token: {e}") + return None + except Exception as e: + logger.error(f"JWT token verification failed: {e}") + return None + + def refresh_token(self, token: str) -> Optional[Dict[str, Any]]: + """ + 刷新JWT token + + Args: + token: 当前的JWT token + + Returns: + Dict: 新的token信息，失败返回None + """ + try: + # 验证当前token（忽略过期时间） + payload = jwt.decode( + token, + self.secret_key, + algorithms=[self.algorithm], + options={"verify_exp": False} # 不验证过期时间 + ) + + # 生成新token + new_token_info = self.generate_token( + user_id=payload["user_id"], + username=payload["username"], + email=payload["email"] + ) + + logger.info(f"Token refreshed for user: {payload['username']}") + return new_token_info + + except Exception as e: + logger.error(f"Token refresh failed: {e}") + return None + + def decode_token_without_verification(self, token: str) -> Optional[Dict[str, Any]]: + """ + 解码token但不验证（用于获取过期token的信息） + + Args: + token: JWT token字符串 + + Returns: + Dict: 解码后的payload + """ + try: + payload = jwt.decode( + token, + options={"verify_signature": False, "verify_exp": False} + ) + return payload + except Exception as e: + logger.error(f"Token decode failed: {e}") + return None + + def get_token_info(self, token: str) -> Dict[str, Any]: + """ + 获取token信息 + + Args: + token: JWT token字符串 + + Returns: + Dict: token信息 + """ + payload = self.decode_token_without_verification(token) + if not payload: + return {"valid": False, "error": "Invalid token format"} + + try: + exp_timestamp = payload.get("exp") + if exp_timestamp: + exp_datetime = datetime.fromtimestamp(exp_timestamp) + is_expired = datetime.utcnow() > exp_datetime + time_remaining = exp_datetime - datetime.utcnow() + else: + is_expired = True + time_remaining = timedelta(0) + + return { + "valid": not is_expired, + "user_id": payload.get("user_id"), + "username": payload.get("username"), + "email": payload.get("email"), + "issued_at": payload.get("iat"), + "expires_at": payload.get("exp"), + "is_expired": is_expired, + "time_remaining": str(time_remaining) if not is_expired else "Expired" + } + + except Exception as e: + return {"valid": False, "error": str(e)} + + +# 创建全局JWT认证实例 +jwt_auth = JWTAuth() + + +def generate_access_token(user_id: str, username: str, email: str) -> Dict[str, Any]: + """生成访问token的便捷函数""" + return jwt_auth.generate_token(user_id, username, email) + + +def verify_access_token(token: str) -> Optional[Dict[str, Any]]: + """验证访问token的便捷函数""" + return jwt_auth.verify_token(token) + + +def refresh_access_token(token: str) -> Optional[Dict[str, Any]]: + """刷新访问token的便捷函数""" + return jwt_auth.refresh_token(token) diff --git a/test_auth_system.py b/test_auth_system.py new file mode 100644 index 0000000..f906836 --- /dev/null +++ b/test_auth_system.py @@ -0,0 +1,316 @@ +#!/usr/bin/env python3 +""" +测试Python用户认证系统 +""" + +import sys +import os +sys.path.insert(0, '/root/projects/mixvideo_v2') + +from python_core.api.auth_api import auth_api +from python_core.utils.jwt_auth import jwt_auth +from python_core.services.user_storage import user_storage + + +def test_user_registration(): + """测试用户注册""" + print("🧪 测试用户注册...") + + try: + # 注册测试用户 + result = auth_api.register({ + "username": "testuser", + "email": "test@example.com", + "password": "password123", + "display_name": "测试用户" + }) + + if result["success"]: + print(f"✅ 注册成功: {result['data']['user']['username']}") + print(f" 用户ID: {result['data']['user']['id']}") + print(f" Token: {result['data']['token'][:50]}...") + return result['data']['token'] + else: + print(f"❌ 注册失败: {result['message']}") + return None + + except Exception as e: + print(f"❌ 注册测试失败: {e}") + return None + + +def test_user_login(): + """测试用户登录""" + print("\n🧪 测试用户登录...") + + try: + # 登录测试用户 + result = auth_api.login({ + "username_or_email": "testuser", + "password": "password123" + }) + + if result["success"]: + print(f"✅ 登录成功: {result['data']['user']['username']}") + print(f" 显示名称: {result['data']['user']['display_name']}") + print(f" Token: {result['data']['token'][:50]}...") + return result['data']['token'] + else: + print(f"❌ 登录失败: {result['message']}") + return None + + except Exception as e: + print(f"❌ 登录测试失败: {e}") + return None + + +def test_token_verification(token): + """测试Token验证""" + print("\n🧪 测试Token验证...") + + try: + # 验证token + result = auth_api.verify_token({ + "token": token + }) + + if result["success"]: + user = result['data']['user'] + print(f"✅ Token验证成功: {user['username']}") + print(f" 用户ID: {user['user_id']}") + print(f" 邮箱: {user['email']}") + + # 获取token详细信息 + token_info = jwt_auth.get_token_info(token) + print(f" Token有效性: {token_info['valid']}") + print(f" 剩余时间: {token_info['time_remaining']}") + + return True + else: + print(f"❌ Token验证失败: {result['message']}") + return False + + except Exception as e: + print(f"❌ Token验证测试失败: {e}") + return False + + +def test_get_current_user(token): + """测试获取当前用户""" + print("\n🧪 测试获取当前用户...") + + try: + # 获取当前用户 + result = auth_api.get_current_user({ + "token": token + }) + + if result["success"]: + user = result['data']['user'] + print(f"✅ 获取用户成功: {user['username']}") + print(f" 显示名称: {user['display_name']}") + print(f" 创建时间: {user['created_at']}") + print(f" 最后登录: {user['last_login']}") + return True + else: + print(f"❌ 获取用户失败: {result['message']}") + return False + + except Exception as e: + print(f"❌ 获取用户测试失败: {e}") + return False + + +def test_user_storage(): + """测试用户存储""" + print("\n🧪 测试用户存储...") + + try: + # 获取所有用户 + users = user_storage.get_all_users() + print(f"✅ 获取到 {len(users)} 个用户") + + # 获取用户统计 + stats = user_storage.get_user_count() + print(f" 总用户数: {stats['total']}") + print(f" 活跃用户: {stats['active']}") + print(f" 禁用用户: {stats['inactive']}") + + # 搜索用户 + search_results = user_storage.search_users("test") + print(f" 搜索'test'找到 {len(search_results)} 个用户") + + return True + + except Exception as e: + print(f"❌ 用户存储测试失败: {e}") + return False + + +def test_duplicate_registration(): + """测试重复注册""" + print("\n🧪 测试重复注册...") + + try: + # 尝试重复注册 + result = auth_api.register({ + "username": "testuser", + "email": "test@example.com", + "password": "password123", + "display_name": "重复用户" + }) + + if not result["success"]: + print(f"✅ 正确阻止重复注册: {result['message']}") + return True + else: + print(f"❌ 未能阻止重复注册") + return False + + except Exception as e: + print(f"❌ 重复注册测试失败: {e}") + return False + + +def test_invalid_login(): + """测试无效登录""" + print("\n🧪 测试无效登录...") + + try: + # 尝试错误密码登录 + result = auth_api.login({ + "username_or_email": "testuser", + "password": "wrongpassword" + }) + + if not result["success"]: + print(f"✅ 正确拒绝错误密码: {result['message']}") + else: + print(f"❌ 未能拒绝错误密码") + return False + + # 尝试不存在的用户登录 + result = auth_api.login({ + "username_or_email": "nonexistentuser", + "password": "password123" + }) + + if not result["success"]: + print(f"✅ 正确拒绝不存在用户: {result['message']}") + return True + else: + print(f"❌ 未能拒绝不存在用户") + return False + + except Exception as e: + print(f"❌ 无效登录测试失败: {e}") + return False + + +def test_jwt_features(): + """测试JWT特性""" + print("\n🧪 测试JWT特性...") + + try: + # 生成token + token_info = jwt_auth.generate_token("test_user_id", "testuser", "test@example.com") + token = token_info["token"] + print(f"✅ 生成Token成功") + print(f" 过期时间: {token_info['expires_at']}") + print(f" 有效期: {token_info['expires_in']} 秒") + + # 验证token + payload = jwt_auth.verify_token(token) + if payload: + print(f"✅ Token验证成功") + print(f" 用户ID: {payload['user_id']}") + print(f" 用户名: {payload['username']}") + print(f" 签发者: {payload['iss']}") + else: + print(f"❌ Token验证失败") + return False + + # 获取token信息 + info = jwt_auth.get_token_info(token) + print(f"✅ Token信息获取成功") + print(f" 有效性: {info['valid']}") + print(f" 剩余时间: {info['time_remaining']}") + + return True + + except Exception as e: + print(f"❌ JWT特性测试失败: {e}") + return False + + +def main(): + """主测试函数""" + print("🚀 开始测试Python用户认证系统...") + + test_results = [] + + # 测试JWT特性 + test_results.append(("JWT特性", test_jwt_features())) + + # 测试用户注册 + token = test_user_registration() + test_results.append(("用户注册", token is not None)) + + if token: + # 测试Token验证 + test_results.append(("Token验证", test_token_verification(token))) + + # 测试获取当前用户 + test_results.append(("获取当前用户", test_get_current_user(token))) + + # 测试用户登录 + login_token = test_user_login() + test_results.append(("用户登录", login_token is not None)) + + # 测试用户存储 + test_results.append(("用户存储", test_user_storage())) + + # 测试重复注册 + test_results.append(("重复注册检查", test_duplicate_registration())) + + # 测试无效登录 + test_results.append(("无效登录检查", test_invalid_login())) + + # 显示测试结果 + print("\n📊 测试结果汇总:") + print("=" * 50) + + passed = 0 + total = len(test_results) + + for test_name, result in test_results: + status = "✅ 通过" if result else "❌ 失败" + print(f"{test_name:20} {status}") + if result: + passed += 1 + + print("=" * 50) + print(f"总计: {passed}/{total} 个测试通过") + + if passed == total: + print("\n🎉 所有测试通过！Python用户认证系统工作正常！") + + print("\n📖 使用方法:") + print(" # 注册用户") + print(" python3 -m python_core.cli auth register username email@example.com") + print(" # 登录用户") + print(" python3 -m python_core.cli auth login username") + print(" # 验证Token") + print(" python3 -m python_core.cli auth verify ") + print(" # 查看用户列表") + print(" python3 -m python_core.cli auth list") + print(" # 查看统计信息") + print(" python3 -m python_core.cli auth stats") + + else: + print(f"\n❌ {total - passed} 个测试失败！") + sys.exit(1) + + +if __name__ == "__main__": + main()