From 418f5baf36e3efe8bf3598e921c3fc7a5ada8ed8 Mon Sep 17 00:00:00 2001 From: ek0ms savi0r Date: Sun, 21 Jun 2026 17:44:13 +0000 Subject: [PATCH] Upload files to "tests" --- tests/test_cli.py | 57 +++++++++++++++ tests/test_exposure.py | 129 +++++++++++++++++++++++++++++++++ tests/test_storage.py | 43 +++++++++++ tests/test_visualize.py | 130 ++++++++++++++++++++++++++++++++++ tests/test_vulnerabilities.py | 48 +++++++++++++ 5 files changed, 407 insertions(+) create mode 100644 tests/test_cli.py create mode 100644 tests/test_exposure.py create mode 100644 tests/test_storage.py create mode 100644 tests/test_visualize.py create mode 100644 tests/test_vulnerabilities.py diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..5861853 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,57 @@ +import io +import unittest +from contextlib import redirect_stderr +from unittest.mock import patch + +from redflare.cli import main +from redflare.interactive import interactive_arguments + + +class CLITests(unittest.TestCase): + def test_scan_requires_authorization_acknowledgement(self): + error = io.StringIO() + with redirect_stderr(error): + code = main(["scan", "http://127.0.0.1"]) + self.assertEqual(code, 2) + self.assertIn("--authorized", error.getvalue()) + + @patch( + "builtins.input", + side_effect=[ + "1", # full profile + "1", # direct targets + "https://example.test", # target + "", # no scope file + "yes", # authorized + "no", # not public + "yes", # full interaction permitted + "runs", # output + "1", # workers + "10", # timeout + "", # no wordlist + "1", # rate + "25", # max paths + "30", # max crawl pages + "2", # max crawl depth + "no", # no GraphQL introspection + "75", # max exposure endpoints + "https://github.com/o/r", # repository + ], + ) + def test_interactive_wizard_builds_full_pipeline(self, _): + arguments = interactive_arguments() + self.assertIn("full", arguments) + self.assertIn("--authorized", arguments) + self.assertIn("--github-repo", arguments) + self.assertIn("https://example.test", arguments) + + @patch("redflare.interactive.recent_runs", return_value=[]) + @patch("builtins.input", side_effect=["4", "file:///tmp/run_fixture", "8765", "yes"]) + def test_interactive_menu_builds_visualizer_command(self, _, __): + arguments = interactive_arguments() + self.assertEqual(arguments[:2], ["visualize", "file:///tmp/run_fixture"]) + self.assertNotIn("--no-browser", arguments) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_exposure.py b/tests/test_exposure.py new file mode 100644 index 0000000..17b8c00 --- /dev/null +++ b/tests/test_exposure.py @@ -0,0 +1,129 @@ +from __future__ import annotations + +import json +import io +import tempfile +import threading +import unittest +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +from redflare.core.models import Target +from redflare.core.standards import enrich_finding +from redflare.core.storage import RunStore +from redflare.modules.base import ModuleContext +from redflare.modules.exposure import SensitiveExposureModule +from redflare.ui import LiveConsole + + +FAKE_TOKEN = "ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890" +FAKE_HASH = "$2b$12$" + "A" * 53 + + +class ExposureFixture(BaseHTTPRequestHandler): + def do_GET(self): + if self.path == "/": + body = b'' + content_type = "text/html" + status = 200 + elif self.path == "/app.js": + body = ( + f'const api_key = "{FAKE_TOKEN}";\n' + f'const password_hash = "{FAKE_HASH}";\n' + 'const backend = "10.20.30.40";\n' + ).encode() + content_type = "application/javascript" + status = 200 + else: + body = b"not found" + content_type = "text/plain" + status = 404 + self.send_response(status) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *_): + return + + +class ExposureTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.server = ThreadingHTTPServer(("127.0.0.1", 0), ExposureFixture) + cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True) + cls.thread.start() + + @classmethod + def tearDownClass(cls): + cls.server.shutdown() + cls.server.server_close() + + def test_detects_and_redacts_sensitive_values(self): + text = f'api_key="{FAKE_TOKEN}"; password_hash="{FAKE_HASH}"; host="10.20.30.40"' + matches = SensitiveExposureModule.detect(text, "https://example.test/app.js") + types = {item["type"] for item in matches} + self.assertIn("GitHub token", types) + self.assertIn("bcrypt password hash", types) + self.assertIn("private/internal IP address", types) + serialized = json.dumps(matches) + self.assertNotIn(FAKE_TOKEN, serialized) + self.assertNotIn(FAKE_HASH, serialized) + self.assertIn("10.20.30.40", serialized) + + def test_detects_labeled_public_ip_without_flagging_documentation_ip(self): + matches = SensitiveExposureModule.detect( + 'client_ip="8.8.8.8"; client_ip="192.0.2.10"', + "https://example.test/profile", + ) + previews = {item["value_preview"] for item in matches} + self.assertIn("8.8.8.8", previews) + self.assertNotIn("192.0.2.10", previews) + + def test_scans_graph_endpoints_and_writes_masked_artifact(self): + with tempfile.TemporaryDirectory() as directory: + url = f"http://127.0.0.1:{self.server.server_port}" + target = Target(url, "127.0.0.1", "http", self.server.server_port) + messages = [] + context = ModuleContext( + "exposure-test", + Path(directory), + timeout=2, + reporter=lambda *items: messages.append(items), + ) + context.surface_graph.add_endpoint( + target.url, f"{url}/app.js", method="GET", source="test" + ) + result = SensitiveExposureModule().run(target, context) + for finding in result.findings: + enrich_finding(finding) + self.assertGreaterEqual(len(result.findings), 3) + self.assertTrue(any(item[2] == "finding" for item in messages)) + self.assertTrue(all(item.test_id == "RFV2-DATA-001" for item in result.findings)) + artifact = Path(result.artifacts[0]) + self.assertTrue(artifact.exists()) + saved = artifact.read_text(encoding="utf-8") + self.assertNotIn(FAKE_TOKEN, saved) + self.assertNotIn(FAKE_HASH, saved) + self.assertIn("10.20.30.40", saved) + + store = RunStore(directory, "report-test") + store.write_result(result) + summary = store.finalize([result], surface_graph=context.surface_graph.snapshot()) + self.assertEqual(summary["sensitive_exposures"], len(result.findings)) + report = (store.root / "report.html").read_text(encoding="utf-8") + jsonl = (store.root / "findings.jsonl").read_text(encoding="utf-8") + self.assertIn("RFV2-DATA-001", report) + self.assertIn("10.20.30.40", report) + self.assertNotIn(FAKE_TOKEN, report + jsonl) + terminal = io.StringIO() + LiveConsole(stream=terminal).final_report(summary, [result], store.root) + terminal_output = terminal.getvalue() + self.assertIn("RFV2-DATA-001", terminal_output) + self.assertIn("10.20.30.40", terminal_output) + self.assertNotIn(FAKE_TOKEN, terminal_output) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_storage.py b/tests/test_storage.py new file mode 100644 index 0000000..3f325d0 --- /dev/null +++ b/tests/test_storage.py @@ -0,0 +1,43 @@ +import tempfile +import unittest + +from redflare.core.models import Finding +from redflare.core.models import Target +from redflare.core.storage import RunStore, deduplicate_findings, target_run_id + + +class StorageTests(unittest.TestCase): + def test_target_run_name_is_human_readable(self): + value = target_run_id([Target("https://example.test", "example.test", "https", 443)]) + self.assertRegex(value, r"^scan_example\.test_\d{8}_\d{6}$") + + def test_run_folders_do_not_collide(self): + with tempfile.TemporaryDirectory() as directory: + first = RunStore(directory, "run_test") + second = RunStore(directory, "run_test") + self.assertNotEqual(first.root, second.root) + self.assertTrue(first.root.is_dir()) + self.assertTrue(second.root.is_dir()) + + def test_corroborated_header_findings_are_deduplicated(self): + common = { + "run_id": "run_test", + "target": "https://example.test", + "severity": "low", + "confidence": 0.95, + "description": "missing headers", + } + findings = [ + Finding(module="http_headers", category="security-headers", title="Missing headers", **common), + Finding(module="gatekeeper", category="browser-security-headers", title="Browser missing headers", **common), + ] + deduplicated = deduplicate_findings(findings) + self.assertEqual(len(deduplicated), 1) + self.assertEqual( + deduplicated[0].evidence["corroborated_by"], + ["gatekeeper", "http_headers"], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_visualize.py b/tests/test_visualize.py new file mode 100644 index 0000000..c14c868 --- /dev/null +++ b/tests/test_visualize.py @@ -0,0 +1,130 @@ +from __future__ import annotations + +import json +import tempfile +import threading +import unittest +from importlib.resources import files +from pathlib import Path +from urllib.request import urlopen + +from redflare.visualize import VisualServer, build_visual_graph, resolve_run_directory + + +class VisualizeTests(unittest.TestCase): + def make_run(self, directory: str) -> Path: + root = Path(directory) / "run_visual" + root.mkdir() + (root / "summary.json").write_text( + json.dumps({"run_id": "run_visual", "findings": 1}), encoding="utf-8" + ) + (root / "attack_surface.json").write_text( + json.dumps( + { + "summary": {"targets": 1}, + "targets": { + "https://example.test": { + "endpoints": [ + { + "url": "https://example.test/api/users", + "methods": ["GET"], + "parameters": [ + {"name": "id", "location": "query", "required": True} + ], + "sources": ["javascript-route"], + } + ], + "edges": [], + "documents": [ + {"kind": "openapi", "title": "Fixture API", "url": "https://example.test/openapi.json"} + ], + } + }, + } + ), + encoding="utf-8", + ) + finding = { + "id": "finding123", + "target": "https://example.test", + "module": "sensitive_exposure", + "category": "sensitive-data-exposure", + "title": "Potential token exposed", + "severity": "critical", + "evidence": { + "url": "https://example.test/api/users", + "value_preview": "ghp_…7890", + }, + "standards": { + "CWE": [{"id": "CWE-798", "url": "https://cwe.mitre.org/data/definitions/798.html"}], + "CVE": [{"id": "CVE-2026-12345", "url": "https://nvd.nist.gov/vuln/detail/CVE-2026-12345"}], + }, + } + (root / "findings.jsonl").write_text(json.dumps(finding) + "\n", encoding="utf-8") + modules = root / "modules" + modules.mkdir() + (modules / "example__sensitive_exposure.json").write_text( + json.dumps( + { + "module": "sensitive_exposure", + "target": "https://example.test", + "status": "completed", + "findings": [finding], + "observations": {"responses_scanned": 2}, + "artifacts": [], + "errors": [], + "duration_seconds": 0.1, + } + ), + encoding="utf-8", + ) + return root + + def test_normalizes_run_into_typed_graph(self): + with tempfile.TemporaryDirectory() as directory: + graph = build_visual_graph(self.make_run(directory)) + types = {node["type"] for node in graph["nodes"]} + self.assertTrue({"run", "target", "endpoint", "parameter", "document", "module", "exposure", "standard", "cve"} <= types) + relations = {edge["type"] for edge in graph["edges"]} + self.assertTrue({"contains", "serves", "accepts", "executed", "reported", "exposes", "maps_to"} <= relations) + self.assertEqual(graph["metadata"]["severity_counts"]["critical"], 1) + + def test_loopback_server_serves_ui_and_graph(self): + with tempfile.TemporaryDirectory() as directory: + server = VisualServer(self.make_run(directory), 0) + thread = threading.Thread(target=server.httpd.serve_forever, daemon=True) + thread.start() + try: + with urlopen(server.url, timeout=2) as response: + self.assertIn(b"Visual investigation console", response.read()) + self.assertIn("default-src 'self'", response.headers["Content-Security-Policy"]) + with urlopen(server.url + "api/graph", timeout=2) as response: + graph = json.loads(response.read()) + self.assertEqual(graph["metadata"]["run_id"], "run_visual") + finally: + server.httpd.shutdown() + server.httpd.server_close() + thread.join(timeout=2) + + def test_rejects_non_run_directory(self): + with tempfile.TemporaryDirectory() as directory: + with self.assertRaises(ValueError): + build_visual_graph(directory) + + def test_accepts_local_file_url_for_run_directory(self): + with tempfile.TemporaryDirectory() as directory: + run = self.make_run(directory) + self.assertEqual(resolve_run_directory(run.as_uri()), run.resolve()) + self.assertEqual(build_visual_graph(run.as_uri())["metadata"]["run_id"], "run_visual") + + def test_visual_assets_keep_node_clicks_on_the_node(self): + app = files("redflare.web").joinpath("app.js").read_text(encoding="utf-8") + styles = files("redflare.web").joinpath("styles.css").read_text(encoding="utf-8") + self.assertIn("group.setPointerCapture(event.pointerId)", app) + self.assertIn("selectNode(node.id)", app) + self.assertIn("centerOnNode(node)", app) + self.assertIn('.graph-stage.dense .node[data-type="endpoint"] .node-label', styles) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_vulnerabilities.py b/tests/test_vulnerabilities.py new file mode 100644 index 0000000..8bc5e6f --- /dev/null +++ b/tests/test_vulnerabilities.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from redflare.core.models import Target +from redflare.core.standards import enrich_finding +from redflare.modules.base import ModuleContext +from redflare.modules.http import HTTPResponse +from redflare.modules.vulnerabilities import CVEIntelligenceModule, fingerprint_response + + +class CVEIntelligenceTests(unittest.TestCase): + def test_fingerprints_only_explicit_versions(self): + body = b'' + values = fingerprint_response({"server": "nginx/1.24.0", "x-powered-by": "Express"}, body) + products = {(item.product, item.version) for item in values} + self.assertEqual(products, {("nginx", "1.24.0"), ("WordPress", "6.4.2"), ("jQuery", "3.6.0")}) + self.assertNotIn("Express", {item.product for item in values}) + + @patch("redflare.modules.vulnerabilities.query_nvd") + @patch("redflare.modules.vulnerabilities.request") + def test_emits_cve_findings_with_clickable_reference(self, get, nvd): + get.return_value = HTTPResponse("https://example.test", 200, {"server": "nginx/1.24.0"}, b"") + nvd.return_value = [{ + "id": "CVE-2026-12345", + "published": "2026-01-02T00:00:00Z", + "lastModified": "2026-01-03T00:00:00Z", + "vulnStatus": "Analyzed", + "descriptions": [{"lang": "en", "value": "Fixture vulnerability."}], + "metrics": {"cvssMetricV31": [{"cvssData": {"baseScore": 9.8, "baseSeverity": "CRITICAL", "vectorString": "CVSS:3.1/AV:N"}}]}, + "references": [{"url": "https://vendor.example/advisory"}], + }] + with tempfile.TemporaryDirectory() as directory: + context = ModuleContext("run_test", Path(directory), timeout=1) + result = CVEIntelligenceModule().run(Target("https://example.test", "example.test", "https", 443), context) + self.assertEqual(len(result.findings), 1) + finding = enrich_finding(result.findings[0]) + self.assertEqual(finding.test_id, "RFV2-COMP-001") + self.assertEqual(finding.severity, "critical") + self.assertEqual(finding.standards["CVE"][0]["id"], "CVE-2026-12345") + self.assertTrue(finding.standards["CVE"][0]["url"].startswith("https://nvd.nist.gov/")) + + +if __name__ == "__main__": + unittest.main()