black 79 symbols

Author: k1r10n

misp_modules/modules/expansion/_rstcloud/__init__.py

@@ -1,4 +1,4 @@
-"""Shared RST Cloud helpers for the expansion modules (not a registered module)."""
+"""Shared RST Cloud helpers for expansion modules (not registered)."""
 
 from .client import (  # noqa: F401
     apply_to_source_attribute,

misp_modules/modules/expansion/_rstcloud/client.py

@@ -1,4 +1,4 @@
-"""rst_cs_beacon — scan a target for a Cobalt Strike beacon (GET /scan/cs-beacon)."""
+"""rst_cs_beacon — scan for Cobalt Strike beacon (GET /scan/cs-beacon)."""
 
 from __future__ import annotations
 
@@ -20,29 +20,50 @@
 
 misperrors = {"error": "Error"}
 
-_INPUTS = ["ip-dst", "ip-src", "url", "domain", "hostname",
-           "ip-dst|port", "ip-src|port", "hostname|port", "domain|port"]
+_INPUTS = [
+    "ip-dst",
+    "ip-src",
+    "url",
+    "domain",
+    "hostname",
+    "ip-dst|port",
+    "ip-src|port",
+    "hostname|port",
+    "domain|port",
+]
 # misp_standard: on a hit, return the beacon blob sha256(s) as pivotable
 # attributes tagged to the Cobalt Strike galaxy.
 mispattributes = {"input": _INPUTS, "format": "misp_standard"}
 
 moduleinfo = {
     "version": "0.2",
     "author": "RST Cloud",
-    "description": "Scan a target IP[:port] for a Cobalt Strike beacon configuration via RST Scan API.",
+    "description": (
+        "Scan a target IP[:port] for a Cobalt Strike beacon configuration"
+        " via RST Scan API."
+    ),
     "module-type": ["expansion"],
     "name": "RST Cloud Cobalt Strike Beacon",
     "requirements": ["An RST Cloud API key.", "rstapi>=1.2.0 (PyPI)."],
     "features": (
-        "Probes the target for Cobalt Strike beacon configurations via RST Scan "
-        "GET /scan/cs-beacon. On a hit, returns file MISP object(s) with pivotable "
-        "SHA-256 hashes tagged to the Cobalt Strike galaxy."
+        "Probes the target for Cobalt Strike beacon configurations via RST"
+        " Scan GET /scan/cs-beacon. On a hit, returns file MISP object(s)"
+        " with pivotable SHA-256 hashes tagged to the Cobalt Strike"
+        " galaxy."
+    ),
+    "references": [
+        "https://api.rstcloud.net/",
+        "https://pypi.org/project/rstapi/",
+    ],
+    "input": (
+        "IP, URL, domain, or hostname attribute (optional port via config)."
+    ),
+    "output": (
+        "file MISP object(s) with beacon hashes and Cobalt Strike galaxy tag."
     ),
-    "references": ["https://api.rstcloud.net/", "https://pypi.org/project/rstapi/"],
-    "input": "IP, URL, domain, or hostname attribute (optional port via config).",
-    "output": "file MISP object(s) with beacon hashes and Cobalt Strike galaxy tag.",
 }
-# 'port' (optional): port to probe when the attribute carries none (default 443).
+# 'port' (optional): port to probe when the attribute carries none
+# (default 443).
 moduleconfig = ["api_key", "base_url", "port", "timeout"]
 
 _CS_TAG = 'misp-galaxy:tool="Cobalt Strike"'
@@ -74,7 +95,10 @@ def handler(q=False):
     request = json.loads(q)
     config = request.get("config")
     if not rst_kwargs(config)["APIKEY"]:
-        return error("An RST Cloud API key is required (set api_key in the module config).")
+        return error(
+            "An RST Cloud API key is required (set api_key in the module"
+            " config)."
+        )
     target = scan_target(request, _INPUTS, config, default_port=443)
     if not target:
         return error("No target found in the request.")
@@ -83,16 +107,23 @@ def handler(q=False):
     if err:
         return error(f"RST CS beacon scan failed: {err}")
     if not isinstance(data, dict) or not data:
-        return text_result(f"{target}: no Cobalt Strike beacon found", "RST CS Beacon")
+        return text_result(
+            f"{target}: no Cobalt Strike beacon found", "RST CS Beacon"
+        )
 
     # The scanner ALWAYS returns x86/x64 probe blocks; an actual beacon is only
     # present when a block carries a parsed `config` (or a non-zero `size`). An
     # empty config / size 0 means "probed, nothing found" — NOT a detection.
     blocks = {"x86": _arch(data.get("x86")), "x64": _arch(data.get("x64"))}
-    hits = {arch: b for arch, b in blocks.items()
-            if b.get("config") or _to_int(b.get("size")) > 0}
+    hits = {
+        arch: b
+        for arch, b in blocks.items()
+        if b.get("config") or _to_int(b.get("size")) > 0
+    }
     if not hits:
-        return text_result(f"{target}: no Cobalt Strike beacon detected", "RST CS Beacon")
+        return text_result(
+            f"{target}: no Cobalt Strike beacon detected", "RST CS Beacon"
+        )
 
     from pymisp import MISPObject
 
@@ -105,8 +136,9 @@ def handler(q=False):
         if not sha or sha in seen:
             continue
         seen.add(sha)
-        # The beacon payload is a file; group its hash + config as a file object
-        # so the detection is tied to the scanned host, not a loose sha256.
+        # The beacon payload is a file; group its hash + config as a file
+        # object so the detection is tied to the scanned host, not a loose
+        # sha256.
         fobj = MISPObject("file")
         sha_attr = fobj.add_attribute("sha256", value=sha)
         sha_attr.add_tag(_CS_TAG)  # tags attach to attributes, not objects
@@ -115,8 +147,13 @@ def handler(q=False):
         if block.get("size"):
             fobj.add_attribute("size-in-bytes", value=block["size"])
         cfg = block.get("config") or {}
-        fobj.add_attribute("text", value=f"Cobalt Strike beacon ({arch}) on {target}; "
-                                          f"config: {json.dumps(cfg)[:400]}")
+        fobj.add_attribute(
+            "text",
+            value=(
+                f"Cobalt Strike beacon ({arch}) on {target}; "
+                f"config: {json.dumps(cfg)[:400]}"
+            ),
+        )
         fobj.comment = "RST CS Beacon"
         if anchor:
             fobj.add_reference(anchor, "characterizes")

misp_modules/modules/expansion/rst_cs_beacon.py

@@ -1,4 +1,4 @@
-"""rst_favicon — favicon hashes + image as a file object (GET /scan/favicon)."""
+"""rst_favicon — favicon hashes as a file object (GET /scan/favicon)."""
 
 from __future__ import annotations
 
@@ -23,27 +23,45 @@
 
 misperrors = {"error": "Error"}
 
-_INPUTS = ["url", "domain", "hostname", "ip-src", "ip-dst",
-           "ip-src|port", "ip-dst|port", "hostname|port", "domain|port"]
-# misp_standard: file object (md5/sha1/sha256 pivotable in Netlas/Censys) plus a
-# standalone favicon_hash attribute (Murmur3/MMH3, pivotable in Shodan/FOFA).
+_INPUTS = [
+    "url",
+    "domain",
+    "hostname",
+    "ip-src",
+    "ip-dst",
+    "ip-src|port",
+    "ip-dst|port",
+    "hostname|port",
+    "domain|port",
+]
+# misp_standard: file object (md5/sha1/sha256 pivotable in Netlas/Censys)
+# plus a standalone favicon_hash attribute (Murmur3/MMH3, Shodan/FOFA).
 mispattributes = {"input": _INPUTS, "format": "misp_standard"}
 
 moduleinfo = {
     "version": "0.3",
     "author": "RST Cloud",
-    "description": "Fetch a target's favicon (image + all hashes for Shodan/Netlas/Censys pivoting) via RST Scan API.",
+    "description": (
+        "Fetch a target's favicon (image + all hashes for"
+        " Shodan/Netlas/Censys pivoting) via RST Scan API."
+    ),
     "module-type": ["expansion"],
     "name": "RST Cloud Favicon",
     "requirements": ["An RST Cloud API key.", "rstapi>=1.2.0 (PyPI)."],
     "features": (
-        "Retrieves the favicon image and cryptographic hashes via RST Scan GET "
-        "/scan/favicon. Returns a file MISP object with MD5/SHA-1/SHA-256 for Censys/Netlas pivoting and a "
-        "standalone Murmur3 favicon-hash attribute for Shodan/FOFA-style pivoting."
+        "Retrieves the favicon image and cryptographic hashes via RST Scan"
+        " GET /scan/favicon. Returns a file MISP object with"
+        " MD5/SHA-1/SHA-256 for Censys/Netlas pivoting and a standalone"
+        " Murmur3 favicon-hash attribute for Shodan/FOFA-style pivoting."
     ),
-    "references": ["https://api.rstcloud.net/", "https://pypi.org/project/rstapi/"],
+    "references": [
+        "https://api.rstcloud.net/",
+        "https://pypi.org/project/rstapi/",
+    ],
     "input": "URL, domain, hostname, or IP attribute.",
-    "output": "file MISP object, favicon-hash attribute, and resolved favicon URL.",
+    "output": (
+        "file MISP object, favicon-hash attribute, and resolved favicon URL."
+    ),
 }
 moduleconfig = ["api_key", "base_url", "timeout"]
 
@@ -63,15 +81,22 @@ def handler(q=False):
     request = json.loads(q)
     config = request.get("config")
     if not rst_kwargs(config)["APIKEY"]:
-        return error("An RST Cloud API key is required (set api_key in the module config).")
+        return error(
+            "An RST Cloud API key is required (set api_key in the module"
+            " config)."
+        )
     # Favicon endpoint expects a bare host or a full URL — never host:port.
     # The API fetches the page over HTTP/HTTPS itself; adding ":443" breaks it.
     raw = value_from_request(request, _INPUTS)
     if not raw:
         return error("No target found in the request.")
     target = raw if raw.startswith(("http://", "https://")) else host_only(raw)
 
-    data, err = unwrap(rstapi.scan(**scan_kwargs(config)).GetFavicon(target, include_base64=True))
+    data, err = unwrap(
+        rstapi.scan(**scan_kwargs(config)).GetFavicon(
+            target, include_base64=True
+        )
+    )
     if err:
         return error(f"RST favicon scan failed: {err}")
     if not isinstance(data, dict) or not data.get("favicon_hash"):
@@ -84,7 +109,9 @@ def handler(q=False):
     content_type = data.get("req_content_type") or "image/x-icon"
 
     # Real filename from the resolved favicon URL (e.g. "drive_2026_32dp.ico")
-    raw_fname = req_loc.rstrip("/").split("/")[-1].split("?")[0] if req_loc else ""
+    raw_fname = (
+        req_loc.rstrip("/").split("/")[-1].split("?")[0] if req_loc else ""
+    )
     fname = raw_fname if (raw_fname and "." in raw_fname) else "favicon.ico"
 
     event, source = misp_event_with_source(request)
@@ -101,7 +128,11 @@ def handler(q=False):
 
     # Attach the raw image when the API returned base64
     try:
-        raw = base64.b64decode(data["base64_image"]) if data.get("base64_image") else None
+        raw = (
+            base64.b64decode(data["base64_image"])
+            if data.get("base64_image")
+            else None
+        )
     except Exception:
         raw = None
     if raw:
@@ -114,12 +145,18 @@ def handler(q=False):
 
     # Resolved favicon URL — where the image actually lives after redirects
     if req_loc:
-        event.add_attribute("link", req_loc, comment="RST Favicon resolved URL", to_ids=False)
-
-    # favicon_hash (Murmur3/MMH3): standalone attribute so it correlates independently
-    # across events and is searchable in Shodan/FOFA-style hunting workflows.
+        event.add_attribute(
+            "link",
+            req_loc,
+            comment="RST Favicon resolved URL",
+            to_ids=False,
+        )
+
+    # favicon_hash (Murmur3/MMH3): standalone attribute for independent
+    # correlation across events and Shodan/FOFA-style hunting workflows.
     fav_attr = event.add_attribute(
-        "other", fhash,
+        "other",
+        fhash,
         comment=f"Murmur3 favicon hash for {target} (Shodan/FOFA pivot)",
         to_ids=False,
     )