Ssl ja3 c2 (#1068)

Co-authored-by: Thomas Chopitea <tomchop@gmail.com>

.vscode/settings.json

@@ -10,6 +10,9 @@
     "python.testing.unittestEnabled": true,
     "python.analysis.typeCheckingMode": "basic",
     "editor.defaultFormatter": "charliermarsh.ruff",
-    "mypy-type-checker.importStrategy": "fromEnvironment"
+    "mypy-type-checker.importStrategy": "fromEnvironment",
+    "[python]": {
+        "editor.defaultFormatter": "charliermarsh.ruff"
+    }
 
 }

core/schemas/observable.py

@@ -29,6 +29,8 @@ class ObservableType(str, Enum):
     imphash = "imphash"
     ipv4 = "ipv4"
     ipv6 = "ipv6"
+    ja3 = "ja3"
+    jarm = "jarm"
     mac_address = "mac_address"
     md5 = "md5"
     generic = "generic"
@@ -203,6 +205,8 @@ def find_type(value: str) -> ObservableType | None:
     imphash,  # noqa: F401
     ipv4,  # noqa: F401
     ipv6,  # noqa: F401
+    ja3,  # noqa: F401
+    jarm,  # noqa: F401
     mac_address,  # noqa: F401
     md5,  # noqa: F401
     path,  # noqa: F401

core/schemas/observables/ja3.py

@@ -0,0 +1,10 @@
+from typing import Literal
+
+from core.schemas import observable
+
+
+class JA3(observable.Observable):
+    type: Literal[observable.ObservableType.ja3] = observable.ObservableType.ja3
+
+
+observable.TYPE_MAPPING[observable.ObservableType.ja3] = JA3

core/schemas/observables/jarm.py

@@ -0,0 +1,10 @@
+from typing import Literal
+
+from core.schemas import observable
+
+
+class JARM(observable.Observable):
+    type: Literal[observable.ObservableType.jarm] = observable.ObservableType.jarm
+
+
+observable.TYPE_MAPPING[observable.ObservableType.ja3] = JARM

plugins/feeds/public/sslblacklist_ja3.py

@@ -0,0 +1,57 @@
+from datetime import timedelta
+from io import StringIO
+from typing import ClassVar
+
+import pandas as pd
+
+from core import taskmanager
+from core.schemas import task
+from core.schemas.observables import ja3
+
+
+class SSLBlacklistJA3(task.FeedTask):
+    _defaults = {
+        "frequency": timedelta(hours=1),
+        "name": "SSLBlacklistJA3",
+        "description": "This feed contains JA3 hashes of SSL by AbuseCH",
+    }
+
+    _SOURCE: ClassVar["str"] = "https://sslbl.abuse.ch/blacklist/ja3_fingerprints.csv"
+    _NAMES = ["ja3", "first_seen", "last_seen", "threat"]
+
+    def run(self):
+        response = self._make_request(self._SOURCE, auth=None, verify=True)
+        if response:
+            data = StringIO(response.text)
+
+            df = pd.read_csv(
+                data,
+                delimiter=",",
+                comment="#",
+                names=self._NAMES,
+                parse_dates=["first_seen"],
+            )
+            df = self._filter_observables_by_time(df, "last_seen")
+            df = df.fillna("")
+            for _, row in df.iterrows():
+                self.analyze(row)
+
+    def analyze(self, row):
+        ja3_md5 = row["ja3_md5"]
+        first_seen = row["first_seen"]
+        last_seen = row["last_seen"]
+        threat = row["threat"]
+
+        ja3_obs = ja3.JA3(value=ja3_md5).save()
+
+        context = {}
+        context["first_seen"] = first_seen
+        context["last_seen"] = last_seen
+
+        ja3_obs.add_context(self.name, context)
+
+        if threat:
+            ja3_obs.tag([threat])
+
+
+taskmanager.TaskManager.register_task(SSLBlacklistJA3)

tests/feeds.py

@@ -10,6 +10,7 @@
     hybrid_analysis,
     lolbas,
     openphish,
+    sslblacklist_ja3,
     timesketch,
     tor_exit_nodes,
     tweetlive,
@@ -76,3 +77,8 @@ def test_tweetlive(self):
         defaults = tweetlive.TweetLive._defaults.copy()
         feed = tweetlive.TweetLive(**defaults)
         feed.run()
+
+    def test_sslblacklist_ja3(self):
+        defaults = sslblacklist_ja3.SSLBlacklistJA3._defaults.copy()
+        feed = sslblacklist_ja3.SSLBlacklistJA3(**defaults)
+        feed.run()

tests/schemas/observable.py

@@ -19,6 +19,7 @@
     imphash,
     ipv4,
     ipv6,
+    ja3,
     mac_address,
     md5,
     mutex,
@@ -364,6 +365,13 @@ def test_create_ipv6(self) -> None:
         self.assertEqual(observable.value, "::1")
         self.assertIsInstance(observable, ipv6.IPv6)
 
+    def test_create_ja3(self) -> None:
+        """Tests creating a JA3."""
+        observable = ja3.JA3(value="1234567890").save()
+        self.assertIsNotNone(observable.id)
+        self.assertEqual(observable.value, "1234567890")
+        self.assertEqual(observable.type, "ja3")
+
     def test_create_mac_address(self) -> None:
         """Tests creating a MAC address."""
         observable = mac_address.MacAddress(value="00:00:00:00:00:00").save()