ja4db analyzer, closes #2361

api_app/analyzers_manager/migrations/0102_analyzer_config_ja4_db.py

@@ -0,0 +1,123 @@
+from django.db import migrations
+from django.db.models.fields.related_descriptors import (
+    ForwardManyToOneDescriptor,
+    ForwardOneToOneDescriptor,
+    ManyToManyDescriptor,
+)
+
+plugin = {
+    "python_module": {
+        "health_check_schedule": None,
+        "update_schedule": {
+            "minute": "0",
+            "hour": "0",
+            "day_of_week": "*",
+            "day_of_month": "*",
+            "month_of_year": "*",
+        },
+        "module": "ja4_db.Ja4DB",
+        "base_path": "api_app.analyzers_manager.observable_analyzers",
+    },
+    "name": "JA4_DB",
+    "description": "[JA4_DB](https://ja4db.com/) lets you search a fingerprint in the public JA4 database.",
+    "disabled": False,
+    "soft_time_limit": 20,
+    "routing_key": "default",
+    "health_check_status": True,
+    "type": "observable",
+    "docker_based": False,
+    "maximum_tlp": "AMBER",
+    "observable_supported": ["generic"],
+    "supported_filetypes": [],
+    "run_hash": False,
+    "run_hash_type": "",
+    "not_supported_filetypes": [],
+    "model": "analyzers_manager.AnalyzerConfig",
+}
+
+params = []
+
+values = []
+
+
+def _get_real_obj(Model, field, value):
+    def _get_obj(Model, other_model, value):
+        if isinstance(value, dict):
+            real_vals = {}
+            for key, real_val in value.items():
+                real_vals[key] = _get_real_obj(other_model, key, real_val)
+            value = other_model.objects.get_or_create(**real_vals)[0]
+        # it is just the primary key serialized
+        else:
+            if isinstance(value, int):
+                if Model.__name__ == "PluginConfig":
+                    value = other_model.objects.get(name=plugin["name"])
+                else:
+                    value = other_model.objects.get(pk=value)
+            else:
+                value = other_model.objects.get(name=value)
+        return value
+
+    if (
+        type(getattr(Model, field))
+        in [ForwardManyToOneDescriptor, ForwardOneToOneDescriptor]
+        and value
+    ):
+        other_model = getattr(Model, field).get_queryset().model
+        value = _get_obj(Model, other_model, value)
+    elif type(getattr(Model, field)) in [ManyToManyDescriptor] and value:
+        other_model = getattr(Model, field).rel.model
+        value = [_get_obj(Model, other_model, val) for val in value]
+    return value
+
+
+def _create_object(Model, data):
+    mtm, no_mtm = {}, {}
+    for field, value in data.items():
+        value = _get_real_obj(Model, field, value)
+        if type(getattr(Model, field)) is ManyToManyDescriptor:
+            mtm[field] = value
+        else:
+            no_mtm[field] = value
+    try:
+        o = Model.objects.get(**no_mtm)
+    except Model.DoesNotExist:
+        o = Model(**no_mtm)
+        o.full_clean()
+        o.save()
+        for field, value in mtm.items():
+            attribute = getattr(o, field)
+            if value is not None:
+                attribute.set(value)
+        return False
+    return True
+
+
+def migrate(apps, schema_editor):
+    Parameter = apps.get_model("api_app", "Parameter")
+    PluginConfig = apps.get_model("api_app", "PluginConfig")
+    python_path = plugin.pop("model")
+    Model = apps.get_model(*python_path.split("."))
+    if not Model.objects.filter(name=plugin["name"]).exists():
+        exists = _create_object(Model, plugin)
+        if not exists:
+            for param in params:
+                _create_object(Parameter, param)
+            for value in values:
+                _create_object(PluginConfig, value)
+
+
+def reverse_migrate(apps, schema_editor):
+    python_path = plugin.pop("model")
+    Model = apps.get_model(*python_path.split("."))
+    Model.objects.get(name=plugin["name"]).delete()
+
+
+class Migration(migrations.Migration):
+    atomic = False
+    dependencies = [
+        ("api_app", "0062_alter_parameter_python_module"),
+        ("analyzers_manager", "0101_analyzer_config_adguard"),
+    ]
+
+    operations = [migrations.RunPython(migrate, reverse_migrate)]

