From aba4034d0e9a92a5a060cd700e136ee86a512377 Mon Sep 17 00:00:00 2001 From: ek0ms savi0r Date: Sun, 21 Jun 2026 17:44:43 +0000 Subject: [PATCH] Upload files to "tests" --- tests/test_mapping.py | 145 ++++++++++++++++++++++++++++++++++++++++ tests/test_scope.py | 26 +++++++ tests/test_standards.py | 24 +++++++ 3 files changed, 195 insertions(+) create mode 100644 tests/test_mapping.py create mode 100644 tests/test_scope.py create mode 100644 tests/test_standards.py diff --git a/tests/test_mapping.py b/tests/test_mapping.py new file mode 100644 index 0000000..2312b70 --- /dev/null +++ b/tests/test_mapping.py @@ -0,0 +1,145 @@ +from __future__ import annotations + +import json +import threading +import unittest +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +from redflare.core.models import Target +from redflare.modules.base import ModuleContext +from redflare.modules.mapping import ApplicationMappingModule + + +class MappingFixture(BaseHTTPRequestHandler): + def do_GET(self): + if self.path == "/": + body = b'''Account +
+ ''' + content_type = "text/html" + status = 200 + elif self.path == "/app.js": + body = b'fetch("/api/session"); axios.post("/api/messages", {});' + content_type = "application/javascript" + status = 200 + elif self.path == "/openapi.json": + body = json.dumps( + { + "openapi": "3.0.3", + "info": {"title": "Fixture API"}, + "paths": { + "/api/users/{id}": { + "get": { + "parameters": [ + {"name": "id", "in": "path", "required": True, "schema": {"type": "string"}} + ], + "security": [{"bearerAuth": []}], + } + }, + "/api/users": { + "post": { + "requestBody": { + "required": True, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["email"], + "properties": {"email": {"type": "string"}}, + } + } + }, + } + } + }, + }, + } + ).encode() + content_type = "application/json" + status = 200 + elif self.path.startswith("/account"): + body = b"account" + content_type = "text/html" + 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 do_POST(self): + if self.path != "/graphql": + self.send_response(404) + self.end_headers() + return + body = json.dumps( + { + "data": { + "__schema": { + "queryType": {"name": "Query"}, + "mutationType": {"name": "Mutation"}, + "types": [ + {"name": "Query", "kind": "OBJECT", "fields": [{"name": "viewer", "args": []}]}, + {"name": "Mutation", "kind": "OBJECT", "fields": [{"name": "updateProfile", "args": []}]}, + ], + } + } + } + ).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *_): + return + + +class MappingTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.server = ThreadingHTTPServer(("127.0.0.1", 0), MappingFixture) + 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_builds_graph_from_html_javascript_and_openapi(self): + url = f"http://127.0.0.1:{self.server.server_port}" + target = Target(url, "127.0.0.1", "http", self.server.server_port) + context = ModuleContext( + "mapping-test", + Path("/tmp"), + timeout=2, + max_crawl_pages=4, + max_crawl_depth=1, + graphql_introspection=True, + ) + result = ApplicationMappingModule().run(target, context) + self.assertEqual(result.status, "completed") + endpoints = context.surface_graph.snapshot()["targets"][url]["endpoints"] + by_url = {item["url"]: item for item in endpoints} + self.assertIn(f"{url}/login", by_url) + self.assertIn(f"{url}/api/session", by_url) + self.assertIn(f"{url}/api/messages", by_url) + self.assertIn(f"{url}/api/users/{{id}}", by_url) + self.assertIn("bearerAuth", by_url[f"{url}/api/users/{{id}}"]["authentication"]) + body_parameters = by_url[f"{url}/api/users"]["parameters"] + self.assertIn("email", {item["name"] for item in body_parameters}) + self.assertEqual( + by_url[f"{url}/graphql"]["metadata"]["graphql_operations"], + ["updateProfile", "viewer"], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_scope.py b/tests/test_scope.py new file mode 100644 index 0000000..9d305b4 --- /dev/null +++ b/tests/test_scope.py @@ -0,0 +1,26 @@ +import unittest +from unittest.mock import patch + +from redflare.core.scope import ScopeError, ScopePolicy, normalize_target + + +class ScopeTests(unittest.TestCase): + def test_normalizes_bare_hostname(self): + target = normalize_target("localhost:8080") + self.assertEqual(target.url, "https://localhost:8080") + self.assertEqual(target.port, 8080) + + @patch("redflare.core.scope.is_public_host", return_value=False) + def test_allowed_host(self, _): + target = normalize_target("http://localhost:8080") + ScopePolicy({"localhost"}).validate(target) + + @patch("redflare.core.scope.is_public_host", return_value=False) + def test_rejects_host_outside_manifest(self, _): + target = normalize_target("http://other.local") + with self.assertRaises(ScopeError): + ScopePolicy({"allowed.local"}).validate(target) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_standards.py b/tests/test_standards.py new file mode 100644 index 0000000..23af3ea --- /dev/null +++ b/tests/test_standards.py @@ -0,0 +1,24 @@ +import unittest + +from redflare.core.models import Finding +from redflare.core.standards import TEST_REGISTRY, enrich_finding + + +class StandardsTests(unittest.TestCase): + def test_registry_ids_are_unique_and_versioned(self): + ids = [item.id for item in TEST_REGISTRY] + self.assertEqual(len(ids), len(set(ids))) + self.assertTrue(all(item.wstg and item.asvs and item.cwe and item.api_security for item in TEST_REGISTRY)) + self.assertTrue(all(ref.startswith("WSTG-") for item in TEST_REGISTRY for ref in item.wstg)) + self.assertTrue(all(ref.startswith("v5.0.0-") for item in TEST_REGISTRY for ref in item.asvs)) + + def test_enriches_known_finding(self): + finding = Finding("run", "https://example.test", "http_headers", "security-headers", "Missing", "low", 1.0, "desc") + enrich_finding(finding) + self.assertEqual(finding.test_id, "RFV2-CONF-001") + self.assertIn("OWASP_ASVS", finding.standards) + self.assertEqual(finding.standards["OWASP_WSTG"][0]["version"], "4.2") + + +if __name__ == "__main__": + unittest.main()