backend/tests/test_sql_injection.py

import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.core.config import settings
from app.core.security import create_access_token
from app.models.user import User
from datetime import timedelta

client = TestClient(app)

def test_sql_injection_selection() -> None:
    """测试选股接口是否防护SQL注入"""
    # 创建测试用户token
    access_token_expires: timedelta = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token: str = create_access_token(
        data={"sub": "test_user"},
        expires_delta=access_token_expires
    )
    
    # 尝试SQL注入攻击
    injection_payload: str = "1' OR '1'='1'--"
    response = client.get(
        f"/api/v1/strategies/selections?strategy_id={injection_payload}",
        headers={"Authorization": f"Bearer {access_token}"}
    )
    
    # 验证系统返回空结果而非敏感数据
    assert response.status_code == 200
    assert response.json() == []

def test_sql_injection_portfolio() -> None:
    """测试投资组合接口是否防护SQL注入"""
    # 创建测试用户token
    access_token_expires: timedelta = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token: str = create_access_token(
        data={"sub": "test_user"},
        expires_delta=access_token_expires
    )
    
    # 尝试SQL注入攻击
    injection_payload: str = "1; DROP TABLE portfolios;--"
    response = client.get(
        f"/api/v1/portfolios/{injection_payload}",
        headers={"Authorization": f"Bearer {access_token}"}
    )
    
    # 验证系统返回404而非执行恶意SQL
    assert response.status_code == 404
    assert response.json()["detail"] == "投资组合未找到"

def test_sql_injection_register() -> None:
    """测试用户注册接口是否防护SQL注入"""
    # 尝试SQL注入攻击
    injection_payload_username: str = "test_user_sql' OR '1'='1'--"
    injection_payload_email: str = "test_sql@example.com' OR '1'='1'--"
    
    # 测试用户名注入
    response_username = client.post(
        "/api/v1/auth/register",
        json={
            "username": injection_payload_username,
            "email": "test_sql_username@example.com",
            "password": "testpassword"
        }
    )
    # 验证系统返回错误或已存在信息，而不是成功注册
    assert response_username.status_code == 400
    assert "用户名已存在" in response_username.json()["detail"] or "注册失败" in response_username.json()["detail"]

    # 测试邮箱注入
    response_email = client.post(
        "/api/v1/auth/register",
        json={
            "username": "test_user_email_sql",
            "email": injection_payload_email,
            "password": "testpassword"
        }
    )
    # 验证系统返回错误或已存在信息，而不是成功注册
    assert response_email.status_code == 400
    assert "邮箱已被注册" in response_email.json()["detail"] or "注册失败" in response_email.json()["detail"]

def test_sql_injection_login() -> None:
    """测试用户登录接口是否防护SQL注入"""
    # 尝试SQL注入攻击
    injection_payload_username: str = "admin' OR '1'='1'--"
    injection_payload_password: str = "password' OR '1'='1'--"

    # 尝试用户名注入
    response_username = client.post(
        "/api/v1/auth/login",
        data={
            "username": injection_payload_username,
            "password": "anypassword"
        }
    )
    # 验证系统返回未授权错误
    assert response_username.status_code == 401
    assert response_username.json()["detail"] == "用户名或密码错误"

    # 尝试密码注入
    response_password = client.post(
        "/api/v1/auth/login",
        data={
            "username": "test_user", # 使用一个存在的或不存在的用户名
            "password": injection_payload_password
        }
    )
    # 验证系统返回未授权错误
    assert response_password.status_code == 401
    assert response_password.json()["detail"] == "用户名或密码错误"

def test_sql_injection_get_backtest_results() -> None:
    """测试获取回测结果列表接口是否防护SQL注入"""
    # 创建测试用户token
    access_token_expires: timedelta = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token: str = create_access_token(
        data={"sub": "test_user"},
        expires_delta=access_token_expires
    )

    # 尝试strategy_id参数注入
    injection_payload: str = "1' OR '1'='1'--"
    response = client.get(
        f"/api/v1/backtests/?strategy_id={injection_payload}",
        headers={"Authorization": f"Bearer {access_token}"}
    )
    # 验证系统返回422 Unprocessable Entity，因为strategy_id期望整数
    assert response.status_code == 422
    assert "value is not a valid integer" in response.json()["detail"][0]["msg"]

def test_sql_injection_create_portfolio() -> None:
    """测试创建投资组合接口是否防护SQL注入"""
    # 创建测试用户token
    access_token_expires: timedelta = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token: str = create_access_token(
        data={"sub": "test_user"},
        expires_delta=access_token_expires
    )

    # 尝试name参数注入
    injection_payload_name: str = "My Portfolio' OR '1'='1'--"
    response = client.post(
        "/api/v1/portfolios/",
        headers={"Authorization": f"Bearer {access_token}"},
        json={
            "name": injection_payload_name,
            "description": "A test portfolio"
        }
    )
    # 验证系统返回成功，但注入内容被转义或过滤
    # 由于Pydantic和ORM的防护，这里预期是成功创建，但名称是字面量
    assert response.status_code == 201
    assert response.json()["name"] == injection_payload_name

