-
-
Notifications
You must be signed in to change notification settings - Fork 426
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* ailtyposquatting * restore a file that was deleted * fix * fix * changes * tests * no files * logs * files * variables * test * test * enum * tests * tests * dns_resolve * migration * a log :p --------- Co-authored-by: g4ze <[email protected]>
- Loading branch information
Showing
4 changed files
with
220 additions
and
1 deletion.
There are no files selected for viewing
151 changes: 151 additions & 0 deletions
151
api_app/analyzers_manager/migrations/0093_analyzer_config_ailtyposquatting.py
This file contains 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,151 @@ | ||
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": None, | ||
"module": "ailtyposquatting.AilTypoSquatting", | ||
"base_path": "api_app.analyzers_manager.observable_analyzers", | ||
}, | ||
"name": "AILTypoSquatting", | ||
"description": "[AILTypoSquatting](https://github.com/typosquatter/ail-typo-squatting) is a Python library to generate list of potential typo squatting domains with domain name permutation engine to feed AIL and other systems.", | ||
"disabled": False, | ||
"soft_time_limit": 60, | ||
"routing_key": "default", | ||
"health_check_status": True, | ||
"type": "observable", | ||
"docker_based": False, | ||
"maximum_tlp": "RED", | ||
"observable_supported": ["domain"], | ||
"supported_filetypes": [], | ||
"run_hash": False, | ||
"run_hash_type": "", | ||
"not_supported_filetypes": [], | ||
"model": "analyzers_manager.AnalyzerConfig", | ||
} | ||
|
||
params = [ | ||
{ | ||
"python_module": { | ||
"module": "ailtyposquatting.AilTypoSquatting", | ||
"base_path": "api_app.analyzers_manager.observable_analyzers", | ||
}, | ||
"name": "dns_resolving", | ||
"type": "bool", | ||
"description": "dns_resolving for AilTypoSquatting; only works for TLP CLEAR", | ||
"is_secret": False, | ||
"required": False, | ||
}, | ||
] | ||
values = [ | ||
{ | ||
"parameter": { | ||
"python_module": { | ||
"module": "ailtyposquatting.AilTypoSquatting", | ||
"base_path": "api_app.analyzers_manager.observable_analyzers", | ||
}, | ||
"name": "dns_resolving", | ||
"type": "bool", | ||
"description": "dns_resolving for AilTypoSquatting; only works for TLP CLEAR", | ||
"is_secret": False, | ||
"required": False, | ||
}, | ||
"analyzer_config": "AILTypoSquatting", | ||
"connector_config": None, | ||
"visualizer_config": None, | ||
"ingestor_config": None, | ||
"pivot_config": None, | ||
"for_organization": False, | ||
"value": False, | ||
"updated_at": "2024-05-26T00:10:15.236358Z", | ||
"owner": None, | ||
}, | ||
] | ||
|
||
|
||
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", "0092_alter_validin_desc"), | ||
] | ||
|
||
operations = [migrations.RunPython(migrate, reverse_migrate)] |
67 changes: 67 additions & 0 deletions
67
api_app/analyzers_manager/observable_analyzers/ailtyposquatting.py
This file contains 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,67 @@ | ||
import logging | ||
import math | ||
|
||
from ail_typo_squatting import typo | ||
from ail_typo_squatting.dns_local import resolving | ||
|
||
from api_app.analyzers_manager import classes | ||
from tests.mock_utils import if_mock_connections, patch | ||
|
||
logger = logging.getLogger(__name__) | ||
|
||
|
||
class AilTypoSquatting(classes.ObservableAnalyzer): | ||
""" | ||
wrapper for https://github.com/typosquatter/ail-typo-squatting | ||
""" | ||
|
||
dns_resolving: bool = False | ||
|
||
def update(self) -> bool: | ||
pass | ||
|
||
def run(self): | ||
response = {} | ||
logger.info( | ||
f"""running AilTypoSquatting on {self.observable_name} | ||
with tlp {self._job.tlp} | ||
and dns resolving {self.dns_resolving}""" | ||
) | ||
|
||
response["algorithms"] = typo.runAll( | ||
domain=self.observable_name, | ||
limit=math.inf, | ||
formatoutput="text", | ||
pathOutput=None, | ||
) | ||
if self._job.tlp == self._job.TLP.CLEAR.value and self.dns_resolving: | ||
# for "x.com", response["algorithms"][0]=".com" | ||
# which is not valid for look up | ||
if len(self.observable_name.split(".")[0]) == 1: | ||
logger.info( | ||
f"""running dns resolving on {self.observable_name} | ||
excluding {response['algorithms'][0]}""" | ||
) | ||
response["dnsResolving"] = resolving.dnsResolving( | ||
resultList=response["algorithms"][1:], | ||
domain=self.observable_name, | ||
pathOutput=None, | ||
) | ||
else: | ||
response["dnsResolving"] = resolving.dnsResolving( | ||
resultList=response["algorithms"], | ||
domain=self.observable_name, | ||
pathOutput=None, | ||
) | ||
|
||
return response | ||
|
||
@classmethod | ||
def _monkeypatch(cls): | ||
patches = [ | ||
if_mock_connections( | ||
patch.object(typo, "runAll", return_value=None), | ||
patch.object(resolving, "dnsResolving", return_value=None), | ||
) | ||
] | ||
return super()._monkeypatch(patches=patches) |
This file contains 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 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