From 4e6027d18c18ea1670d83ddcf143977776f46f87 Mon Sep 17 00:00:00 2001 From: mugiwarix Date: Mon, 3 Aug 2026 12:02:06 +0200 Subject: [PATCH] Add Gowitness JSONL report parser --- CHANGELOG/current/add_gowitness_plugin.md | 1 + .../plugins/repo/gowitness/__init__.py | 5 + .../plugins/repo/gowitness/plugin.py | 242 ++++++++++++++++++ tests/data/gowitness/gowitness_3_1_1.jsonl | 5 + tests/test_gowitness.py | 195 ++++++++++++++ 5 files changed, 448 insertions(+) create mode 100644 CHANGELOG/current/add_gowitness_plugin.md create mode 100644 faraday_plugins/plugins/repo/gowitness/__init__.py create mode 100644 faraday_plugins/plugins/repo/gowitness/plugin.py create mode 100644 tests/data/gowitness/gowitness_3_1_1.jsonl create mode 100644 tests/test_gowitness.py diff --git a/CHANGELOG/current/add_gowitness_plugin.md b/CHANGELOG/current/add_gowitness_plugin.md new file mode 100644 index 00000000..864545a7 --- /dev/null +++ b/CHANGELOG/current/add_gowitness_plugin.md @@ -0,0 +1 @@ +[ADD] Add Gowitness 3.x JSONL report plugin. diff --git a/faraday_plugins/plugins/repo/gowitness/__init__.py b/faraday_plugins/plugins/repo/gowitness/__init__.py new file mode 100644 index 00000000..3198dd4a --- /dev/null +++ b/faraday_plugins/plugins/repo/gowitness/__init__.py @@ -0,0 +1,5 @@ +""" +Faraday Penetration Test IDE +Copyright (C) 2026 Infobyte LLC (http://www.infobytesec.com/) +See the file 'doc/LICENSE' for the license information +""" diff --git a/faraday_plugins/plugins/repo/gowitness/plugin.py b/faraday_plugins/plugins/repo/gowitness/plugin.py new file mode 100644 index 00000000..67703161 --- /dev/null +++ b/faraday_plugins/plugins/repo/gowitness/plugin.py @@ -0,0 +1,242 @@ +""" +Faraday Penetration Test IDE +Copyright (C) 2026 Infobyte LLC (http://www.infobytesec.com/) +See the file 'doc/LICENSE' for the license information +""" +import json +from urllib.parse import urlsplit + +from dateutil.parser import parse + +from faraday_plugins.plugins.plugin import PluginMultiLineJsonFormat + + +class GowitnessPlugin(PluginMultiLineJsonFormat): + """Parse Gowitness 3.x reports produced by --write-jsonl.""" + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.id = "gowitness" + self.name = "Gowitness" + self.plugin_version = "1.0.0" + self.version = "3.1.1" + self.extension = [".json", ".jsonl"] + self.json_keys = { + "url", + "final_url", + "response_code", + "perception_hash", + "file_name", + } + + @staticmethod + def _parse_run_date(value): + if not value: + return None + try: + return parse(value) + except (TypeError, ValueError, OverflowError): + return None + + @staticmethod + def _url_without_credentials(value): + if not isinstance(value, str): + return value + value = value.strip() + try: + parsed_url = urlsplit(value) + hostname = parsed_url.hostname + port = parsed_url.port + except ValueError: + return "" + if not hostname or ( + parsed_url.username is None and parsed_url.password is None + ): + return value + + authority = f"[{hostname}]" if ":" in hostname else hostname + if port is not None: + authority += f":{port}" + return parsed_url._replace(netloc=authority).geturl() + + @staticmethod + def _build_response(result): + response = [] + response_code = result.get("response_code") + if response_code: + status = " ".join(filter(None, [ + str(result.get("protocol") or "HTTP"), + str(response_code), + str(result.get("response_reason") or ""), + ])) + response.append(status) + if result.get("content_length") is not None: + response.append(f"Content-Length: {result['content_length']}") + return "\n".join(response) + + @staticmethod + def _build_technical_data(result, source_url, target_url): + data = [] + if source_url and source_url != target_url: + data.append(f"Source URL: {source_url}") + data.append(f"Final URL: {target_url}") + else: + data.append(f"URL: {target_url}") + + fields = ( + ("Title", "title"), + ("Protocol", "protocol"), + ("Response code", "response_code"), + ("Response reason", "response_reason"), + ("Content length", "content_length"), + ("Screenshot file", "file_name"), + ("Perception hash", "perception_hash"), + ) + for label, key in fields: + value = result.get(key) + if value not in (None, ""): + data.append(f"{label}: {value}") + + technologies = result.get("technologies") or [] + if not isinstance(technologies, list): + technologies = [] + technologies = sorted({ + str(technology.get("value")).strip() + for technology in technologies + if isinstance(technology, dict) and technology.get("value") + }) + if technologies: + data.append(f"Technologies: {', '.join(technologies)}") + + tls = result.get("tls") or {} + tls_keys = ( + "protocol", + "key_exchange", + "cipher", + "subject_name", + "issuer", + "san_list", + "server_signature_algorithm", + "encrypted_client_hello", + ) + if isinstance(tls, dict) and any(tls.get(key) for key in tls_keys): + tls_fields = ( + ("TLS protocol", "protocol"), + ("TLS key exchange", "key_exchange"), + ("TLS cipher", "cipher"), + ("TLS subject", "subject_name"), + ("TLS issuer", "issuer"), + ("TLS valid from", "valid_from"), + ("TLS valid to", "valid_to"), + ("TLS server signature algorithm", "server_signature_algorithm"), + ) + for label, key in tls_fields: + value = tls.get(key) + if value not in (None, "", 0): + data.append(f"{label}: {value}") + + san_list = tls.get("san_list") or [] + if not isinstance(san_list, list): + san_list = [] + san_values = sorted({ + str(san.get("value")).strip() + for san in san_list + if isinstance(san, dict) and san.get("value") + }) + if san_values: + data.append(f"TLS SANs: {', '.join(san_values)}") + if tls.get("encrypted_client_hello"): + data.append("TLS encrypted client hello: true") + + return "\n".join(data) + + def parseOutputString(self, output, debug=False): + for line_number, line in enumerate(output.splitlines(), start=1): + line = line.strip() + if not line: + continue + try: + result = json.loads(line) + except (json.JSONDecodeError, TypeError) as error: + self.logger.warning( + "Skipping invalid Gowitness record on line %s: %s", + line_number, + error, + ) + continue + + if not isinstance(result, dict): + self.logger.warning( + "Skipping non-object Gowitness record on line %s", + line_number, + ) + continue + if result.get("failed"): + continue + + source_url = self._url_without_credentials(result.get("url")) + final_url = self._url_without_credentials(result.get("final_url")) + target_url = final_url or source_url + if not isinstance(target_url, str): + continue + target_url = target_url.strip() + + try: + parsed_url = urlsplit(target_url) + scheme = parsed_url.scheme.lower() + hostname = parsed_url.hostname + port = parsed_url.port + except ValueError: + continue + if scheme not in ("http", "https") or not hostname: + continue + if port is None: + port = 443 if scheme == "https" else 80 + if port < 1: + continue + + host_id = self.createAndAddHost( + name=self.resolve_hostname(hostname), + hostnames=[hostname], + ) + + negotiated_protocol = result.get("protocol") or "" + service_description = "Gowitness web service" + if negotiated_protocol: + service_description += f" using {negotiated_protocol}" + service_id = self.createAndAddServiceToHost( + host_id=host_id, + name=scheme, + protocol="tcp", + ports=port, + status="open", + version=negotiated_protocol, + description=service_description, + ) + + title = result.get("title") or "" + description = f"Gowitness captured {target_url}" + if title: + description += f' with title "{title}"' + if source_url and source_url != target_url: + description += f" after redirecting from {source_url}" + + self.createAndAddVulnWebToService( + host_id=host_id, + service_id=service_id, + name=f"Gowitness capture: {target_url}", + desc=description, + severity="info", + website=f"{scheme}://{parsed_url.netloc}", + path=parsed_url.path or "/", + query=parsed_url.query, + method="GET", + response=self._build_response(result), + status_code=result.get("response_code"), + run_date=self._parse_run_date(result.get("probed_at")), + data=self._build_technical_data(result, source_url, target_url), + ) + + +def createPlugin(*args, **kwargs): + return GowitnessPlugin(*args, **kwargs) diff --git a/tests/data/gowitness/gowitness_3_1_1.jsonl b/tests/data/gowitness/gowitness_3_1_1.jsonl new file mode 100644 index 00000000..cf49b620 --- /dev/null +++ b/tests/data/gowitness/gowitness_3_1_1.jsonl @@ -0,0 +1,5 @@ +{"id":1,"url":"https://example.test/","probed_at":"2026-07-10T12:34:56.789Z","final_url":"https://example.test/","response_code":200,"response_reason":"OK","protocol":"h2","content_length":4210,"html":"GOWITNESS_HTML_PAYLOAD_MARKER","title":"Example Home","perception_hash":"phash-home","perception_hash_group_id":1,"screenshot":"data:image/png;base64,GOWITNESS_SCREENSHOT_PAYLOAD_MARKER","file_name":"home.png","is_pdf":false,"failed":false,"failed_reason":"","tls":{"id":1,"resultid":1,"protocol":"TLS 1.3","key_exchange":"X25519","cipher":"AES_128_GCM","subject_name":"example.test","san_list":[{"id":1,"tls_id":1,"value":"example.test"},{"id":2,"tls_id":1,"value":"www.example.test"}],"issuer":"Example Test CA","valid_from":"2026-01-01T00:00:00Z","valid_to":"2027-01-01T00:00:00Z","server_signature_algorithm":2052,"encrypted_client_hello":true},"technologies":[{"id":1,"result_id":1,"value":"nginx"},{"id":2,"result_id":1,"value":"React"}],"headers":[{"id":1,"result_id":1,"key":"X-Fixture","value":"GOWITNESS_HEADER_MARKER"}],"network":[{"id":1,"result_id":1,"request_type":0,"status_code":200,"url":"https://example.test/app.js","remote_ip":"203.0.113.10","mime_type":"application/javascript","time":"2026-07-10T12:34:56.789Z","content":"R09XSVRORVNTX05FVFdPUktfTUFSS0VS","error":"GOWITNESS_NETWORK_MARKER"}],"console":[{"id":1,"result_id":1,"type":"log","value":"GOWITNESS_CONSOLE_MARKER"}],"cookies":[{"id":1,"result_id":1,"name":"session","value":"GOWITNESS_COOKIE_MARKER","domain":"example.test","path":"/","expires":"2026-07-11T12:34:56.789Z","size":24,"http_only":true,"secure":true,"session":false,"priority":"Medium","source_scheme":"Secure","source_port":443}]} +{"id":2,"url":"http://example.test/legacy","probed_at":"2026-07-10T12:35:56Z","final_url":"https://redirected.example.test/admin?view=full","response_code":200,"response_reason":"OK","protocol":"h2","content_length":1024,"html":"","title":"Administration","perception_hash":"phash-admin","perception_hash_group_id":2,"screenshot":"","file_name":"admin.png","is_pdf":false,"failed":false,"failed_reason":"","tls":{"protocol":"TLS 1.3","key_exchange":"X25519","cipher":"AES_128_GCM","subject_name":"redirected.example.test","san_list":[{"value":"redirected.example.test"}],"issuer":"Example Test CA","valid_from":"2026-01-01T00:00:00Z","valid_to":"2027-01-01T00:00:00Z","server_signature_algorithm":2052,"encrypted_client_hello":false},"technologies":[{"value":"Go"}],"headers":[],"network":[],"console":[],"cookies":[]} +{"id":3,"url":"http://example.test/health","probed_at":"2026-07-10T12:36:56Z","final_url":"http://example.test/health","response_code":204,"response_reason":"No Content","protocol":"http/1.1","content_length":0,"html":"","title":"","perception_hash":"phash-health","perception_hash_group_id":3,"screenshot":"","file_name":"health.png","is_pdf":false,"failed":false,"failed_reason":"","tls":{"protocol":"","key_exchange":"","cipher":"","subject_name":"","san_list":[],"issuer":"","valid_from":"0001-01-01T00:00:00Z","valid_to":"0001-01-01T00:00:00Z","server_signature_algorithm":0,"encrypted_client_hello":false},"technologies":[],"headers":[],"network":[],"console":[],"cookies":[]} +{"id":4,"url":"https://example.test:8443/login","probed_at":"2026-07-10T12:37:56Z","final_url":"https://example.test:8443/login","response_code":401,"response_reason":"Unauthorized","protocol":"http/1.1","content_length":512,"html":"","title":"Sign in","perception_hash":"phash-login","perception_hash_group_id":4,"screenshot":"","file_name":"login.png","is_pdf":false,"failed":false,"failed_reason":"","tls":{"protocol":"TLS 1.2","key_exchange":"ECDHE","cipher":"AES_256_GCM","subject_name":"example.test","san_list":[{"value":"example.test"}],"issuer":"Example Test CA","valid_from":"2026-01-01T00:00:00Z","valid_to":"2027-01-01T00:00:00Z","server_signature_algorithm":2052,"encrypted_client_hello":false},"technologies":[{"value":"Caddy"}],"headers":[],"network":[],"console":[],"cookies":[]} +{"id":5,"url":"https://unreachable.example.test/","probed_at":"2026-07-10T12:38:56Z","final_url":"","response_code":0,"response_reason":"","protocol":"","content_length":0,"html":"","title":"","perception_hash":"","perception_hash_group_id":0,"screenshot":"","file_name":"","is_pdf":false,"failed":true,"failed_reason":"net::ERR_NAME_NOT_RESOLVED","tls":{"protocol":"","key_exchange":"","cipher":"","subject_name":"","san_list":[],"issuer":"","valid_from":"0001-01-01T00:00:00Z","valid_to":"0001-01-01T00:00:00Z","server_signature_algorithm":0,"encrypted_client_hello":false},"technologies":[],"headers":[],"network":[],"console":[],"cookies":[]} diff --git a/tests/test_gowitness.py b/tests/test_gowitness.py new file mode 100644 index 00000000..da23930b --- /dev/null +++ b/tests/test_gowitness.py @@ -0,0 +1,195 @@ +import json +from datetime import datetime, timezone +from pathlib import Path +from unittest.mock import Mock + +from faraday_plugins.plugins.manager import PluginsManager, ReportAnalyzer +from faraday_plugins.plugins.repo.gowitness.plugin import GowitnessPlugin + + +REPORT_FILE = Path(__file__).parent / "data" / "gowitness" / "gowitness_3_1_1.jsonl" + + +def _get_host(ip, hosts): + return next(host for host in hosts if host["ip"] == ip) + + +def _get_service(port, services): + return next(service for service in services if service["port"] == port) + + +def _valid_result(url, **overrides): + result = { + "url": url, + "probed_at": "2026-07-10T12:34:56Z", + "final_url": url, + "response_code": 200, + "response_reason": "OK", + "protocol": "h2", + "content_length": 100, + "title": "Test page", + "perception_hash": "test-hash", + "file_name": "test.png", + "failed": False, + "tls": {}, + "technologies": [], + } + result.update(overrides) + return result + + +def _processed_report(): + plugin = ReportAnalyzer(PluginsManager()).get_plugin(REPORT_FILE) + plugin.resolve_hostname = Mock(return_value="203.0.113.10") + plugin.processReport(REPORT_FILE) + return plugin, json.loads(plugin.get_json()) + + +def test_report_analyzer_detects_jsonl_and_json(tmp_path): + analyzer = ReportAnalyzer(PluginsManager()) + + plugin = analyzer.get_plugin(REPORT_FILE) + assert plugin is not None + assert plugin.id == "gowitness" + assert plugin.name == "Gowitness" + assert plugin.version == "3.1.1" + assert plugin.extension == [".json", ".jsonl"] + + json_report = tmp_path / "gowitness.json" + json_report.write_text(REPORT_FILE.read_text()) + assert analyzer.get_plugin(json_report).id == "gowitness" + + +def test_process_report_maps_hosts_services_and_web_records(): + plugin, report = _processed_report() + + assert plugin.resolve_hostname.call_count == 4 + plugin.resolve_hostname.assert_any_call("redirected.example.test") + assert len(report["hosts"]) == 1 + host = _get_host("203.0.113.10", report["hosts"]) + assert set(host["hostnames"]) == { + "example.test", + "redirected.example.test", + } + + assert len(host["services"]) == 3 + services = { + (service["name"], service["protocol"], service["port"]) + for service in host["services"] + } + assert services == { + ("http", "tcp", 80), + ("https", "tcp", 443), + ("https", "tcp", 8443), + } + + https_service = _get_service(443, host["services"]) + assert https_service["version"] == "h2" + assert https_service["description"] == "Gowitness web service using h2" + assert len(https_service["vulnerabilities"]) == 2 + assert { + vulnerability["path"] + for vulnerability in https_service["vulnerabilities"] + } == {"/", "/admin"} + + redirected = next( + vulnerability + for vulnerability in https_service["vulnerabilities"] + if vulnerability["path"] == "/admin" + ) + assert redirected["type"] == "VulnerabilityWeb" + assert redirected["severity"] == "info" + assert redirected["status"] == "open" + assert redirected["website"] == "https://redirected.example.test" + assert redirected["query"] == "view=full" + assert redirected["method"] == "GET" + assert redirected["status_code"] == 200 + assert redirected["run_date"] == datetime( + 2026, 7, 10, 12, 35, 56, tzinfo=timezone.utc + ).timestamp() + assert "http://example.test/legacy" in redirected["desc"] + assert "h2 200 OK" in redirected["response"] + assert "Screenshot file: admin.png" in redirected["data"] + assert "Technologies: Go" in redirected["data"] + assert "TLS protocol: TLS 1.3" in redirected["data"] + + explicit_port = _get_service(8443, host["services"])["vulnerabilities"][0] + assert explicit_port["website"] == "https://example.test:8443" + + +def test_skips_bad_records_without_losing_later_results(): + plugin = GowitnessPlugin() + plugin.logger = Mock() + plugin.resolve_hostname = Mock(return_value="203.0.113.20") + records = [ + json.dumps(_valid_result("https://example.test/before")), + "", + "{malformed json", + json.dumps(_valid_result("https://example.test/failed", failed=True)), + json.dumps(_valid_result("ftp://example.test/not-http")), + json.dumps(_valid_result("https://example.test:invalid/bad-port")), + json.dumps(["not", "an", "object"]), + json.dumps(_valid_result("https://example.test/after")), + ] + + plugin.parseOutputString("\n".join(records)) + report = json.loads(plugin.get_json()) + vulnerabilities = [ + vulnerability + for host in report["hosts"] + for service in host["services"] + for vulnerability in service["vulnerabilities"] + ] + + assert {vulnerability["path"] for vulnerability in vulnerabilities} == { + "/before", + "/after", + } + assert plugin.logger.warning.call_count == 2 + + +def test_does_not_copy_embedded_or_sensitive_payloads(): + _, report = _processed_report() + serialized = json.dumps(report) + + excluded_markers = { + "GOWITNESS_HTML_PAYLOAD_MARKER", + "GOWITNESS_SCREENSHOT_PAYLOAD_MARKER", + "GOWITNESS_HEADER_MARKER", + "GOWITNESS_NETWORK_MARKER", + "GOWITNESS_CONSOLE_MARKER", + "GOWITNESS_COOKIE_MARKER", + } + assert not any(marker in serialized for marker in excluded_markers) + assert "Screenshot file: home.png" in serialized + + +def test_falls_back_from_blank_final_url_and_redacts_url_credentials(): + plugin = GowitnessPlugin(hostname_resolution=False) + result = _valid_result( + "https://user:secret@example.test/private", + final_url=" ", + ) + + plugin.parseOutputString(json.dumps(result)) + serialized = plugin.get_json() + + assert "user:secret" not in serialized + assert "secret" not in serialized + assert "https://example.test/private" in serialized + + +def test_plugin_options_are_forwarded_to_plugin_base(): + plugin = GowitnessPlugin( + ignore_info=True, + hostname_resolution=False, + vuln_tag="gowitness", + service_tag="web", + host_tag="recon", + ) + + assert plugin.ignore_info is True + assert plugin.hostname_resolution is False + assert plugin.vuln_tag == "gowitness" + assert plugin.service_tag == "web" + assert plugin.host_tag == "recon"