feat(whoisfreaks): introduce WhoisFreaks connector and formatting updates

.gitignore

@@ -14,6 +14,8 @@ Pipfile*
 venv*
 venv_*
 .temp_venv
+.temp_venv_run
+.temp_venv_review
 black
 *.key
 *.crt

external-import/comlaude/src/comlaude/__init__.py

@@ -243,6 +243,8 @@ def _response_error(message, response):
     else:
         error_message = json.loads(response.text)
 
-    raise Exception(f"""Message:{message}.
+    raise Exception(
+        f"""Message:{message}.
             Response code returned:{response.status_code}.
-            Error message returned:{error_message}.""")
+            Error message returned:{error_message}."""
+    )

external-import/tenable-vuln-management/tests/test_converter_to_stix.py

@@ -43,7 +43,8 @@ def mock_config():
 
 @pytest.fixture
 def fake_asset():
-    return Asset.model_validate_json("""
+    return Asset.model_validate_json(
+        """
         {
         "device_type": "general-purpose",
         "fqdn": "sharepoint2016.target.example.com",
@@ -59,12 +60,14 @@ def fake_asset():
         "tracked": true,
         "last_scan_target": "192.0.0.1"
         }
-        """)
+        """
+    )
 
 
 @pytest.fixture
 def fake_plugin():
-    return Plugin.model_validate_json("""
+    return Plugin.model_validate_json(
+        """
         {
           "bid": [
             156641
@@ -190,12 +193,14 @@ def fake_plugin():
           ],
           "type": "local"
         }
-    """)
+    """
+    )
 
 
 @pytest.fixture
 def fake_vuln_finding():
-    return VulnerabilityFinding.model_validate_json("""{
+    return VulnerabilityFinding.model_validate_json(
+        """{
             "asset": {
               "device_type": "hypervisor",
               "fqdn": "vcsa8.target.example.com",
@@ -270,7 +275,8 @@ def fake_vuln_finding():
             "indexed": "2023-05-04T09:44:55.673359Z",
             "source": "NESSUS"
           }
-          """)
+          """
+    )
 
 
 def test_tlp_marking_definition_handler_should_fails_with_unsupported_TLP():

internal-enrichment/import-external-reference/src/import-external-reference.py

@@ -275,7 +275,8 @@ async def _dismiss_cookies(self, page: Page) -> bool:
         # Fallback: try to hide/remove banners/popups via JS if nothing was clicked
         if not found:
             try:
-                await page.evaluate("""
+                await page.evaluate(
+                    """
                     const keywords = ["cookie", "consent", "banner", "popup", "gdpr", "privacy", "notice", "modal", "alert", "dialog"];
                     let removed = false;
                     for (const kw of keywords) {
@@ -290,7 +291,8 @@ async def _dismiss_cookies(self, page: Page) -> bool:
                         }
                     }
                     return removed;
-                    """)
+                    """
+                )
                 self.helper.log_debug("Ran JS fallback to hide banners/popups.")
             except Exception as e:
                 self.helper.log_debug(f"JS fallback for hiding banners failed: {e}")
@@ -310,13 +312,15 @@ async def _browser_worker(self):
                         viewport={"width": 1280, "height": 800},
                         extra_http_headers=self.headers,
                     )
-                    await ctx.add_init_script("""
+                    await ctx.add_init_script(
+                        """
                         Object.defineProperty(navigator, 'webdriver', { get: () => undefined });
                         Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] });
                         Object.defineProperty(navigator, 'platform', { get: () => 'Win32' });
                         Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] });
                         window.chrome = { runtime: {}, loadTimes: () =>  { return null; }, csi: () => { return {}; } };
-                        """)
+                        """
+                    )
                     page = await ctx.new_page()
 
                     status_code = None

internal-enrichment/polyswarm-enrichment/tests/test_e2e_keydoor.py

@@ -120,7 +120,8 @@ def delete_observable(sha256: str):
 # Query helpers
 # ---------------------------------------------------------------------------
 def get_observable_by_hash(sha256: str) -> dict | None:
-    query = """
+    query = (
+        """
         {
             stixCyberObservables(
                 filters: {
@@ -141,14 +142,17 @@ def get_observable_by_hash(sha256: str) -> dict | None:
                 }
             }
         }
-    """ % sha256
+    """
+        % sha256
+    )
     result = graphql(query)
     edges = result["stixCyberObservables"]["edges"]
     return edges[0]["node"] if edges else None
 
 
 def get_notes_for_observable(observable_id: str) -> list[dict]:
