"""
安全测试套件
测试内容：
1. SQL注入测试
2. XSS攻击测试
3. CSRF保护测试
4. 认证授权测试
5. 输入验证测试
"""
import pytest
import httpx
from typing import Dict, Any
class SecurityTestBase:
"""安全测试基类"""
def __init__(self, base_url: str = "http://localhost:8084"):
self.base_url = base_url
self.client = httpx.Client(timeout=30.0)
self.token = None
def login(self, username: str = "admin", password: str = "admin123") -> str:
"""登录获取token"""
response = self.client.post(
f"{self.base_url}/api/auth/login",
json={"username": username, "password": password}
)
assert response.status_code == 200
data = response.json()
return data.get("token")
def setup_auth(self):
"""设置认证token"""
if not self.token:
self.token = self.login()
def get_headers(self) -> Dict[str, str]:
"""获取请求头"""
headers = {"Content-Type": "application/json"}
if self.token:
headers["Authorization"] = f"Bearer {self.token}"
return headers
def cleanup(self):
"""清理资源"""
self.client.close()
class TestSQLInjection(SecurityTestBase):
"""SQL注入测试"""
@pytest.fixture(autouse=True)
def setup(self):
self.setup_auth()
yield
self.cleanup()
def test_sql_injection_in_login(self):
"""测试登录接口的SQL注入防护"""
malicious_inputs = [
"admin' OR '1'='1",
"admin' --",
"admin' #",
"admin'/*",
"admin' or 1=1--",
"admin' union select * from users--",
]
for payload in malicious_inputs:
response = self.client.post(
f"{self.base_url}/api/auth/login",
json={"username": payload, "password": "password"}
)
# 应该返回401（认证失败），而不是绕过认证
assert response.status_code == 401, f"SQL注入攻击未阻止: {payload}"
def test_sql_injection_in_user_search(self):
"""测试用户搜索接口的SQL注入防护"""
self.setup_auth()
malicious_inputs = [
"test' OR '1'='1",
"test' UNION SELECT * FROM users--",
"test'; DROP TABLE users--",
"1' OR 1=1--",
]
for payload in malicious_inputs:
response = self.client.get(
f"{self.base_url}/api/users",
params={"username": payload},
headers=self.get_headers()
)
# 应该返回400（错误请求）或正常结果，但不应该暴露数据库错误
assert response.status_code in [200, 400], f"SQL注入攻击未正确处理: {payload}"
class TestXSS(SecurityTestBase):
"""XSS攻击测试"""
@pytest.fixture(autouse=True)
def setup(self):
self.setup_auth()
yield
self.cleanup()
def test_xss_in_user_creation(self):
"""测试用户创建接口的XSS防护"""
xss_payloads = [
"",
"
",
"