From 5dff6f6f510bd2c21355a22bbf639ff9b9737298 Mon Sep 17 00:00:00 2001 From: ek0ms savi0r Date: Sun, 21 Jun 2026 17:55:04 +0000 Subject: [PATCH] Upload files to "redflare/core" --- redflare/core/__init__.py | 1 + redflare/core/correlation.py | 44 +++++++++++ redflare/core/standards.py | 147 +++++++++++++++++++++++++++++++++++ 3 files changed, 192 insertions(+) create mode 100644 redflare/core/__init__.py create mode 100644 redflare/core/correlation.py create mode 100644 redflare/core/standards.py diff --git a/redflare/core/__init__.py b/redflare/core/__init__.py new file mode 100644 index 0000000..2229756 --- /dev/null +++ b/redflare/core/__init__.py @@ -0,0 +1 @@ +"""Core orchestration primitives.""" diff --git a/redflare/core/correlation.py b/redflare/core/correlation.py new file mode 100644 index 0000000..26c5903 --- /dev/null +++ b/redflare/core/correlation.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +from collections import defaultdict + +from .models import Finding, ModuleResult + + +def correlate(run_id: str, results: list[ModuleResult]) -> list[ModuleResult]: + by_target: dict[str, list[ModuleResult]] = defaultdict(list) + for result in results: + by_target[result.target].append(result) + + correlated = [] + for target, target_results in by_target.items(): + path_hits = [ + finding + for result in target_results + for finding in result.findings + if finding.category == "discovered-path" and finding.severity == "medium" + ] + header_gaps = [ + finding + for result in target_results + for finding in result.findings + if finding.category == "security-headers" + ] + if path_hits and header_gaps: + finding = Finding( + run_id, + target, + "correlation", + "exposed-surface", + "Sensitive web surface with weak response hardening", + "medium", + 0.85, + "REDflare correlated an accessible sensitive path with missing security headers.", + { + "paths": [item.evidence.get("path") for item in path_hits], + "missing_headers": header_gaps[0].evidence.get("missing", []), + }, + ["correlated", "web"], + ) + correlated.append(ModuleResult("correlation", target, findings=[finding])) + return correlated diff --git a/redflare/core/standards.py b/redflare/core/standards.py new file mode 100644 index 0000000..5972ca2 --- /dev/null +++ b/redflare/core/standards.py @@ -0,0 +1,147 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from .models import Finding + + +@dataclass(frozen=True) +class TestDefinition: + id: str + name: str + module: str + category: str + wstg: tuple[str, ...] = () + asvs: tuple[str, ...] = () + cwe: tuple[str, ...] = () + api_security: tuple[str, ...] = () + + def references(self) -> dict[str, list[dict[str, str]]]: + return { + "OWASP_WSTG": [reference("OWASP_WSTG", value) for value in self.wstg], + "OWASP_ASVS": [reference("OWASP_ASVS", value) for value in self.asvs], + "CWE": [reference("CWE", value) for value in self.cwe], + "OWASP_API_SECURITY": [reference("OWASP_API_SECURITY", value) for value in self.api_security], + } + + +WSTG_URLS = { + "WSTG-INFO-06": "https://owasp.org/www-project-web-security-testing-guide/v42/4-Web_Application_Security_Testing/01-Information_Gathering/06-Identify_Application_Entry_Points", + "WSTG-INFO-07": "https://owasp.org/www-project-web-security-testing-guide/v42/4-Web_Application_Security_Testing/01-Information_Gathering/07-Map_Execution_Paths_Through_Application", + "WSTG-INFO-05": "https://owasp.org/www-project-web-security-testing-guide/v42/4-Web_Application_Security_Testing/01-Information_Gathering/05-Review_Webpage_Content_for_Information_Leakage", + "WSTG-CONF-04": "https://owasp.org/www-project-web-security-testing-guide/v42/4-Web_Application_Security_Testing/02-Configuration_and_Deployment_Management_Testing/04-Review_Old_Backup_and_Unreferenced_Files_for_Sensitive_Information", + "WSTG-CONF-05": "https://owasp.org/www-project-web-security-testing-guide/v42/4-Web_Application_Security_Testing/02-Configuration_and_Deployment_Management_Testing/05-Enumerate_Infrastructure_and_Application_Admin_Interfaces", + "WSTG-CONF-07": "https://owasp.org/www-project-web-security-testing-guide/v42/4-Web_Application_Security_Testing/02-Configuration_and_Deployment_Management_Testing/07-Test_HTTP_Strict_Transport_Security", + "WSTG-ATHZ-02": "https://owasp.org/www-project-web-security-testing-guide/v42/4-Web_Application_Security_Testing/05-Authorization_Testing/02-Testing_for_Bypassing_Authorization_Schema", + "WSTG-CLNT-04": "https://owasp.org/www-project-web-security-testing-guide/v42/4-Web_Application_Security_Testing/11-Client-side_Testing/04-Testing_for_Client-side_URL_Redirect", +} + +API_SECURITY_URLS = { + "API1:2023": "https://owasp.org/API-Security/editions/2023/en/0xa1-broken-object-level-authorization/", + "API3:2023": "https://owasp.org/API-Security/editions/2023/en/0xa3-broken-object-property-level-authorization/", + "API8:2023": "https://owasp.org/API-Security/editions/2023/en/0xa8-security-misconfiguration/", + "API9:2023": "https://owasp.org/API-Security/editions/2023/en/0xa9-improper-inventory-management/", + "API10:2023": "https://owasp.org/API-Security/editions/2023/en/0xaa-unsafe-consumption-of-apis/", +} + + +def reference(family: str, identifier: str) -> dict[str, str]: + if family == "OWASP_WSTG": + return {"id": identifier, "version": "4.2", "url": WSTG_URLS[identifier]} + if family == "OWASP_ASVS": + return { + "id": identifier, + "version": "5.0.0", + "url": "https://owasp.org/www-project-application-security-verification-standard/", + } + if family == "CWE": + number = identifier.removeprefix("CWE-") + return {"id": identifier, "url": f"https://cwe.mitre.org/data/definitions/{number}.html"} + return { + "id": identifier, + "version": "2023", + "url": API_SECURITY_URLS[identifier], + } + + +TEST_REGISTRY: tuple[TestDefinition, ...] = ( + TestDefinition( + "RFV2-COMP-001", "Correlate disclosed component versions with known vulnerabilities", + "cve_intelligence", "known-vulnerable-component", + ("WSTG-INFO-05",), ("v5.0.0-14.2.1",), ("CWE-1104",), ("API9:2023",), + ), + TestDefinition( + "RFV2-MAP-001", "Build application entry-point inventory", "application_mapping", "attack-surface-inventory", + ("WSTG-INFO-06", "WSTG-INFO-07"), ("v5.0.0-8.1.1", "v5.0.0-13.4.5"), ("CWE-1059",), ("API9:2023",), + ), + TestDefinition( + "RFV2-CONF-001", "Review browser security headers", "http_headers", "security-headers", + ("WSTG-CONF-07",), + ("v5.0.0-3.4.1", "v5.0.0-3.4.3", "v5.0.0-3.4.4", "v5.0.0-3.4.5"), + ("CWE-693",), ("API8:2023",), + ), + TestDefinition( + "RFV2-INFO-002", "Identify credential-entry surfaces", "surface_analysis", "credential-surface", + ("WSTG-INFO-06",), ("v5.0.0-6.1.3",), ("CWE-200",), ("API9:2023",), + ), + TestDefinition( + "RFV2-CLNT-001", "Identify client-side redirects", "surface_analysis", "client-redirect", + ("WSTG-CLNT-04",), ("v5.0.0-1.2.2",), ("CWE-601",), ("API10:2023",), + ), + TestDefinition( + "RFV2-CONF-002", "Discover exposed application paths", "path_discovery", "discovered-path", + ("WSTG-CONF-04", "WSTG-CONF-05"), ("v5.0.0-13.4.5",), ("CWE-200",), ("API9:2023",), + ), + TestDefinition( + "RFV2-DATA-001", "Detect sensitive data in client-accessible responses", "sensitive_exposure", "sensitive-data-exposure", + ("WSTG-INFO-05",), ("v5.0.0-13.4.7", "v5.0.0-14.2.3"), ("CWE-200", "CWE-798"), ("API3:2023", "API8:2023"), + ), + TestDefinition( + "RFV2-BROW-001", "Corroborate headers in a browser runtime", "browser_runtime", "browser-security-headers", + ("WSTG-CONF-07",), ("v5.0.0-3.4.1", "v5.0.0-3.4.3"), ("CWE-693",), ("API8:2023",), + ), + TestDefinition( + "RFV2-BROW-002", "Inventory browser-observed endpoints", "browser_runtime", "browser-endpoints", + ("WSTG-INFO-06", "WSTG-INFO-07"), ("v5.0.0-8.1.1",), ("CWE-1059",), ("API9:2023",), + ), + TestDefinition( + "RFV2-ATHZ-001", "Identify unauthenticated service surfaces", "authorization_surface", "unauthenticated-service", + ("WSTG-ATHZ-02",), ("v5.0.0-8.2.1",), ("CWE-862",), ("API1:2023",), + ), + TestDefinition( + "RFV2-CORR-001", "Correlate exposed surfaces and weak hardening", "correlation", "exposed-surface", + ("WSTG-INFO-06",), ("v5.0.0-13.4.5",), ("CWE-693",), ("API8:2023",), + ), +) + +_BY_KEY = {(item.module, item.category): item for item in TEST_REGISTRY} + + +def enrich_finding(finding: "Finding") -> "Finding": + definition = _BY_KEY.get((finding.module, finding.category)) + if definition: + finding.test_id = definition.id + dynamic = finding.standards + finding.standards = definition.references() + for family, references in dynamic.items(): + existing = {item.get("id") for item in finding.standards.setdefault(family, [])} + finding.standards[family].extend(item for item in references if item.get("id") not in existing) + return finding + + +def registry_document() -> dict: + return { + "schema_version": "1.0", + "tests": [ + { + "id": item.id, + "name": item.name, + "module": item.module, + "category": item.category, + "references": item.references(), + } + for item in TEST_REGISTRY + ], + }