File size: 7,919 Bytes
bc6415e 4643d92 bc6415e 4643d92 bc6415e 5c95040 bc6415e | 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 | import asyncio
import json
from pathlib import Path
from unittest.mock import patch
import httpx
import pytest
from app import client_modules as tools, config, main, nuclei_catalog, pipeline, web_modules
from test_worker_pipeline import run_job
def test_script_inventory_keeps_same_origin_and_javascript_inline_only():
parser = tools.ScriptInventory("https://example.org/")
parser.feed('<script src="/app.js"></script><script src="https://outside.org/a.js"></script>'
'<script src="https://example.org:8443/a.js"></script>'
'<script>/*! jQuery v1.6.1 */</script><script type="application/ld+json">{}</script>')
assert parser.urls == ["https://example.org/app.js"]
assert len(parser.inline) == 1
def test_retire_no_javascript_skips_without_starting_node():
async def fetch(*args): return b"<html>No scripts</html>", "text/html", "https://example.org/"
with patch.object(tools, "fetch_sample", fetch), patch.object(tools, "execute") as execute:
result = asyncio.run(tools.retirejs("https://example.org/", "example.org", "Test", 1.5, []))
assert result["status"] == "skipped"
assert not execute.called
def test_retire_fetch_failure_does_not_claim_a_clean_scan():
async def fetch(*args): raise httpx.ReadTimeout("mock")
with patch.object(tools, "fetch_sample", fetch), patch.object(tools, "execute") as execute:
result = asyncio.run(tools.retirejs("https://example.org/", "example.org", "Test", 1.5, []))
assert result["status"] == "skipped"
assert "ReadTimeout" in result["reason"]
assert not execute.called
def test_retire_samples_bounded_files_uses_bundled_database_and_maps_findings():
fetches = []
async def fetch(url, *args):
fetches.append(url)
if url == "https://example.org/":
scripts = "".join(f'<script src="/jquery-{i}.js"></script>' for i in range(8))
return scripts.encode(), "text/html", url
return b"/* sample */", "application/javascript", url
async def execute(args, timeout):
assert "--includeOsv" not in args
assert args[args.index("--jsrepo") + 1] == "/opt/retire-db/jsrepository-v5.json"
root = Path(args[args.index("--path") + 1])
files = sorted(root.rglob("*.js"))
assert len(files) == config.RETIRE_MAX_FILES
assert files[0].name == "jquery-0.js"
report = {"data": [{"file": str(files[0]), "results": [
{"component": "jquery", "version": "1.6.1", "vulnerabilities": [
{"severity": "high", "identifiers": {"CVE": ["CVE-TEST"], "summary": "Mock candidate"}}]}]}],
"errors": []}
Path(args[args.index("--outputpath") + 1]).write_text(json.dumps(report))
return ["[diagnostic] Process exit code: 13"], "failed"
async def sleep(*args): pass
with patch.object(tools, "fetch_sample", fetch), patch.object(tools, "execute", execute), \
patch.object(tools.asyncio, "sleep", sleep):
result = asyncio.run(tools.retirejs("https://example.org/", "example.org", "Test", 1.5,
["https://outside.org/out.js"]))
assert result["status"] == "findings" and result["early_stop"]
assert result["coverage_complete"] is False # Deliberate sample cap, findings preserved.
assert result["sampled_files"] == 4
assert result["alerts"][0]["url"] == "https://example.org/jquery-0.js"
assert result["alerts"][0]["risk"] == "High"
assert len(fetches) == 5
assert "Process exit code: 13" not in result["reason"]
def test_retire_unknown_report_path_is_never_assigned_to_the_target(tmp_path):
known = str((tmp_path / "known.js").resolve())
report = {"data": [{"file": str(tmp_path / "outside.js"), "results": [
{"component": "x", "version": "1", "vulnerabilities": [{"severity": "high"}]}]}]}
assert tools.parse_retire(report, {known: "https://example.org/a.js"}) == []
@pytest.mark.parametrize("guard_stats,expected", [
({"requests": 2}, "completed"),
({"requests": 12, "limit_reached": True}, "partial"),
({"requests": 1, "blocked": 1}, "partial"),
])
def test_wapiti_cli_profile_and_coverage(guard_stats, expected):
async def execute(args, timeout):
assert args[args.index("-m") + 1] == "csp,http_headers,cookieflags"
assert args[args.index("--scope") + 1] == "url"
assert args[args.index("--headless") + 1] == "no"
assert "--no-bugreport" in args
Path(args[args.index("-o") + 1]).write_text(json.dumps({
"vulnerabilities": {"Header": [{"method": "GET", "path": "/",
"level": 1, "info": "missing", "module": "http_headers"}]}}))
Path(args[args.index("--asfb-stats") + 1]).write_text(json.dumps(guard_stats))
return [], "completed"
with patch.object(tools, "execute", execute):
result = asyncio.run(tools.wapiti("https://example.org/", "example.org", "Test", 1.5))
assert result["status"] == expected
assert result["alerts"][0]["source"] == "wapiti"
def test_wapiti_missing_report_is_failed():
async def execute(*args): return [], "completed"
with patch.object(tools, "execute", execute):
result = asyncio.run(tools.wapiti("https://example.org/", "example.org", "Test", 1.5))
assert result["status"] == "failed"
def test_wapiti_parser_filters_other_origins_and_methods():
report = {"vulnerabilities": {"Mock": [
{"method": "GET", "path": "https://outside.org/", "info": "outside"},
{"method": "POST", "path": "/", "info": "wrong method"},
{"method": "GET", "path": "/", "info": "kept", "level": 1}]}}
findings = tools.parse_wapiti(report, "https://example.org/")
assert len(findings) == 1 and findings[0]["description"] == "kept"
def test_new_module_dispatch_receives_discovery_inventory():
seen = []
async def retire(*args):
seen.extend(args[-1])
return {"status": "skipped", "alerts": [], "reason": "No JS"}
with patch.object(tools, "retirejs", retire):
result = asyncio.run(web_modules.run_web_module(
"retirejs", "https://example.org/", "example.org", "Test",
discovered=["https://example.org/a.js"]))
assert result["status"] == "skipped"
assert seen == ["https://example.org/a.js"]
def test_worker_new_modules_and_skip_are_isolated():
run, job = run_job(["retirejs", "wapiti", "httpx"])
async def module(name, *args):
return {"status": "skipped" if name == "retirejs" else "completed", "alerts": [], "output": name}
async def probe(*args): return {"status": "completed", "probes": []}
with patch.object(main, "public_host"), patch.object(main, "run_web_module", module), \
patch.object(main, "probe", probe):
asyncio.run(run())
assert job["status"] == "completed"
assert job["pipeline"]["retirejs"] == "skipped"
assert job["pipeline"]["httpx"] == "completed"
def test_optional_nuclei_templates_validate_and_do_not_change_defaults():
for name in ("directory-listing", "openapi-json"):
assert name in nuclei_catalog.builtins()
assert name not in config.NUCLEI_DEFAULT_IDS
content = nuclei_catalog.bundled_content(name)
nuclei_catalog.validate_template(content)
assert nuclei_catalog.estimate_requests(content) <= 2
def test_openapi_fixed_paths_run_only_at_origin():
async def execute(args, deadline):
targets = Path(args[args.index("-l") + 1]).read_text().splitlines()
assert targets == ["https://example.org/"]
return [], "completed"
with patch.object(pipeline, "execute", execute):
result = asyncio.run(pipeline.nuclei("https://example.org/", "example.org", "Test",
["https://example.org/app.js"], default_ids=["openapi-json"]))
assert result["targets_checked"] == 1
|