def test_sql_injection_search_stocks() -> None:
    """测试搜索股票接口是否防护SQL注入"""
    # 创建测试用户token
    access_token_expires: timedelta = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token: str = create_access_token(
        data={"sub": "test_user"},
        expires_delta=access_token_expires
    )

    # 尝试query参数注入
    injection_payload: str = "AAPL' OR '1'='1'--"
    response = client.get(
        f"/api/v1/stocks/search?query={injection_payload}",
        headers={"Authorization": f"Bearer {access_token}"}
    )
    # 验证系统返回空列表或正常结果，而不是敏感数据
    assert response.status_code == 200
    # 预期是空列表，因为注入的查询不会匹配任何有效股票
    assert response.json() == []

def test_sql_injection_get_stock_ranking() -> None:
    """测试获取股票排名接口是否防护SQL注入"""
    # 创建测试用户token
    access_token_expires: timedelta = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token: str = create_access_token(
        data={"sub": "test_user"},
        expires_delta=access_token_expires
    )

    # 尝试ranking_type路径参数注入
    injection_payload: str = "gainers' OR '1'='1'--"
    response = client.get(
        f"/api/v1/stocks/ranking/{injection_payload}",
        headers={"Authorization": f"Bearer {access_token}"}
    )
    # 验证系统返回422 Unprocessable Entity 或 500 Internal Server Error
    # 因为ranking_type应该是一个预定义的值，注入会导致无效的类型或查询
    assert response.status_code == 422 or response.status_code == 500
    # 预期是422，因为FastAPI会验证路径参数的有效性
    if response.status_code == 422:
        assert "value is not a valid enumeration member" in response.json()["detail"][0]["msg"]
    else:
        assert "获取股票排名失败" in response.json()["detail"]

def test_sql_injection_get_strategy() -> None:
    """测试获取策略详情接口是否防护SQL注入"""
    # 创建测试用户token
    access_token_expires: timedelta = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token: str = create_access_token(
        data={"sub": "test_user"},
        expires_delta=access_token_expires
    )

    # 尝试strategy_id路径参数注入
    injection_payload: str = "invalid_uuid' OR '1'='1'--"
    response = client.get(
        f"/api/v1/strategies/{injection_payload}",
        headers={"Authorization": f"Bearer {access_token}"}
    )
    # 验证系统返回404或500，而不是敏感数据
    assert response.status_code == 404
    assert response.json()["detail"] == "策略不存在"

def test_sql_injection_update_strategy() -> None:
    """测试更新策略接口是否防护SQL注入"""
    # 创建测试用户token
    access_token_expires: timedelta = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token: str = create_access_token(
        data={"sub": "test_user"},
        expires_delta=access_token_expires
    )

    # 首先创建一个策略用于更新
    create_response = client.post(
        "/api/v1/strategies/",
        headers={"Authorization": f"Bearer {access_token}"},
        json={
            "name": "Strategy to Update",
            "description": "Original description",
            "strategy_type": "momentum",
            "parameters": {}
        }
    )
    assert create_response.status_code == 201
    strategy_id = create_response.json()["id"]

    # 尝试name参数注入
    injection_payload_name: str = "Updated Strategy' OR '1'='1'--"
    response = client.put(
        f"/api/v1/strategies/{strategy_id}",
        headers={"Authorization": f"Bearer {access_token}"},
        json={
            "name": injection_payload_name,
            "description": "Updated description"
        }
    )
    # 验证系统返回成功，但注入内容被转义或过滤
    assert response.status_code == 200
    assert response.json()["name"] == injection_payload_name

def test_sql_injection_delete_strategy() -> None:
    """测试删除策略接口是否防护SQL注入"""
    # 创建测试用户token
    access_token_expires: timedelta = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token: str = create_access_token(
        data={"sub": "test_user"},
        expires_delta=access_token_expires
    )

    # 首先创建一个策略用于删除
    create_response = client.post(
        "/api/v1/strategies/",
        headers={"Authorization": f"Bearer {access_token}"},
        json={
            "name": "Strategy to Delete",
            "description": "Description",
            "strategy_type": "momentum",
            "parameters": {}
        }
    )
    assert create_response.status_code == 201
    strategy_id = create_response.json()["id"]

    # 尝试strategy_id路径参数注入
    injection_payload: str = "invalid_uuid' OR '1'='1'--"
    response = client.delete(
        f"/api/v1/strategies/{injection_payload}",
        headers={"Authorization": f"Bearer {access_token}"}
    )
    # 验证系统返回404或500，而不是执行删除
    assert response.status_code == 404
    assert response.json()["detail"] == "策略不存在"

    # 尝试删除实际存在的策略，验证其行为
    response_valid_delete = client.delete(
        f"/api/v1/strategies/{strategy_id}",
        headers={"Authorization": f"Bearer {access_token}"}
    )
    assert response_valid_delete.status_code == 200
    assert response_valid_delete.json()["message"] == "策略已成功标记为非活跃"