-    query = """
+    query = (
+        """
         {
             notes(
                 filters: {
@@ -160,7 +164,9 @@ def get_notes_for_observable(observable_id: str) -> list[dict]:
                 edges { node { id attribute_abstract content } }
             }
         }
-    """ % observable_id
+    """
+        % observable_id
+    )
     result = graphql(query)
     return [e["node"] for e in result["notes"]["edges"]]
 

internal-enrichment/polyswarm-enrichment/tests/test_e2e_network_iocs.py

@@ -132,7 +132,8 @@ def create_observable_via_pycti(sha256: str, description: str = "") -> dict:
 # ---------------------------------------------------------------------------
 def get_observable_by_hash(sha256: str) -> dict | None:
     """Look up a StixFile observable by SHA-256 hash."""
-    query = """
+    query = (
+        """
         {
             stixCyberObservables(
                 filters: {
@@ -153,15 +154,18 @@ def get_observable_by_hash(sha256: str) -> dict | None:
                 }
             }
         }
-    """ % sha256
+    """
+        % sha256
+    )
     result = graphql(query)
     edges = result["stixCyberObservables"]["edges"]
     return edges[0]["node"] if edges else None
 
 
 def get_notes_for_observable(observable_id: str) -> list[dict]:
     """Get Notes linked to an observable."""
-    query = """
+    query = (
+        """
         {
             notes(
                 filters: {
@@ -173,7 +177,9 @@ def get_notes_for_observable(observable_id: str) -> list[dict]:
                 edges { node { id attribute_abstract content } }
             }
         }
-    """ % observable_id
+    """
+        % observable_id
+    )
     result = graphql(query)
     return [e["node"] for e in result["notes"]["edges"]]
 

internal-enrichment/polyswarm-enrichment/tests/test_e2e_opencti.py

@@ -118,7 +118,8 @@ def delete_observable(internal_id: str):
 # ---------------------------------------------------------------------------
 def get_observable_by_hash(sha256: str) -> dict | None:
     """Look up a StixFile observable by SHA-256 hash."""
-    query = """
+    query = (
+        """
         {
             stixCyberObservables(
                 filters: {
@@ -143,15 +144,18 @@ def get_observable_by_hash(sha256: str) -> dict | None:
                 }
             }
         }
-    """ % sha256
+    """
+        % sha256
+    )
     result = graphql(query)
     edges = result["stixCyberObservables"]["edges"]
     return edges[0]["node"] if edges else None
 
 
 def get_notes_for_observable(observable_id: str) -> list[dict]:
     """Get Notes linked to an observable."""
-    query = """
+    query = (
+        """
         {
             notes(
                 filters: {
@@ -169,7 +173,9 @@ def get_notes_for_observable(observable_id: str) -> list[dict]:
                 }
             }
         }
-    """ % observable_id
+    """
+        % observable_id
+    )
     result = graphql(query)
     return [e["node"] for e in result["notes"]["edges"]]
 
@@ -370,11 +376,13 @@ def feeder_enriched():
     """Wait for any feeder-created observable to be enriched."""
     deadline = time.time() + ENRICHMENT_TIMEOUT
     while time.time() < deadline:
-        result = graphql("""
+        result = graphql(
+            """
             { stixCyberObservables(first: 20, orderBy: created_at, orderMode: desc) {
                 edges { node { id observable_value x_opencti_score } }
             } }
-        """)
+        """
+        )
         for edge in result["stixCyberObservables"]["edges"]:
             node = edge["node"]
             if node["x_opencti_score"] and node["x_opencti_score"] > 50:

internal-enrichment/polyswarm-sandbox/tests/test_e2e_sandbox.py

@@ -159,7 +159,8 @@ def get_artifact_by_id(artifact_id: str) -> dict | None:
 
 
 def get_notes_for_observable(observable_id: str) -> list[dict]:
-    query = """
+    query = (
+        """
         {
             notes(
                 filters: {
@@ -177,7 +178,9 @@ def get_notes_for_observable(observable_id: str) -> list[dict]:
                 }
             }
         }
-    """ % observable_id
+    """
+        % observable_id
+    )
     result = graphql(query)
     return [e["node"] for e in result["notes"]["edges"]]
 

internal-enrichment/reversinglabs-malware-presence/src/connector/connector.py

