imetting/backend/test/test_api_security.py

#!/usr/bin/env python3
"""
API安全性测试脚本
测试添加JWT验证后，API端点是否正确拒绝未授权访问

运行方法:
cd /Users/jiliu/工作/projects/imeeting/backend
source venv/bin/activate
python test/test_api_security.py
"""
import requests
import json

BASE_URL = "http://127.0.0.1:8000"
PROXIES = {'http': None, 'https': None}

def test_unauthorized_access():
    """测试未授权访问各个API端点"""
    print("=== API安全性测试 ===")
    print("测试未授权访问是否被正确拒绝\n")
    
    # 需要验证的API端点
    protected_endpoints = [
        # Users endpoints
        ("GET", "/api/users", "获取所有用户"),
        ("GET", "/api/users/1", "获取用户详情"),
        
        # Meetings endpoints  
        ("GET", "/api/meetings", "获取会议列表"),
        ("GET", "/api/meetings/1", "获取会议详情"),
        ("GET", "/api/meetings/1/transcript", "获取会议转录"),
        ("GET", "/api/meetings/1/edit", "获取会议编辑信息"),
        ("GET", "/api/meetings/1/audio", "获取会议音频"),
        ("POST", "/api/meetings/1/regenerate-summary", "重新生成摘要"),
        ("GET", "/api/meetings/1/summaries", "获取会议摘要"),
        ("GET", "/api/meetings/1/transcription/status", "获取转录状态"),
        
        # Auth endpoints (需要token的)
        ("GET", "/api/auth/me", "获取用户信息"),
        ("POST", "/api/auth/logout", "登出"),
        ("POST", "/api/auth/logout-all", "登出所有设备"),
    ]
    
    success_count = 0
    total_count = len(protected_endpoints)
    
    for method, endpoint, description in protected_endpoints:
        try:
            url = f"{BASE_URL}{endpoint}"
            
            if method == "GET":
                response = requests.get(url, proxies=PROXIES, timeout=5)
            elif method == "POST":
                response = requests.post(url, proxies=PROXIES, timeout=5)
            elif method == "PUT":
                response = requests.put(url, proxies=PROXIES, timeout=5)
            elif method == "DELETE":
                response = requests.delete(url, proxies=PROXIES, timeout=5)
            
            if response.status_code == 401:
                print(f"✅ {method} {endpoint} - {description}")
                print(f"   正确返回401 Unauthorized")
                success_count += 1
            else:
                print(f"❌ {method} {endpoint} - {description}")
                print(f"   错误：返回 {response.status_code}，应该返回401")
                print(f"   响应: {response.text[:100]}...")
                
        except requests.exceptions.RequestException as e:
            print(f"❌ {method} {endpoint} - {description}")
            print(f"   请求异常: {e}")
        
        print()
    
    print(f"=== 测试结果 ===")
    print(f"通过: {success_count}/{total_count}")
    print(f"成功率: {success_count/total_count*100:.1f}%")
    
    if success_count == total_count:
        print("🎉 所有API端点都正确实施了JWT验证！")
    else:
        print("⚠️  有些API端点未正确实施JWT验证，需要修复")
    
    return success_count == total_count

def test_valid_token_access():
    """测试有效token的访问"""
    print("\n=== 测试有效Token访问 ===")
    
    # 1. 先登录获取token
    login_data = {"username": "mula", "password": "781126"}
    try:
        response = requests.post(f"{BASE_URL}/api/auth/login", json=login_data, proxies=PROXIES)
        if response.status_code != 200:
            print("❌ 无法登录获取测试token")
            print(f"登录响应: {response.status_code} - {response.text}")
            return False
            
        user_data = response.json()
        token = user_data["token"]
        headers = {"Authorization": f"Bearer {token}"}
        
        print(f"✅ 登录成功，获得token")
        
        # 2. 测试几个主要API端点
        test_endpoints = [
            ("GET", "/api/auth/me", "获取当前用户信息"),
            ("GET", "/api/users", "获取用户列表"),
            ("GET", "/api/meetings", "获取会议列表"),
        ]
        
        success_count = 0
        for method, endpoint, description in test_endpoints:
            try:
                url = f"{BASE_URL}{endpoint}"
                response = requests.get(url, headers=headers, proxies=PROXIES, timeout=5)
                
                if response.status_code == 200:
                    print(f"✅ {method} {endpoint} - {description}")
                    print(f"   正确返回200 OK")
                    success_count += 1
                elif response.status_code == 500:
                    print(f"⚠️  {method} {endpoint} - {description}")
                    print(f"   返回500 (可能是数据库连接问题，但JWT验证通过了)")
                    success_count += 1
                else:
                    print(f"❌ {method} {endpoint} - {description}")
                    print(f"   意外响应: {response.status_code}")
                    print(f"   响应内容: {response.text[:100]}...")
                    
            except requests.exceptions.RequestException as e:
                print(f"❌ {method} {endpoint} - {description}")
                print(f"   请求异常: {e}")
        
        print(f"\n有效token测试: {success_count}/{len(test_endpoints)} 通过")
        return success_count == len(test_endpoints)
        
    except Exception as e:
        print(f"❌ 测试失败: {e}")
        return False

if __name__ == "__main__":
    print("API JWT安全性测试工具")
    print("=" * 50)
    
    # 测试未授权访问
    unauthorized_ok = test_unauthorized_access()
    
    # 测试授权访问
    authorized_ok = test_valid_token_access()
    
    print("\n" + "=" * 50)
    if unauthorized_ok and authorized_ok:
        print("🎉 JWT验证实施成功！")
        print("✅ 未授权访问被正确拒绝")
        print("✅ 有效token可以正常访问")
    else:
        print("⚠️  JWT验证实施不完整")
        if not unauthorized_ok:
            print("❌ 部分API未正确拒绝未授权访问")
        if not authorized_ok:
            print("❌ 有效token访问存在问题")