72 lines
3.0 KiB
Python
72 lines
3.0 KiB
Python
# requires: fastapi, httpx
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from secure_app import (DB, app, check_fetch_url, find_orders_safe, find_orders_unsafe, render_answer_safe,
|
|
render_answer_unsafe)
|
|
|
|
client = TestClient(app, base_url="https://testserver")
|
|
|
|
|
|
def test_sql_injection_unsafe_leaks_other_tenant():
|
|
rows = find_orders_unsafe(DB, "t1", "x' OR 1=1 --")
|
|
assert ("B001", "t2") in rows # 证明漏洞真实存在
|
|
|
|
|
|
def test_sql_injection_safe():
|
|
assert find_orders_safe(DB, "t1", "x' OR 1=1 --") == []
|
|
assert find_orders_safe(DB, "t1", "A00") == [("A001", "t1"), ("A002", "t1")]
|
|
|
|
|
|
def test_xss_escaped():
|
|
evil = '<img src=x onerror="alert(document.cookie)">'
|
|
assert "<img" in render_answer_unsafe(evil)
|
|
assert "<img" not in render_answer_safe(evil) and "<img" in render_answer_safe(evil)
|
|
r = client.get("/answer", params={"q": "<script>alert(1)</script>"})
|
|
assert "<script>" not in r.text and "script-src 'self'" in r.headers["content-security-policy"]
|
|
|
|
|
|
def test_csrf():
|
|
c = TestClient(app, base_url="https://testserver")
|
|
c.post("/login")
|
|
# 模拟跨站伪造：浏览器会自动带上 Cookie，但攻击页面读不到 csrf_token，无法设置请求头
|
|
assert c.post("/transfer", data={"amount": 100}).status_code == 403
|
|
token = c.cookies.get("csrf_token")
|
|
r = c.post("/transfer", data={"amount": 100}, headers={"X-CSRF-Token": token})
|
|
assert r.status_code == 200
|
|
|
|
|
|
def test_cors_allowlist():
|
|
ok = client.get("/orders", headers={"Origin": "https://app.example.com"})
|
|
assert ok.headers.get("access-control-allow-origin") == "https://app.example.com"
|
|
evil = client.get("/orders", headers={"Origin": "https://evil.example"})
|
|
assert "access-control-allow-origin" not in evil.headers
|
|
pre = client.options("/orders", headers={"Origin": "https://evil.example", "Access-Control-Request-Method": "GET"})
|
|
assert pre.status_code == 400
|
|
|
|
|
|
def fake_resolve(mapping):
|
|
return lambda host, port: [(None, None, None, "", (mapping[host], port))]
|
|
|
|
|
|
@pytest.mark.parametrize("url,ip", [
|
|
("http://169.254.169.254/latest/meta-data/", "169.254.169.254"), # 云元数据
|
|
("http://localhost:6379/", "127.0.0.1"),
|
|
("http://intranet.corp/", "10.0.0.8"),
|
|
("http://rebind.example/", "192.168.1.10"), # 公网域名解析到内网
|
|
])
|
|
def test_ssrf_blocks_internal(url, ip):
|
|
host = url.split("/")[2].split(":")[0]
|
|
with pytest.raises(ValueError):
|
|
check_fetch_url(url, resolve=fake_resolve({host: ip}))
|
|
|
|
|
|
@pytest.mark.parametrize("url", ["file:///etc/passwd", "gopher://x/", "http://user:pw@example.com/"])
|
|
def test_ssrf_blocks_schemes_and_credentials(url):
|
|
with pytest.raises(ValueError):
|
|
check_fetch_url(url, resolve=fake_resolve({"example.com": "93.184.215.14", "x": "93.184.215.14"}))
|
|
|
|
|
|
def test_ssrf_allows_public():
|
|
assert check_fetch_url("https://example.com/a", resolve=fake_resolve({"example.com": "93.184.215.14"}))
|