File size: 8,211 Bytes
eb0499d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 | # ruff: noqa: E402
import sys
from pathlib import Path
import pytest
from httpx import ASGITransport, AsyncClient
root_dir = Path(__file__).resolve().parents[3]
if str(root_dir) not in sys.path:
sys.path.insert(0, str(root_dir))
from app.infrastructure.security.perimeter_waf import (
PerimeterWAFGuard,
WAFRuleCategory,
)
from app.main import app
@pytest.mark.asyncio
async def test_api_routes_strict_security_headers():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
resp = await client.get("/health")
assert resp.status_code == 200
csp = resp.headers.get("Content-Security-Policy", "")
assert "default-src 'none'" in csp
assert "script-src 'none'" in csp
assert resp.headers.get("X-Frame-Options") == "DENY"
assert resp.headers.get("X-Content-Type-Options") == "nosniff"
@pytest.mark.asyncio
async def test_comprehensive_http_security_headers():
"""Verify all 7 standard defensive HTTP response headers are present on API responses."""
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
resp = await client.get("/health")
assert resp.status_code == 200
headers = resp.headers
assert "Strict-Transport-Security" in headers
assert "max-age=31536000" in headers["Strict-Transport-Security"]
assert "includeSubDomains" in headers["Strict-Transport-Security"]
assert headers.get("X-Content-Type-Options") == "nosniff"
assert headers.get("X-Frame-Options") == "DENY"
assert headers.get("Referrer-Policy") == "no-referrer"
assert "Permissions-Policy" in headers
assert "geolocation=()" in headers["Permissions-Policy"]
assert headers.get("X-XSS-Protection") == "1; mode=block"
@pytest.mark.asyncio
async def test_docs_and_scalar_permissive_csp_for_swagger_cdn():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
# Test /scalar endpoint
resp_scalar = await client.get("/scalar")
assert resp_scalar.status_code == 200
csp_scalar = resp_scalar.headers.get("Content-Security-Policy", "")
assert "https://cdn.jsdelivr.net" in csp_scalar
assert "script-src" in csp_scalar
assert "'unsafe-inline'" in csp_scalar
# Test /docs endpoint
resp_docs = await client.get("/docs")
assert resp_docs.status_code == 200
csp_docs = resp_docs.headers.get("Content-Security-Policy", "")
assert "https://cdn.jsdelivr.net" in csp_docs
assert "'unsafe-inline'" in csp_docs
# Test /redoc endpoint and worker-src / blob: CSP support for ReDoc Web Workers
resp_redoc = await client.get("/redoc")
assert resp_redoc.status_code == 200
assert "redoc-container" in resp_redoc.text
csp_redoc = resp_redoc.headers.get("Content-Security-Policy", "")
assert "worker-src" in csp_redoc
assert "blob:" in csp_redoc
# ── Perimeter WAF Header Inspection & Memory Pruning Tests ────────────────────
def test_perimeter_waf_header_sqli_inspection():
"""Verify WAF inspects request headers and blocks header-based SQL injection."""
waf = PerimeterWAFGuard()
malicious_headers = {"X-Filter": "UNION SELECT * FROM accounts;--"}
result = waf.inspect_request(
client_ip="192.168.1.50",
path="/api/v1/predict",
headers=malicious_headers,
body="{}",
)
assert result.allowed is False
assert result.rule_triggered == WAFRuleCategory.SQLI_INJECTION
def test_perimeter_waf_header_xss_inspection():
"""Verify WAF inspects request headers and blocks header-based XSS attacks."""
waf = PerimeterWAFGuard()
malicious_headers = {"User-Agent": "Mozilla/5.0 <script>alert(document.cookie)</script>"}
result = waf.inspect_request(
client_ip="192.168.1.51",
path="/api/v1/alerts",
headers=malicious_headers,
body="{}",
)
assert result.allowed is False
assert result.rule_triggered == WAFRuleCategory.XSS_ATTACK
def test_perimeter_waf_header_null_byte_inspection():
"""Verify WAF inspects request headers and blocks null bytes in header values."""
waf = PerimeterWAFGuard()
malicious_headers = {"X-Custom-Tenant": "bank_a\x00_bypass"}
result = waf.inspect_request(
client_ip="192.168.1.52",
path="/api/v1/cases",
headers=malicious_headers,
body="{}",
)
assert result.allowed is False
assert result.rule_triggered == WAFRuleCategory.NULL_BYTE_DETECTED
def test_perimeter_waf_memory_pruning():
"""Verify WAF prunes expired IP tracking entries when size exceeds max_tracked_ips."""
import time
waf = PerimeterWAFGuard(max_tracked_ips=10, lockout_duration_seconds=5)
now = time.time()
expired_time = now - 10.0
# Fill internal tracking with 15 expired IPs
for i in range(15):
waf._auth_failures[f"192.168.10.{i}"] = [expired_time]
assert len(waf._auth_failures) == 15
# Trigger record_failed_auth_attempt for a new active IP
waf.record_failed_auth_attempt("10.0.0.99")
# Expired entries should be pruned, retaining only active IP
assert len(waf._auth_failures) == 1
assert "10.0.0.99" in waf._auth_failures
def test_perimeter_waf_lockout_reset():
"""Verify WAF resets client lockout upon explicit reset_client_lockout invocation."""
waf = PerimeterWAFGuard(max_auth_failures=3)
test_ip = "192.168.20.5"
for _ in range(3):
waf.record_failed_auth_attempt(test_ip)
assert waf.is_client_locked_out(test_ip) is True
# Reset lockout
waf.reset_client_lockout(test_ip)
assert waf.is_client_locked_out(test_ip) is False
def test_perimeter_waf_concurrency_thread_safety():
"""Verify concurrent failed auth recordings across multiple threads maintain exact count."""
import concurrent.futures
waf = PerimeterWAFGuard(max_auth_failures=100)
test_ip = "192.168.30.10"
def record():
waf.record_failed_auth_attempt(test_ip)
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
futures = [executor.submit(record) for _ in range(50)]
concurrent.futures.wait(futures)
assert len(waf._auth_failures[test_ip]) == 50
# ── Cloudflare WAF Setup Script Tests ──────────────────────────────────────────
def test_cloudflare_waf_setup_script_api_calls(monkeypatch):
"""Verify Cloudflare WAF setup script configures settings, rules, and rate limits."""
import importlib
cf_module = importlib.import_module("scripts.setup_cloudflare_waf")
configure_security_settings = cf_module.configure_security_settings
configure_waf_rules = cf_module.configure_waf_rules
configure_rate_limiting_rules = cf_module.configure_rate_limiting_rules
recorded_calls: list[dict] = []
def mock_cf_request(endpoint: str, method: str = "GET", token: str = "", data: dict | None = None):
recorded_calls.append({"endpoint": endpoint, "method": method, "data": data})
return {"success": True, "result": {}}
monkeypatch.setattr(cf_module, "cf_request", mock_cf_request)
configure_security_settings("zone_123", "token_abc")
configure_waf_rules("zone_123", "token_abc")
configure_rate_limiting_rules("zone_123", "token_abc")
# Verify settings calls were made (8 settings)
setting_calls = [c for c in recorded_calls if "settings" in c["endpoint"]]
assert len(setting_calls) == 8
# Verify WAF ruleset call was made
waf_calls = [c for c in recorded_calls if "rulesets" in c["endpoint"]]
assert len(waf_calls) == 1
assert waf_calls[0]["method"] == "PUT"
assert len(waf_calls[0]["data"]["rules"]) >= 2
# Verify rate limits call was made
rate_calls = [c for c in recorded_calls if "rate_limits" in c["endpoint"]]
assert len(rate_calls) == 1
assert rate_calls[0]["method"] == "POST"
assert rate_calls[0]["data"]["threshold"] == 100
|