Open raw ↗"""Tier 5: Adversarial Hardening, Security & Concurrency Stress Tests.
Probes system resilience against attack vectors and stress conditions:
- Path traversal exploits (/static/../..)
- Brute-force password spraying and 5-failure lockout enforcement
- SQL injection payloads across authentication, reports, and query parameters
- Cross-Site Scripting (XSS) and Content Security Policy (CSP) headers
- Session hijacking and cookie tampering
- Concurrent multi-threaded report submission (SQLite WAL concurrency stress)
"""
from concurrent.futures import ThreadPoolExecutor
import pytest
import requests
@pytest.mark.tier5
class TestPathTraversalSecurity:
"""Directory containment and path traversal defenses."""
def test_adversarial_path_traversal_dot_dot(self, api_client):
# Attempt to access app.py through static handler
resp = api_client.get("/static/../../app.py")
assert resp.status_code in (400, 403, 404)
assert "class HADRequestHandler" not in resp.text
def test_adversarial_path_traversal_encoded(self, api_client):
# URL encoded path traversal
resp = api_client.get("/static/..%2f..%2fconfig_manager.py")
assert resp.status_code in (400, 403, 404)
assert "DEFAULT_CONFIG" not in resp.text
def test_adversarial_path_traversal_database(self, api_client):
# Attempt to access SQLite database directly via static route
resp = api_client.get("/static/../data/had.db")
assert resp.status_code in (400, 403, 404)
assert "SQLite format" not in resp.text
def test_adversarial_static_nonexistent_file(self, api_client):
resp = api_client.get("/static/nonexistent_asset_xyz.js")
assert resp.status_code == 404
@pytest.mark.tier5
class TestBruteForceAndLockoutHardening:
"""Brute-force password attack resilience and 5-failure account lockout."""
def test_adversarial_brute_force_5_lockout(self, isolated_server):
from conftest import HADClient
client = HADClient(isolated_server.base_url)
# 5 consecutive failed login attempts on patient account
for attempt in range(1, 6):
r = client.login("patient.durand", f"wrong_guess_{attempt}")
assert r.status_code == 401
# 6th attempt with the CORRECT password ('demo123') must be locked out
locked_resp = client.login("patient.durand", "demo123")
assert locked_resp.status_code == 401
data = locked_resp.json()
assert "locked" in data.get("error", "").lower() or "invalid" in data.get("error", "").lower()
# Verify audit log recorded the failed login attempts
audit_rows = isolated_server.query_db(
"SELECT action, details FROM audit_log WHERE action = 'login_failed' ORDER BY id DESC LIMIT 5"
)
assert len(audit_rows) >= 5
assert all("patient.durand" in r["details"] for r in audit_rows)
@pytest.mark.tier5
class TestSQLInjectionResilience:
"""SQL Injection attack resilience across endpoints."""
def test_adversarial_sql_injection_login_or(self, api_client):
# Classic tautology bypass payload
resp = api_client.post("/api/login", json={"username": "' OR '1'='1", "password": "any"})
assert resp.status_code == 401
def test_adversarial_sql_injection_login_comment(self, api_client):
# SQL comment injection
resp = api_client.post("/api/login", json={"username": "admin'--", "password": "wrong"})
assert resp.status_code == 401
def test_adversarial_sql_injection_in_report_notes(self, patient_client, session_server):
# Malicious drop table payload inside notes text
sql_payload = "Normal note'); DROP TABLE users; --"
resp = patient_client.post("/api/reports", json={
"patient_id": 1,
"symptom_id": "nausea",
"symptom_category": "gastrointestinal",
"severity_score": 1,
"notes": sql_payload,
})
assert resp.status_code == 201
# Assert users table was not dropped
user_rows = session_server.query_db("SELECT count(*) as cnt FROM users")
assert user_rows[0]["cnt"] >= 10
def test_adversarial_sql_injection_in_patient_id_query(self, oncologist_client):
# Malicious injection in query parameter
resp = oncologist_client.get("/api/reports?patient_id=1%20OR%201=1")
# Should either cleanly return 400 or treat as invalid/empty, never execute arbitrary SQL
assert resp.status_code in (200, 400)
@pytest.mark.tier5
class TestXSSAndSecurityHeaders:
"""Cross-Site Scripting (XSS) payload handling and security headers."""
def test_adversarial_xss_payload_in_report_notes(self, patient_client):
xss_string = "<script>alert('XSS-ATTACK')</script><img src=x onerror=alert(1)>"
resp = patient_client.post("/api/reports", json={
"patient_id": 1,
"symptom_id": "fatigue",
"symptom_category": "constitutional",
"severity_score": 2,
"notes": xss_string,
})
assert resp.status_code == 201
report_id = resp.json()["report_id"]
# Retrieve report and verify payload is returned as raw text data without executing
list_resp = patient_client.get("/api/reports?patient_id=1")
assert list_resp.status_code == 200
reports = list_resp.json().get("reports", [])
created_report = next((r for r in reports if r["id"] == report_id), None)
assert created_report is not None
assert created_report["notes"] == xss_string
def test_adversarial_security_headers_enforced(self, api_client):
resp = api_client.get("/api/whoami")
headers = resp.headers
# Check required security headers
csp = headers.get("Content-Security-Policy", "")
assert "default-src 'self'" in csp
assert headers.get("X-Content-Type-Options") == "nosniff"
assert headers.get("X-Frame-Options") == "DENY"
assert headers.get("Referrer-Policy") == "strict-origin-when-cross-origin"
@pytest.mark.tier5
class TestSessionSecurity:
"""Session token hijacking and cookie manipulation defenses."""
def test_adversarial_fake_uuid_session_rejected(self, api_client):
fake_uuid = "00000000-0000-0000-0000-000000000000"
resp = api_client.get("/api/reports?patient_id=1", headers={"Cookie": f"HAD_SESSION={fake_uuid}"})
assert resp.status_code == 401
def test_adversarial_malformed_session_cookie(self, api_client):
resp = api_client.get("/api/reports?patient_id=1", headers={"Cookie": "HAD_SESSION=../../etc/passwd"})
assert resp.status_code == 401
@pytest.mark.tier5
class TestConcurrencyStress:
"""Multi-threaded rapid concurrent request handling (SQLite WAL stress)."""
def test_adversarial_rapid_concurrent_reports(self, isolated_server):
"""Spawns 20 threads submitting reports concurrently to stress test SQLite."""
from conftest import HADClient
base_url = isolated_server.base_url
# Helper worker function
def submit_report(worker_id: int):
client = HADClient(base_url)
login_r = client.login("patient.durand", "demo123")
if login_r.status_code != 200:
return False, f"Login failed for worker {worker_id}: {login_r.text}"
report_r = client.post("/api/reports", json={
"patient_id": 1,
"symptom_id": "nausea",
"symptom_category": "gastrointestinal",
"severity_score": (worker_id % 3) + 1,
"notes": f"Concurrent stress worker report #{worker_id}",
})
if report_r.status_code != 201:
return False, f"Report failed for worker {worker_id}: {report_r.text}"
return True, report_r.json().get("report_id")
# Execute 20 concurrent submissions
with ThreadPoolExecutor(max_workers=10) as executor:
futures = [executor.submit(submit_report, i) for i in range(20)]
results = [f.result() for f in futures]
# Verify all 20 succeeded
successes = [r for r in results if r[0]]
failures = [r for r in results if not r[0]]
assert len(failures) == 0, f"Concurrent failures: {failures}"
assert len(successes) == 20
# Verify all 20 report IDs are distinct
report_ids = [r[1] for r in successes]
assert len(set(report_ids)) == 20
# Direct DB count check
stress_rows = isolated_server.query_db(
"SELECT count(*) as cnt FROM toxicity_reports WHERE notes LIKE 'Concurrent stress worker report%'"
)
assert stress_rows[0]["cnt"] == 20