@@ -157,15 +157,17 @@ def _generate_stix_note(self, data):
         scanner_percent = data["rl"]["malware_presence"]["scanner_percent"]
 
         abstract = "ReversingLabs Spectra Intelligence Results"
-        content = textwrap.dedent(f"""
+        content = textwrap.dedent(
+            f"""
         ## ReversingLabs Spectra Intelligence Results
         Detected hash {hash} is **{status}**
         | Scanner       |                 |
         | ------------- | --------------- |
         | Count         | {scanner_count} |
         | Match         | {scanner_match} |
         | (%)           | {scanner_percent} |
-        """)
+        """
+        )
 
         # Create Note
         stix_note = stix2.Note(

internal-enrichment/reversinglabs-spectra-analyze/src/main.py

@@ -678,7 +678,8 @@ def _ip_report(self):
         dl_files_statistics = resp_json.get("downloaded_files_statistics", {})
         abstract = "ReversingLabs Spectra Analyze IP address report"
 
-        content = textwrap.dedent(f"""
+        content = textwrap.dedent(
+            f"""
         ## ReversingLabs Spectra Analyze IP address report for {self.ip_sample}
         Third party statistics
         | Status        |  Amount         |
@@ -697,7 +698,8 @@ def _ip_report(self):
         | SUSPICIOUS    | {dl_files_statistics.get("suspicious")} |
         | UNKNOWN       | {dl_files_statistics.get("unknown")} |
         | TOTAL         | {dl_files_statistics.get("total")} |
-        """)
+        """
+        )
 
         note = stix2.Note(
             id=Note.generate_id(None, content),
@@ -842,11 +844,13 @@ def _domain_reports(self, domain_list):
         if selected_domains:
             abstract = "ReversingLabs Spectra Analyze domain statistics"
 
-            accumulated_content = textwrap.dedent("""
+            accumulated_content = textwrap.dedent(
+                """
             ## ReversingLabs Spectra Analyze domain statistics
             | Domain        |  Third party statistics Malicious/Total | Downloaded files statistics Malicious/Total |
             | ------------- | --------------- | --------------- |
-            """)
+            """
+            )
 
             for one_domain in selected_domains:
                 domain_name = one_domain.get("requested_domain")
@@ -981,11 +985,13 @@ def _url_reports(self, url_list):
         if selected_urls:
             abstract = "ReversingLabs Spectra Analyze URL statistics"
 
-            accumulated_content = textwrap.dedent("""
+            accumulated_content = textwrap.dedent(
+                """
             ## ReversingLabs Spectra Analyze URL statistics
             | URL        |  Third party statistics Malicious/Total | Analysis statistics Malicious/Total |
             | ------------- | --------------- | --------------- |
-            """)
+            """
+            )
 
             for one_url in selected_urls:
                 url_name = one_url.get("requested_url")
@@ -1107,7 +1113,8 @@ def _domain_report(self):
         )
         dl_files_statistics = resp_json.get("downloaded_files_statistics", {})
 
-        content = textwrap.dedent(f"""
+        content = textwrap.dedent(
+            f"""
         ## ReversingLabs Spectra Analyze domain report for `{self.domain_sample}`
         Third party statistics
         | Status        |  Amount         |
@@ -1126,7 +1133,8 @@ def _domain_report(self):
         | SUSPICIOUS    | {dl_files_statistics.get("suspicious")} |
         | UNKNOWN       | {dl_files_statistics.get("unknown")} |
         | TOTAL         | {dl_files_statistics.get("total")} |
-        """)
+        """
+        )
 
         note = stix2.Note(
             id=Note.generate_id(None, content),
@@ -1226,7 +1234,8 @@ def _url_report(self):
         analysis_stats = resp_json.get("analysis", {}).get("statistics", {})
         tp_stats = resp_json.get("third_party_reputations", {}).get("statistics", {})
 
-        content = textwrap.dedent(f"""
+        content = textwrap.dedent(
+            f"""
         ## ReversingLabs Spectra Analyze URL report for `{self.url_sample}`
 
         Third party statistics
@@ -1246,7 +1255,8 @@ def _url_report(self):
         | SUSPICIOUS    | {analysis_stats.get("suspicious")} |
         | UNKNOWN       | {analysis_stats.get("unknown")} |
         | TOTAL         | {analysis_stats.get("total")} |
-        """)
+        """
+        )
 
         note = stix2.Note(
             id=Note.generate_id(None, content),