python / test_analysis.py
python / test_analysis.py
import asyncio
import gzip
import json
import os
import sys
from dataclasses import asdict
from pathlib import Path
import httpx
import pytest
from jsonschema import Draft202012Validator
from mcp import Client, StdioServerParameters
import server
def response(headers=None, html="", url="https://example.com/"):
return httpx.Response(200, headers=headers or {}, text=html, request=httpx.Request("GET", url))
def report(headers=None, html="", url="https://example.com/"):
return server.analyze_response(response(headers, html, url), url, [url])
def codes(findings):
return {f.finding for f in findings}
def test_missing_csp_is_conditional():
_, findings = server._parse_csp(None)
f = findings[0]
assert f.finding == "csp_missing"
assert f.standalone_exploitability == "none"
assert "XSS" in f.required_conditions[0]
assert f.conditional_impact["level"] == "high"
assert "does not demonstrate XSS" in f.inference
@pytest.mark.parametrize("policy,expected", [
("script-src 'self' 'unsafe-inline'", True),
("script-src 'unsafe-inline' 'nonce-AbCd'", False),
("script-src 'unsafe-inline' 'sha256-AbCd'", False),
("script-src 'self'; script-src-elem 'unsafe-inline'", True),
("script-src 'unsafe-inline'; script-src-elem 'none'; script-src-attr 'none'", False),
("script-src 'self'; script-src 'unsafe-inline'", False),
])
def test_csp_inline_semantics(policy, expected):
_, fs = server._parse_csp(policy)
assert any(f.finding.startswith("csp_unsafe_inline") for f in fs) == expected
def test_csp_eval_wildcard_and_missing_script():
_, fs = server._parse_csp("script-src * 'unsafe-eval'")
assert {"csp_unsafe_eval", "csp_script_wildcard"} <= codes(fs)
_, fs = server._parse_csp("frame-ancestors 'none'")
assert "csp_script_restriction_missing" in codes(fs)
def test_meta_and_report_only_not_equivalent_to_header():
r = report({"content-type": "text/html", "content-security-policy-report-only": "script-src 'none'"},
'<meta http-equiv="Content-Security-Policy" content="script-src \'self\'">')
assert r["csp"]["present"] is False
f = next(f for f in r["findings"] if f["finding"] == "csp_missing")
assert f["evidence"]["meta_policies_observed"]
assert "meta CSP was observed" in f["inference"]
def test_cookie_secret_redacted_and_roles_not_facts():
cookies, fs = server._analyze_cookies(response([
("set-cookie", "colorme_PHPSESSID=SECRET123; Path=/"),
("set-cookie", "theme=OTHERSECRET; Path=/")]))
assert len(cookies) == 2
assert "SECRET" not in json.dumps([asdict(f) for f in fs])
assert cookies[0]["asset_role"] == "possibly_sensitive_cookie"
assert cookies[1]["asset_role"] == "unknown"
httponly = next(f for f in fs if f.finding == "cookie_missing_httponly")
assert len(httponly.required_conditions) == 2
assert "additionally requires" in httponly.conditional_impact["description"]
def test_cookie_secure_with_hsts_stays_conditional():
_, fs = server._analyze_cookies(response({"set-cookie": "sid=x; HttpOnly; SameSite=Lax", "strict-transport-security": "max-age=31536000"}))
f = next(f for f in fs if f.finding == "cookie_missing_secure")
assert f.severity == "low"
assert "HTTPS traffic stays encrypted" in f.inference
assert any("HSTS" in c for c in f.required_conditions)
@pytest.mark.parametrize("same,code", [
("", "cookie_samesite_missing"), ("; SameSite=weird", "cookie_samesite_invalid"),
("; SameSite=None", "cookie_samesite_none_without_secure")])
def test_samesite_defaults_and_none(same, code):
cookies, fs = server._analyze_cookies(response({"set-cookie": "sid=x; HttpOnly" + same}))
assert code in codes(fs)
if "none" not in code:
assert cookies[0]["effective_modern_default"].startswith("Lax")
assert "does not prove CSRF" in next(f.inference for f in fs if f.finding == code)
def test_cookie_strong_flags_case_insensitive_and_expires():
cookies, fs = server._analyze_cookies(response({"set-cookie": "sid=x; Expires=Wed, 21 Oct 2030 07:28:00 GMT; SECURE; HTTPONLY; samesite=lAx"}))
assert len(cookies) == 1
assert fs == []
def test_samesite_none_secure_is_intent_review():
_, fs = server._analyze_cookies(response({"set-cookie": "sid=x; Secure; HttpOnly; SameSite=None"}))
f = fs[0]
assert f.finding == "cookie_samesite_none"
assert f.severity == "info"
@pytest.mark.parametrize("src,external", [
("/app.js", False), ("https://example.com:443/app.js", False),
("http://example.com/app.js", True), ("https://example.com:444/app.js", True),
("//cdn.example.net/app.js", True)])
def test_sri_full_origin_comparison(src, external):
html, fs = server._analyze_html(response({"content-type": "text/html"}, f'<script src="{src}"></script>'))
assert ("sri_missing" in codes(fs)) == external
@pytest.mark.parametrize("version,expected", [("12", "mutable_or_unpinned"), ("12.1", "mutable_or_unpinned"),
("latest", "mutable_or_unpinned"), ("12.1.2", "exact_version_candidate")])
def test_sri_version_pin(version, expected):
_, fs = server._analyze_html(response({"content-type": "text/html"},
f'<script src="https://cdn.jsdelivr.net/npm/swiper@{version}/swiper.js"></script>'))
assert ("sri_mutable_url" in codes(fs)) == (expected == "mutable_or_unpinned")
assert fs[0].evidence["version_pin"] == expected
def test_sri_attribute_not_hash_verification():
html, fs = server._analyze_html(response({"content-type": "text/html"},
'<script src="https://cdn.example.org/a.js" integrity="sha384-AAAA" crossorigin="anonymous"></script>'))
assert "sri_missing" not in codes(fs)
assert html["sri_observations"][0]["hash_verified"] is False
_, fs = server._analyze_html(response({"content-type": "text/html"},
'<script src="https://cdn.example.org/a.js" integrity="garbage"></script>'))
assert "sri_invalid_syntax" in codes(fs)
@pytest.mark.parametrize("headers,flagged", [({}, True), ({"content-security-policy": "default-src 'self'"}, True),
({"x-frame-options": "SAMEORIGIN"}, False), ({"x-frame-options": "ALLOW-FROM https://x.test"}, True),
({"content-security-policy": "frame-ancestors 'none'"}, False), ({"content-security-policy": "frame-ancestors *"}, True)])
def test_framing_without_csp(headers, flagged):
r = report(headers)
assert any(f["finding"] == "framing_restriction_missing_or_weak" for f in r["findings"]) == flagged
def test_headers_not_assumed_effective():
headers, fs = server._analyze_headers(response({"strict-transport-security": "max-age=0", "x-content-type-options": "wrong"}))
assert "strict_transport_security_missing_or_weak" in codes(fs)
assert "x_content_type_options_missing_or_weak" in codes(fs)
assert headers["strict-transport-security"]["effectiveness"] == "not_verified"
isolation = [f for f in fs if f.category == "isolation"]
assert len(isolation) == 3 and all(f.severity == "info" for f in isolation)
def test_schema_and_all_finding_invariants():
schema = json.loads(Path(__file__).with_name("finding.schema.json").read_text())
validator = Draft202012Validator(schema)
r = report({"content-type": "text/html", "set-cookie": "PHPSESSID=SECRET"},
'<script src="http://cdn.jsdelivr.net/npm/swiper@12/a.js"></script>')
assert "SECRET" not in json.dumps(r)
for finding in r["findings"]:
validator.validate(finding)
assert finding["required_conditions"]
assert finding["standalone_exploitability"] == "none"
assert "heuristic" in r["security_score_semantics"]
def test_ssrf_nonpublic_and_credentials(monkeypatch):
monkeypatch.setattr(server.socket, "getaddrinfo", lambda *a: [(2, 1, 6, "", ("127.0.0.1", 80))])
with pytest.raises(server.ToolError):
server._validate_target("http://localhost/")
for url in ["file:///a", "https://user:pass@example.com"]:
with pytest.raises(server.ToolError):
server._validate_target(url)
def test_fetch_redirect_and_stream_limit(monkeypatch):
seen = []
def validate(url):
seen.append(url)
if "127.0.0.1" in url:
raise server.ToolError("blocked")
return url
monkeypatch.setattr(server, "_validate_target", validate)
real_client = httpx.Client
def make_client(handler):
return lambda **kw: real_client(transport=httpx.MockTransport(handler), **kw)
monkeypatch.setattr(server.httpx, "Client", make_client(lambda req: httpx.Response(302, headers={"location": "http://127.0.0.1/"})))
with pytest.raises(server.ToolError, match="blocked"):
server._fetch("https://example.com/")
assert seen[-1] == "http://127.0.0.1/"
monkeypatch.setattr(server, "MAX_RESPONSE_BYTES", 5)
monkeypatch.setattr(server.httpx, "Client", make_client(lambda req: httpx.Response(200, content=b"123456")))
with pytest.raises(server.ToolError, match="limit"):
server._fetch("https://example.com/")
def test_scan_url_uses_same_schema(monkeypatch, tmp_path):
monkeypatch.setenv("WEB_SECURITY_MCP_REPORT_DIR", str(tmp_path))
monkeypatch.setattr(server, "_fetch", lambda url: (response(), [url]))
result = server.scan_url("https://example.com/")
assert result["schema_version"] == "2.0"
assert "review_markdown" not in result
assert server.scan_url("https://example.com/", include_review_markdown=True)["review_markdown"]
@pytest.mark.parametrize("encoding", ["gzip", "deflate", "identity"])
def test_fetch_compressed_response_decoded_once(monkeypatch, encoding):
import zlib
payload = b'<html><title>Compressed page</title></html>'
raw = gzip.compress(payload) if encoding == "gzip" else zlib.compress(payload) if encoding == "deflate" else payload
real_client = httpx.Client
def handler(request):
return httpx.Response(200, headers={"content-type": "text/html; charset=utf-8",
"content-encoding": encoding, "content-length": str(len(raw))}, stream=httpx.ByteStream(raw))
monkeypatch.setattr(server, "_validate_target", lambda url: url)
monkeypatch.setattr(server.httpx, "Client", lambda **kw: real_client(transport=httpx.MockTransport(handler), **kw))
result, chain = server._fetch("https://example.com/")
assert result.content == payload
assert result.headers["content-encoding"] == encoding
assert result.headers["content-length"] == str(len(raw))
assert server.analyze_response(result, chain[0], chain)["html"]["title"] == "Compressed page"
def test_compressed_expansion_obeys_limit(monkeypatch):
raw = gzip.compress(b"a" * 10000)
real_client = httpx.Client
monkeypatch.setattr(server, "_validate_target", lambda url: url)
monkeypatch.setattr(server, "MAX_RESPONSE_BYTES", 100)
monkeypatch.setattr(server.httpx, "Client", lambda **kw: real_client(
transport=httpx.MockTransport(lambda req: httpx.Response(200,
headers={"content-encoding": "gzip"}, stream=httpx.ByteStream(raw))), **kw))
with pytest.raises(server.ToolError, match="limit"):
server._fetch("https://example.com/")
def test_genuinely_invalid_gzip_is_not_accepted(monkeypatch):
real_client = httpx.Client
monkeypatch.setattr(server, "_validate_target", lambda url: url)
monkeypatch.setattr(server.httpx, "Client", lambda **kw: real_client(
transport=httpx.MockTransport(lambda req: httpx.Response(200,
headers={"content-encoding": "gzip"}, stream=httpx.ByteStream(b"not gzip"))), **kw))
with pytest.raises(server.ToolError, match="HTTP request failed"):
server._fetch("https://example.com/")
def test_stdio_modern_and_legacy():
# Launch the actual entry point; no requests to public sites.
async def run(mode):
params = StdioServerParameters(command=sys.executable,
args=[str(Path(__file__).with_name("server.py"))], env=dict(os.environ))
async with Client(params, mode=mode, read_timeout_seconds=15) as client:
listed = await client.list_tools()
assert {t.name for t in listed.tools} == {"scan_url", "explain_capabilities"}
result = await client.call_tool("explain_capabilities", {})
assert not result.is_error
assert result.structured_content["schema_version"] == "2.0"
bad = await client.call_tool("scan_url", {"url": "file:///forbidden"})
assert bad.is_error
asyncio.run(run("auto"))
asyncio.run(run("legacy"))
import asyncio
import gzip
import json
import os
import sys
from dataclasses import asdict
from pathlib import Path
import httpx
import pytest
from jsonschema import Draft202012Validator
from mcp import Client, StdioServerParameters
import server
def response(headers=None, html="", url="https://example.com/"):
return httpx.Response(200, headers=headers or {}, text=html, request=httpx.Request("GET", url))
def report(headers=None, html="", url="https://example.com/"):
return server.analyze_response(response(headers, html, url), url, [url])
def codes(findings):
return {f.finding for f in findings}
def test_missing_csp_is_conditional():
_, findings = server._parse_csp(None)
f = findings[0]
assert f.finding == "csp_missing"
assert f.standalone_exploitability == "none"
assert "XSS" in f.required_conditions[0]
assert f.conditional_impact["level"] == "high"
assert "does not demonstrate XSS" in f.inference
@pytest.mark.parametrize("policy,expected", [
("script-src 'self' 'unsafe-inline'", True),
("script-src 'unsafe-inline' 'nonce-AbCd'", False),
("script-src 'unsafe-inline' 'sha256-AbCd'", False),
("script-src 'self'; script-src-elem 'unsafe-inline'", True),
("script-src 'unsafe-inline'; script-src-elem 'none'; script-src-attr 'none'", False),
("script-src 'self'; script-src 'unsafe-inline'", False),
])
def test_csp_inline_semantics(policy, expected):
_, fs = server._parse_csp(policy)
assert any(f.finding.startswith("csp_unsafe_inline") for f in fs) == expected
def test_csp_eval_wildcard_and_missing_script():
_, fs = server._parse_csp("script-src * 'unsafe-eval'")
assert {"csp_unsafe_eval", "csp_script_wildcard"} <= codes(fs)
_, fs = server._parse_csp("frame-ancestors 'none'")
assert "csp_script_restriction_missing" in codes(fs)
def test_meta_and_report_only_not_equivalent_to_header():
r = report({"content-type": "text/html", "content-security-policy-report-only": "script-src 'none'"},
'<meta http-equiv="Content-Security-Policy" content="script-src \'self\'">')
assert r["csp"]["present"] is False
f = next(f for f in r["findings"] if f["finding"] == "csp_missing")
assert f["evidence"]["meta_policies_observed"]
assert "meta CSP was observed" in f["inference"]
def test_cookie_secret_redacted_and_roles_not_facts():
cookies, fs = server._analyze_cookies(response([
("set-cookie", "colorme_PHPSESSID=SECRET123; Path=/"),
("set-cookie", "theme=OTHERSECRET; Path=/")]))
assert len(cookies) == 2
assert "SECRET" not in json.dumps([asdict(f) for f in fs])
assert cookies[0]["asset_role"] == "possibly_sensitive_cookie"
assert cookies[1]["asset_role"] == "unknown"
httponly = next(f for f in fs if f.finding == "cookie_missing_httponly")
assert len(httponly.required_conditions) == 2
assert "additionally requires" in httponly.conditional_impact["description"]
def test_cookie_secure_with_hsts_stays_conditional():
_, fs = server._analyze_cookies(response({"set-cookie": "sid=x; HttpOnly; SameSite=Lax", "strict-transport-security": "max-age=31536000"}))
f = next(f for f in fs if f.finding == "cookie_missing_secure")
assert f.severity == "low"
assert "HTTPS traffic stays encrypted" in f.inference
assert any("HSTS" in c for c in f.required_conditions)
@pytest.mark.parametrize("same,code", [
("", "cookie_samesite_missing"), ("; SameSite=weird", "cookie_samesite_invalid"),
("; SameSite=None", "cookie_samesite_none_without_secure")])
def test_samesite_defaults_and_none(same, code):
cookies, fs = server._analyze_cookies(response({"set-cookie": "sid=x; HttpOnly" + same}))
assert code in codes(fs)
if "none" not in code:
assert cookies[0]["effective_modern_default"].startswith("Lax")
assert "does not prove CSRF" in next(f.inference for f in fs if f.finding == code)
def test_cookie_strong_flags_case_insensitive_and_expires():
cookies, fs = server._analyze_cookies(response({"set-cookie": "sid=x; Expires=Wed, 21 Oct 2030 07:28:00 GMT; SECURE; HTTPONLY; samesite=lAx"}))
assert len(cookies) == 1
assert fs == []
def test_samesite_none_secure_is_intent_review():
_, fs = server._analyze_cookies(response({"set-cookie": "sid=x; Secure; HttpOnly; SameSite=None"}))
f = fs[0]
assert f.finding == "cookie_samesite_none"
assert f.severity == "info"
@pytest.mark.parametrize("src,external", [
("/app.js", False), ("https://example.com:443/app.js", False),
("http://example.com/app.js", True), ("https://example.com:444/app.js", True),
("//cdn.example.net/app.js", True)])
def test_sri_full_origin_comparison(src, external):
html, fs = server._analyze_html(response({"content-type": "text/html"}, f'<script src="{src}"></script>'))
assert ("sri_missing" in codes(fs)) == external
@pytest.mark.parametrize("version,expected", [("12", "mutable_or_unpinned"), ("12.1", "mutable_or_unpinned"),
("latest", "mutable_or_unpinned"), ("12.1.2", "exact_version_candidate")])
def test_sri_version_pin(version, expected):
_, fs = server._analyze_html(response({"content-type": "text/html"},
f'<script src="https://cdn.jsdelivr.net/npm/swiper@{version}/swiper.js"></script>'))
assert ("sri_mutable_url" in codes(fs)) == (expected == "mutable_or_unpinned")
assert fs[0].evidence["version_pin"] == expected
def test_sri_attribute_not_hash_verification():
html, fs = server._analyze_html(response({"content-type": "text/html"},
'<script src="https://cdn.example.org/a.js" integrity="sha384-AAAA" crossorigin="anonymous"></script>'))
assert "sri_missing" not in codes(fs)
assert html["sri_observations"][0]["hash_verified"] is False
_, fs = server._analyze_html(response({"content-type": "text/html"},
'<script src="https://cdn.example.org/a.js" integrity="garbage"></script>'))
assert "sri_invalid_syntax" in codes(fs)
@pytest.mark.parametrize("headers,flagged", [({}, True), ({"content-security-policy": "default-src 'self'"}, True),
({"x-frame-options": "SAMEORIGIN"}, False), ({"x-frame-options": "ALLOW-FROM https://x.test"}, True),
({"content-security-policy": "frame-ancestors 'none'"}, False), ({"content-security-policy": "frame-ancestors *"}, True)])
def test_framing_without_csp(headers, flagged):
r = report(headers)
assert any(f["finding"] == "framing_restriction_missing_or_weak" for f in r["findings"]) == flagged
def test_headers_not_assumed_effective():
headers, fs = server._analyze_headers(response({"strict-transport-security": "max-age=0", "x-content-type-options": "wrong"}))
assert "strict_transport_security_missing_or_weak" in codes(fs)
assert "x_content_type_options_missing_or_weak" in codes(fs)
assert headers["strict-transport-security"]["effectiveness"] == "not_verified"
isolation = [f for f in fs if f.category == "isolation"]
assert len(isolation) == 3 and all(f.severity == "info" for f in isolation)
def test_schema_and_all_finding_invariants():
schema = json.loads(Path(__file__).with_name("finding.schema.json").read_text())
validator = Draft202012Validator(schema)
r = report({"content-type": "text/html", "set-cookie": "PHPSESSID=SECRET"},
'<script src="http://cdn.jsdelivr.net/npm/swiper@12/a.js"></script>')
assert "SECRET" not in json.dumps(r)
for finding in r["findings"]:
validator.validate(finding)
assert finding["required_conditions"]
assert finding["standalone_exploitability"] == "none"
assert "heuristic" in r["security_score_semantics"]
def test_ssrf_nonpublic_and_credentials(monkeypatch):
monkeypatch.setattr(server.socket, "getaddrinfo", lambda *a: [(2, 1, 6, "", ("127.0.0.1", 80))])
with pytest.raises(server.ToolError):
server._validate_target("http://localhost/")
for url in ["file:///a", "https://user:pass@example.com"]:
with pytest.raises(server.ToolError):
server._validate_target(url)
def test_fetch_redirect_and_stream_limit(monkeypatch):
seen = []
def validate(url):
seen.append(url)
if "127.0.0.1" in url:
raise server.ToolError("blocked")
return url
monkeypatch.setattr(server, "_validate_target", validate)
real_client = httpx.Client
def make_client(handler):
return lambda **kw: real_client(transport=httpx.MockTransport(handler), **kw)
monkeypatch.setattr(server.httpx, "Client", make_client(lambda req: httpx.Response(302, headers={"location": "http://127.0.0.1/"})))
with pytest.raises(server.ToolError, match="blocked"):
server._fetch("https://example.com/")
assert seen[-1] == "http://127.0.0.1/"
monkeypatch.setattr(server, "MAX_RESPONSE_BYTES", 5)
monkeypatch.setattr(server.httpx, "Client", make_client(lambda req: httpx.Response(200, content=b"123456")))
with pytest.raises(server.ToolError, match="limit"):
server._fetch("https://example.com/")
def test_scan_url_uses_same_schema(monkeypatch, tmp_path):
monkeypatch.setenv("WEB_SECURITY_MCP_REPORT_DIR", str(tmp_path))
monkeypatch.setattr(server, "_fetch", lambda url: (response(), [url]))
result = server.scan_url("https://example.com/")
assert result["schema_version"] == "2.0"
assert "review_markdown" not in result
assert server.scan_url("https://example.com/", include_review_markdown=True)["review_markdown"]
@pytest.mark.parametrize("encoding", ["gzip", "deflate", "identity"])
def test_fetch_compressed_response_decoded_once(monkeypatch, encoding):
import zlib
payload = b'<html><title>Compressed page</title></html>'
raw = gzip.compress(payload) if encoding == "gzip" else zlib.compress(payload) if encoding == "deflate" else payload
real_client = httpx.Client
def handler(request):
return httpx.Response(200, headers={"content-type": "text/html; charset=utf-8",
"content-encoding": encoding, "content-length": str(len(raw))}, stream=httpx.ByteStream(raw))
monkeypatch.setattr(server, "_validate_target", lambda url: url)
monkeypatch.setattr(server.httpx, "Client", lambda **kw: real_client(transport=httpx.MockTransport(handler), **kw))
result, chain = server._fetch("https://example.com/")
assert result.content == payload
assert result.headers["content-encoding"] == encoding
assert result.headers["content-length"] == str(len(raw))
assert server.analyze_response(result, chain[0], chain)["html"]["title"] == "Compressed page"
def test_compressed_expansion_obeys_limit(monkeypatch):
raw = gzip.compress(b"a" * 10000)
real_client = httpx.Client
monkeypatch.setattr(server, "_validate_target", lambda url: url)
monkeypatch.setattr(server, "MAX_RESPONSE_BYTES", 100)
monkeypatch.setattr(server.httpx, "Client", lambda **kw: real_client(
transport=httpx.MockTransport(lambda req: httpx.Response(200,
headers={"content-encoding": "gzip"}, stream=httpx.ByteStream(raw))), **kw))
with pytest.raises(server.ToolError, match="limit"):
server._fetch("https://example.com/")
def test_genuinely_invalid_gzip_is_not_accepted(monkeypatch):
real_client = httpx.Client
monkeypatch.setattr(server, "_validate_target", lambda url: url)
monkeypatch.setattr(server.httpx, "Client", lambda **kw: real_client(
transport=httpx.MockTransport(lambda req: httpx.Response(200,
headers={"content-encoding": "gzip"}, stream=httpx.ByteStream(b"not gzip"))), **kw))
with pytest.raises(server.ToolError, match="HTTP request failed"):
server._fetch("https://example.com/")
def test_stdio_modern_and_legacy():
# Launch the actual entry point; no requests to public sites.
async def run(mode):
params = StdioServerParameters(command=sys.executable,
args=[str(Path(__file__).with_name("server.py"))], env=dict(os.environ))
async with Client(params, mode=mode, read_timeout_seconds=15) as client:
listed = await client.list_tools()
assert {t.name for t in listed.tools} == {"scan_url", "explain_capabilities"}
result = await client.call_tool("explain_capabilities", {})
assert not result.is_error
assert result.structured_content["schema_version"] == "2.0"
bad = await client.call_tool("scan_url", {"url": "file:///forbidden"})
assert bad.is_error
asyncio.run(run("auto"))
asyncio.run(run("legacy"))