api_app/analyzers_manager/observable_analyzers/ja4_db.py

@@ -0,0 +1,158 @@
+import json
+import logging
+import os
+
+import requests
+from django.conf import settings
+
+from api_app.analyzers_manager import classes
+from tests.mock_utils import MockUpResponse, if_mock_connections, patch
+
+logger = logging.getLogger(__name__)
+
+
+class Ja4DB(classes.ObservableAnalyzer):
+    """
+    We are only checking JA4 "traditional" fingerprints here
+    We should support all the JAX types as well but it is difficult
+     to add them considering that
+    it is not easy to understand the format and how to avoid
+     to run this analyzer even in cases
+    where a ja4x has not been submitted.
+    This should probably require a rework where those fingerprints
+     are saved in a table/collection
+    """
+
+    class NotJA4Exception(Exception):
+        pass
+
+    url = " https://ja4db.com/api/read/"
+
+    @classmethod
+    def location(cls) -> str:
+        db_name = "ja4_db.json"
+        return f"{settings.MEDIA_ROOT}/{db_name}"
+
+    def check_ja4_fingerprint(self, observable: str) -> str:
+        message = ""
+        try:
+            # https://github.com/FoxIO-LLC/ja4/blob/main/technical_details/README.md
+            if not observable[0] in ["t", "q"]:
+                # checks for protocol,
+                # TCP(t) and QUIC(q) are the only supported protocols
+                raise self.NotJA4Exception("only TCP and QUIC protocols are supported")
+            if not observable[1:3] in ["12", "13"]:
+                # checks for the version of the protocol
+                raise self.NotJA4Exception("procotol version wrong")
+            if not observable[3] in ["d", "i"]:
+                # SNI or no SNI
+                raise self.NotJA4Exception("SNI value not valid")
+            if not observable[4:8].isdigit():
+                # number of cipher suits and extensions
+                raise self.NotJA4Exception("cipher suite must be a number")
+            if len(observable) > 70 or len(observable) < 20:
+                raise self.NotJA4Exception("invalid length")
+            if not observable.count("_") >= 2:
+                raise self.NotJA4Exception("missing underscores")
+        except self.NotJA4Exception as e:
+            message = f"{self.observable_name} is not valid JA4 because {e}"
+            logger.info(message)
+
+        return message
+
+    @classmethod
+    def update(cls):
+        logger.info(f"Updating database from {cls.url}")
+        response = requests.get(url=cls.url)
+        response.raise_for_status()
+        data = response.json()
+        database_location = cls.location()
+
+        with open(database_location, "w", encoding="utf-8") as f:
+            json.dump(data, f)
+        logger.info(f"Database updated at {database_location}")
+
+    def run(self):
+        reason = self.check_ja4_fingerprint(self.observable_name)
+        if not reason:
+            return {"not_supported": reason}
+
+        database_location = self.location()
+        if not os.path.exists(database_location):
+            logger.info(
+                f"Database does not exist in {database_location}, initialising..."
+            )
+            self.update()
+        with open(database_location, "r") as f:
+            db = json.load(f)
+        for application in db:
+            if application["ja4_fingerprint"] == self.observable_name:
+                return application
+        return {"found": False}
+
+    @classmethod
+    def _monkeypatch(cls):
+        patches = [
+            if_mock_connections(
+                patch(
+                    "requests.get",
+                    return_value=MockUpResponse(
+                        [
+                            {
+                                "application": "Nmap",
+                                "library": None,
+                                "device": None,
+                                "os": None,
+                                "user_agent_string": None,
+                                "certificate_authority": None,
+                                "observation_count": 1,
+                                "verified": True,
+                                "notes": "",
+                                "ja4_fingerprint": None,
+                                "ja4_fingerprint_string": None,
+                                "ja4s_fingerprint": None,
+                                "ja4h_fingerprint": None,
+                                "ja4x_fingerprint": None,
+                                "ja4t_fingerprint": "1024_2_1460_00",
+                                "ja4ts_fingerprint": None,
+                                "ja4tscan_fingerprint": None,
+                            },
+                            {
+                                "application": None,
+                                "library": None,
+                                "device": None,
+                                "os": None,
+                                "user_agent_string": """Mozilla/5.0
+                                (Windows NT 10.0; Win64; x64)
+                                AppleWebKit/537.36 (KHTML, like Gecko)
+                                Chrome/125.0.0.0
+                                Safari/537.36""",
+                                "certificate_authority": None,
+                                "observation_count": 1,
+                                "verified": False,
+                                "notes": None,
+                                "ja4_fingerprint": """t13d1517h2_
+                                8daaf6152771_
+                                b0da82dd1658""",
+                                "ja4_fingerprint_string": """t13d1517h2_002f,0035,009c,
+                                009d,1301,1302,1303,c013,c014,c02b,c02c,c02f,c030,cca8,
+                                cca9_0005,000a,000b,000d,0012,0017,001b,0023,0029,002b,
+                                002d,0033,4469,fe0d,ff01_0403,0804,0401,
+                                0503,0805,0501,0806,0601""",
+                                "ja4s_fingerprint": None,
+                                "ja4h_fingerprint": """ge11cn20enus_
+                                60ca1bd65281_
+                                ac95b44401d9_
+                                8df6a44f726c""",
+                                "ja4x_fingerprint": None,
+                                "ja4t_fingerprint": None,
+                                "ja4ts_fingerprint": None,
+                                "ja4tscan_fingerprint": None,
+                            },
+                        ],
+                        200,
+                    ),
+                ),
+            )
+        ]
+        return super()._monkeypatch(patches=patches)