def test_sql_injection_run_backtest() -> None:
    """测试运行策略回测接口是否防护SQL注入"""
    # 创建测试用户token
    access_token_expires: timedelta = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token: str = create_access_token(
        data={"sub": "test_user"},
        expires_delta=access_token_expires
    )

    # 尝试strategy_id路径参数注入
    injection_payload_id: str = "invalid_uuid' OR '1'='1'--"
    response_id = client.post(
        f"/api/v1/strategies/{injection_payload_id}/backtest",
        headers={"Authorization": f"Bearer {access_token}"},
        json={
            "stock_codes": ["000001.SZ"],
            "start_date": "2020-01-01",
            "end_date": "2021-01-01",
            "initial_capital": 100000
        }
    )
    # 验证系统返回404
    assert response_id.status_code == 404
    assert response_id.json()["detail"] == "策略不存在"

    # 尝试stock_codes参数注入 (假设stock_codes会被用于查询)
    # 由于stock_codes是列表，注入通常发生在列表元素内部
    injection_payload_stock: List[str] = ["000001.SZ", "'; DROP TABLE users;--"]
    # 需要一个有效的strategy_id来测试这个场景
    create_response = client.post(
        "/api/v1/strategies/",
        headers={"Authorization": f"Bearer {access_token}"},
        json={
            "name": "Backtest Strategy",
            "description": "Description",
            "strategy_type": "momentum",
            "parameters": {}
        }
    )
    assert create_response.status_code == 201
    valid_strategy_id = create_response.json()["id"]

    response_stock = client.post(
        f"/api/v1/strategies/{valid_strategy_id}/backtest",
        headers={"Authorization": f"Bearer {access_token}"},
        json={
            "stock_codes": injection_payload_stock,
            "start_date": "2020-01-01",
            "end_date": "2021-01-01",
            "initial_capital": 100000
        }
    )
    # 验证系统返回500或处理错误，而不是执行恶意SQL
    # 预期是500，因为stock_service.get_latest_quotes_batch可能会因为无效的股票代码而失败
    assert response_stock.status_code == 500
    assert "回测失败" in response_stock.json()["detail"]

def test_sql_injection_get_selection_dates() -> None:
    """测试获取选股日期列表接口是否防护SQL注入"""
    # 创建测试用户token
    access_token_expires: timedelta = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token: str = create_access_token(
        data={"sub": "test_user"},
        expires_delta=access_token_expires
    )

    # 尝试strategy_id查询参数注入
    injection_payload: str = "invalid_uuid' OR '1'='1'--"
    response = client.get(
        f"/api/v1/strategies/selections/dates?strategy_id={injection_payload}",
        headers={"Authorization": f"Bearer {access_token}"}
    )
    # 验证系统返回空列表或错误，而不是敏感数据
    assert response.status_code == 200
    assert response.json() == []

def test_sql_injection_get_stock_selections() -> None:
    """测试获取选股结果接口是否防护SQL注入"""
    # 创建测试用户token
    access_token_expires: timedelta = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token: str = create_access_token(
        data={"sub": "test_user"},
        expires_delta=access_token_expires
    )

    # 尝试strategy_id查询参数注入
    injection_payload: str = "invalid_uuid' OR '1'='1'--"
    response = client.get(
        f"/api/v1/strategies/selections?strategy_id={injection_payload}",
        headers={"Authorization": f"Bearer {access_token}"}
    )
    # 验证系统返回空列表或错误，而不是敏感数据
    assert response.status_code == 200
    assert response.json() == []

def test_sql_injection_update_portfolio() -> None:
    """测试更新投资组合接口是否防护SQL注入"""
    # 创建测试用户token
    access_token_expires: timedelta = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token: str = create_access_token(
        data={"sub": "test_user"},
        expires_delta=access_token_expires
    )

    # 首先创建一个投资组合用于更新
    create_response = client.post(
        "/api/v1/portfolios/",
        headers={"Authorization": f"Bearer {access_token}"},
        json={
            "name": "Portfolio to Update",
            "description": "Original description"
        }
    )
    assert create_response.status_code == 201
    portfolio_id = create_response.json()["id"]

    # 尝试name参数注入
    injection_payload_name: str = "Updated Portfolio' OR '1'='1'--"
    response = client.put(
        f"/api/v1/portfolios/{portfolio_id}",
        headers={"Authorization": f"Bearer {access_token}"},
        json={
            "name": injection_payload_name,
            "description": "Updated description"
        }
    )
    # 验证系统返回成功，但注入内容被转义或过滤
    assert response.status_code == 200
    assert response.json()["name"] == injection_payload_name

def test_sql_injection_get_backtest_result_detail() -> None:
    """测试获取回测结果详情接口是否防护SQL注入"""
    # 创建测试用户token
    access_token_expires: timedelta = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token: str = create_access_token(
        data={"sub": "test_user"},
        expires_delta=access_token_expires
    )

    # 尝试backtest_id路径参数注入
    injection_payload: str = "1' OR '1'='1'--"
    response = client.get(
        f"/api/v1/backtests/{injection_payload}",
        headers={"Authorization": f"Bearer {access_token}"}
    )
    # 验证系统返回422 Unprocessable Entity，因为backtest_id期望整数
    assert response.status_code == 422
    assert "value is not a valid integer" in response.json()["detail"][0]["msg"]