mirror of
https://github.com/MobSF/Mobile-Security-Framework-MobSF.git
synced 2026-10-11 22:50:09 +00:00
Improve error handling and logging in MalwareDomainCheck (#2555)
* Improve error handling and logging in MalwareDomainCheck --------- Co-authored-by: Saurabh <[email protected]> Co-authored-by: Ajin Abraham <[email protected]>
This commit is contained in:
Saurabh Kokate
co-authored by
Saurabh
Ajin Abraham
Saurabh
Ajin Abraham
parent
4b004f3817
commit
457d1e61f3
@@ -2,6 +2,7 @@
|
||||
# Module for Malware Analysis
|
||||
import io
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
from pathlib import Path
|
||||
from socket import (
|
||||
@@ -107,13 +108,28 @@ class MalwareDomainCheck:
|
||||
self.IP2Loc.close()
|
||||
|
||||
def malware_check(self):
|
||||
"""Check domains against malware database."""
|
||||
try:
|
||||
mal_db = self.malwaredomainlist
|
||||
|
||||
if not mal_db.exists():
|
||||
logger.warning('Malware domain list not found: %s', mal_db)
|
||||
return
|
||||
|
||||
if not os.access(mal_db, os.R_OK):
|
||||
logger.error('Insufficient permissions to read: %s', mal_db)
|
||||
return
|
||||
|
||||
with io.open(mal_db,
|
||||
mode='r',
|
||||
encoding='utf8',
|
||||
errors='ignore') as flip:
|
||||
entry_list = flip.readlines()
|
||||
|
||||
if not entry_list:
|
||||
logger.warning('Malware domain database is empty')
|
||||
return
|
||||
|
||||
for entry in entry_list:
|
||||
enlist = entry.split('","')
|
||||
if len(enlist) > 5:
|
||||
@@ -129,17 +145,38 @@ class MalwareDomainCheck:
|
||||
and (len(dmn_url) > 1))
|
||||
or details_dict['ip'].startswith(domain)):
|
||||
self.result[domain] = details_dict
|
||||
except FileNotFoundError:
|
||||
logger.error('Malware domain database file not found: %s', mal_db)
|
||||
except PermissionError:
|
||||
logger.error('Permission denied accessing malware database: %s', mal_db)
|
||||
except IOError:
|
||||
logger.exception('I/O error reading malware database')
|
||||
except Exception:
|
||||
logger.exception('[ERROR] Performing Malware check')
|
||||
|
||||
def maltrail_check(self):
|
||||
"""Check domains against maltrail database."""
|
||||
try:
|
||||
mal_db = self.maltrail
|
||||
|
||||
if not mal_db.exists():
|
||||
logger.warning('Maltrail database not found: %s', mal_db)
|
||||
return
|
||||
|
||||
if not os.access(mal_db, os.R_OK):
|
||||
logger.error('Insufficient permissions to read: %s', mal_db)
|
||||
return
|
||||
|
||||
with io.open(mal_db,
|
||||
mode='r',
|
||||
encoding='utf8',
|
||||
errors='ignore') as flip:
|
||||
entry_list = flip.read().splitlines()
|
||||
|
||||
if not entry_list:
|
||||
logger.warning('Maltrail database is empty')
|
||||
return
|
||||
|
||||
for domain in self.domainlist:
|
||||
if domain in entry_list:
|
||||
self.result[domain] = {
|
||||
@@ -148,36 +185,71 @@ class MalwareDomainCheck:
|
||||
'desc': 'Malicious Domain tagged by Maltrail',
|
||||
'bad': 'yes',
|
||||
}
|
||||
except FileNotFoundError:
|
||||
logger.error('Maltrail database file not found: %s', mal_db)
|
||||
except PermissionError:
|
||||
logger.error('Permission denied accessing maltrail database: %s', mal_db)
|
||||
except IOError:
|
||||
logger.exception('I/O error reading maltrail database')
|
||||
except Exception:
|
||||
logger.exception('[ERROR] Performing Maltrail Check')
|
||||
|
||||
def update(self):
|
||||
if is_internet_available():
|
||||
self.update_maltrail_db()
|
||||
else:
|
||||
logger.warning('Internet not available. '
|
||||
'Skipping Malware Database Update.')
|
||||
"""Update malware databases with connection checks."""
|
||||
try:
|
||||
if is_internet_available():
|
||||
logger.info('Checking for malware database updates')
|
||||
self.update_maltrail_db()
|
||||
else:
|
||||
logger.warning('Internet not available. '
|
||||
'Skipping Malware Database Update.')
|
||||
except Exception:
|
||||
logger.exception('Failed to update malware databases')
|
||||
|
||||
def scan(self, checksum, urls):
|
||||
"""Scan method."""
|
||||
if not settings_enabled('DOMAIN_MALWARE_SCAN'):
|
||||
logger.info('Domain Malware check disabled in settings')
|
||||
return self.result
|
||||
|
||||
msg = 'Performing Malware check on extracted domains'
|
||||
urls = [u.lower() for u in urls]
|
||||
append_scan_status(checksum, msg)
|
||||
self.domainlist = get_domains(urls)
|
||||
if self.domainlist:
|
||||
|
||||
try:
|
||||
self.domainlist = get_domains(urls)
|
||||
if not self.domainlist:
|
||||
logger.info('No domains found to analyze')
|
||||
return self.result
|
||||
|
||||
logger.info('Analyzing %d domains', len(self.domainlist))
|
||||
self.update()
|
||||
self.malware_check()
|
||||
self.maltrail_check()
|
||||
self.gelocation()
|
||||
|
||||
# Log analysis results
|
||||
malicious_count = sum(1 for domain in self.result.values()
|
||||
if domain.get('bad') == 'yes')
|
||||
logger.info(
|
||||
'Malware analysis completed: %d malicious domains found',
|
||||
malicious_count)
|
||||
|
||||
except Exception as exp:
|
||||
logger.exception('Critical error during domain malware check')
|
||||
append_scan_status(checksum, 'Domain malware check failed', repr(exp))
|
||||
|
||||
return self.result
|
||||
|
||||
|
||||
# Helper Functions
|
||||
|
||||
def verify_domain(checkeddom):
|
||||
"""Verify domain."""
|
||||
try:
|
||||
if not checkeddom:
|
||||
return False
|
||||
|
||||
if (len(checkeddom) > 2
|
||||
and '.' in checkeddom
|
||||
and (checkeddom.endswith('.') is False
|
||||
@@ -187,6 +259,7 @@ def verify_domain(checkeddom):
|
||||
return False
|
||||
except Exception:
|
||||
logger.exception('[ERROR] Verifying Domain')
|
||||
return False
|
||||
|
||||
|
||||
def get_netloc(url):
|
||||
@@ -202,24 +275,37 @@ def get_netloc(url):
|
||||
return domain
|
||||
except Exception:
|
||||
logger.exception('[ERROR] Extracting Domain form URL')
|
||||
return None
|
||||
|
||||
|
||||
def sanitize_domain(domain):
|
||||
"""Sanitize domain to be RFC1034 compliant."""
|
||||
domain = domain.split('_')[0]
|
||||
domain = re.sub(r'[^\w^\.^\-]', '', domain)
|
||||
if domain.startswith('-'):
|
||||
domain = sanitize_domain(domain[1:])
|
||||
elif domain.endswith('-'):
|
||||
domain = sanitize_domain(domain[:-1])
|
||||
try:
|
||||
if not domain:
|
||||
return domain
|
||||
|
||||
domain = domain.split('_')[0]
|
||||
domain = re.sub(r'[^\w^\.^\-]', '', domain)
|
||||
if domain.startswith('-'):
|
||||
domain = sanitize_domain(domain[1:])
|
||||
elif domain.endswith('-'):
|
||||
domain = sanitize_domain(domain[:-1])
|
||||
except Exception:
|
||||
logger.exception('[ERROR] Sanitizing domain')
|
||||
return domain
|
||||
|
||||
|
||||
def get_domains(urls):
|
||||
"""Get Domains."""
|
||||
domains = set()
|
||||
try:
|
||||
domains = set()
|
||||
if not urls:
|
||||
return domains
|
||||
|
||||
for url in urls:
|
||||
if not url:
|
||||
continue
|
||||
|
||||
parse_uri = urlparse(url)
|
||||
if not parse_uri.scheme:
|
||||
url = '//' + url
|
||||
@@ -228,6 +314,6 @@ def get_domains(urls):
|
||||
'{uri.hostname}'.format(uri=parse_uri))
|
||||
if verify_domain(domain):
|
||||
domains.add(domain)
|
||||
return domains
|
||||
except Exception:
|
||||
logger.exception('[ERROR] Extracting Domain form URL')
|
||||
return domains
|
||||
|
||||
Reference in New Issue
Block a user