docs/source/Usage.md

@@ -263,8 +263,10 @@ The following is the list of the available analyzers you can run out-of-the-box.
 * `MalprobSearch`:[Malprob](https://malprob.io/) is a leading malware detection and identification service, powered by cutting-edge AI technology.
 * `OrklSearch`:[Orkl](https://orkl.eu/) is the Community Driven Cyber Threat Intelligence Library.
 * `Crt_sh`:[Crt_Sh](https://crt.sh/) lets you get certificates info about a domain.
-* `Spamhaus_WQS`:[Spamhaus_WQS](https://docs.spamhaus.com/datasets/docs/source/70-access-methods/web-query-service/000-intro.html) : The Spamhaus Web Query Service (WQS) is a method of accessing Spamhaus block lists using the HTTPS protocol. 
-* `Adguard`: [Adguard](https://github.com/AdguardTeam/AdguardSDNSFilter), a filter composed of several other filters (AdGuard Base filter, Social media filter, Tracking Protection filter, Mobile Ads filter, EasyList and EasyPrivacy) and simplified specifically to be better compatible with DNS-level ad blocking.
+* `Spamhaus_WQS`:[Spamhaus_WQS](https://docs.spamhaus.com/datasets/docs/source/70-access-methods/web-query-service/000-intro.html) The Spamhaus Web Query Service (WQS) is a method of accessing Spamhaus block lists using the HTTPS protocol. 
+* `Adguard`: [Adguard](https://github.com/AdguardTeam/AdguardSDNSFilter), a filter composed of several other filters (AdGuard Base filter, Social media filter, Tracking Protection filter, Mobile Ads filter, EasyList and EasyPrivacy) and simplified specifically to be better compatible with DNS-level ad blocking.* `Adguard`: [Adguard](https://github.com/AdguardTeam/AdguardSDNSFilter), a filter composed of several other filters (AdGuard Base filter, Social media filter, Tracking Protection filter, Mobile Ads filter, EasyList and EasyPrivacy) and simplified specifically to be better compatible with DNS-level ad blocking.
+* `JA4_DB`:[JA4_DB](https://ja4db.com/) lets you search a fingerprint in JA4 databse.
+
 ##### Generic analyzers (email, phone number, etc.; anything really)
 
 Some analyzers require details other than just IP, URL, Domain, etc. We classified them as `generic` Analyzers. Since the type of field is not known, there is a format for strings to be followed.

tests/test_crons.py

@@ -11,6 +11,7 @@
 from api_app.analyzers_manager.observable_analyzers import (
     feodo_tracker,
     greynoise_labs,
+    ja4_db,
     maxmind,
     phishing_army,
     talos,
@@ -172,7 +173,73 @@ def test_feodo_tracker_updater(self, mock_get=None):
     )
     def test_tweetfeed_updater(self, mock_get=None):
         tweetfeeds.TweetFeeds.update()
-        self.assertTrue(os.path.exists(f"{settings.MEDIA_ROOT}/tweetfeed_month.json"))
+        location, _ = tweetfeeds.TweetFeeds.location()
+        self.assertTrue(os.path.exists(location))
+
+    @if_mock_connections(
+        patch(
+            "requests.get",
+            return_value=MockUpResponse(
+                [
+                    {
+                        "application": "Nmap",
+                        "library": None,
+                        "device": None,
+                        "os": None,
+                        "user_agent_string": None,
+                        "certificate_authority": None,
+                        "observation_count": 1,
+                        "verified": True,
+                        "notes": "",
+                        "ja4_fingerprint": None,
+                        "ja4_fingerprint_string": None,
+                        "ja4s_fingerprint": None,
+                        "ja4h_fingerprint": None,
+                        "ja4x_fingerprint": None,
+                        "ja4t_fingerprint": "1024_2_1460_00",
+                        "ja4ts_fingerprint": None,
+                        "ja4tscan_fingerprint": None,
+                    },
+                    {
+                        "application": None,
+                        "library": None,
+                        "device": None,
+                        "os": None,
+                        "user_agent_string": """Mozilla/5.0
+                                (Windows NT 10.0; Win64; x64)
+                                AppleWebKit/537.36 (KHTML, like Gecko)
+                                Chrome/125.0.0.0
+                                Safari/537.36""",
+                        "certificate_authority": None,
+                        "observation_count": 1,
+                        "verified": False,
+                        "notes": None,
+                        "ja4_fingerprint": """t13d1517h2_
+                                8daaf6152771_
+                                b0da82dd1658""",
+                        "ja4_fingerprint_string": """t13d1517h2_002f,0035,009c,
+                                009d,1301,1302,1303,c013,c014,c02b,c02c,c02f,c030,cca8,
+                                cca9_0005,000a,000b,000d,0012,0017,001b,0023,0029,002b,
+                                002d,0033,4469,fe0d,ff01_0403,0804,0401,
+                                0503,0805,0501,0806,0601""",
+                        "ja4s_fingerprint": None,
+                        "ja4h_fingerprint": """ge11cn20enus_
+                                60ca1bd65281_
+                                ac95b44401d9_
+                                8df6a44f726c""",
+                        "ja4x_fingerprint": None,
+                        "ja4t_fingerprint": None,
+                        "ja4ts_fingerprint": None,
+                        "ja4tscan_fingerprint": None,
+                    },
+                ],
+                200,
+            ),
+        ),
+    )
+    def test_ja4_db_updater(self, mock_get=None):
+        ja4_db.Ja4DB.update()
+        self.assertTrue(os.path.exists(ja4_db.Ja4DB.location()))
 
     def test_quark_updater(self):
         from quark.config import DIR_PATH