Add Gowitness JSONL report parser

CHANGELOG/current/add_gowitness_plugin.md

@@ -0,0 +1 @@
+[ADD] Add Gowitness 3.x JSONL report plugin.

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
+"""

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)

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":"<html>GOWITNESS_HTML_PAYLOAD_MARKER</html>","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":[]}