Upload files to "tests"

2026-06-21 17:44:43 +00:00
parent a51dd25ac4
commit aba4034d0e
3 changed files with 195 additions and 0 deletions
@@ -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'''<html><a href="/account?tab=profile">Account</a>
+                <form action="/login" method="post"><input name="username"><input name="password" type="password"></form>
+                <script src="/app.js"></script></html>'''
+            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"<html>account</html>"
+            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()
@@ -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()
@@ -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()