ja4db analyzer, closes #2361 #2402
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 JA4 databse.", | ||
| "disabled": False, | ||
| "soft_time_limit": 20, | ||
| "routing_key": "default", | ||
| "health_check_status": True, | ||
| "type": "observable", | ||
| "docker_based": False, | ||
| "maximum_tlp": "RED", | ||
| "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", "0099_analyzer_config_spamhaus_wqs"), | ||
| ] | ||
|
|
||
| operations = [migrations.RunPython(migrate, reverse_migrate)] | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,113 @@ | ||
| 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): | ||
| url = " https://ja4db.com/api/read/" | ||
|
|
||
| @classmethod | ||
| def location(cls) -> str: | ||
| db_name = "ja4_db.json" | ||
| return f"{settings.MEDIA_ROOT}/{db_name}" | ||
|
|
||
| @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): | ||
| database_location = self.location() | ||
| if not os.path.exists(database_location): | ||
|
Comment on lines
+80
to
+81
Member
There was a problem hiding this comment. ah one last thing. Considering that this analyzer is "generic" and we don't have a ja4 type, we could do some trick here too to avoid useless computation and, at the same time, to avoid that this analyzer fails. https://github.com/FoxIO-LLC/ja4 We could check the length at least, as a very simple filter. |
||
| 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) | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.