mirror of
https://github.com/MobSF/Mobile-Security-Framework-MobSF.git
synced 2026-10-11 22:50:09 +00:00
7a4785fb3b
MALWARE_DB_URL pointed at malwaredomainlist.com, a service that has been defunct for years, and nothing in the codebase ever fetched or refreshed it - the only "data" behind malware_check() was a static snapshot bundled in the repo whose newest entries date back to 2009. That check ran on every scan against 16+ year old data without anyone noticing, while maltrail_check() (which does self-update from a live source) was the only domain check actually doing real work. Removes malware_check(), its now-unused get_netloc() helper, the stale bundled data file, and the unused MALWARE_DB_URL setting. maltrail_check() is unaffected and remains the sole domain check. Co-authored-by: alanhasn Co-authored-by: Ajin Abraham <[email protected]>
252 lines
8.4 KiB
Python
252 lines
8.4 KiB
Python
# -*- coding: utf_8 -*-
|
|
# Module for Malware Analysis
|
|
import io
|
|
import logging
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
from django.conf import settings
|
|
|
|
import IP2Location
|
|
|
|
from mobsf.MobSF.security import resolve_public_ip
|
|
from mobsf.MobSF.utils import (
|
|
append_scan_status,
|
|
is_internet_available,
|
|
settings_enabled,
|
|
update_local_db,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class MalwareDomainCheck:
|
|
|
|
def __init__(self):
|
|
self.sig_dir = Path(settings.SIGNATURE_DIR)
|
|
self.maltrail = self.sig_dir / 'maltrail-malware-domains.txt'
|
|
self.iplocbin = self.sig_dir / 'IP2LOCATION-LITE-DB5.IPV6.BIN'
|
|
self.result = {}
|
|
self.domainlist = None
|
|
self.IP2Loc = IP2Location.IP2Location()
|
|
|
|
def update_maltrail_db(self):
|
|
"""Check for update in maltrail DB."""
|
|
try:
|
|
mal_db = self.maltrail
|
|
resp = update_local_db(
|
|
'Maltrail',
|
|
settings.MALTRAIL_DB_URL,
|
|
mal_db,
|
|
timeout=120)
|
|
if not resp:
|
|
return
|
|
# DB needs update
|
|
# Check2: DB Syntax Changed
|
|
lines = resp.decode('utf-8', 'ignore').splitlines()
|
|
if len(lines) > 100:
|
|
logger.info('Updating Maltrail Database')
|
|
with open(mal_db, 'wb') as wfp:
|
|
wfp.write(resp)
|
|
else:
|
|
logger.warning('Unable to Update Maltrail DB')
|
|
except Exception:
|
|
logger.exception('[ERROR] Maltrail DB Update')
|
|
|
|
def gelocation(self):
|
|
"""Perform Geolocation."""
|
|
try:
|
|
ofac_list = {
|
|
'cuba', 'iran', 'north korea',
|
|
'russia', 'syria', 'balkans',
|
|
'belarus', 'myanmar', 'congo',
|
|
'ethiopia', 'hong kong', 'iraq',
|
|
'lebanon', 'libya', 'sudan',
|
|
'venezuela', 'yemen', 'zimbabwe',
|
|
'crimea', 'donetsk', 'luhansk',
|
|
'afghanistan', 'china', 'ivory coast',
|
|
'cyprus', 'eritrea', 'haiti',
|
|
'liberia', 'somalia', 'sri lanka',
|
|
'vietnam', 'south sudan',
|
|
}
|
|
self.IP2Loc.open(self.iplocbin)
|
|
for domain in self.domainlist:
|
|
# Tag Good Domains
|
|
if domain not in self.result:
|
|
tmp_d = {}
|
|
tmp_d['bad'] = 'no'
|
|
self.result[domain] = tmp_d
|
|
# GeoIP
|
|
# Never pass internal DNS results into subsequent analysis.
|
|
# This domain originates in an untrusted application.
|
|
ip = resolve_public_ip(domain)
|
|
if ip:
|
|
rec = self.IP2Loc.get_all(ip)
|
|
self.result[domain]['geolocation'] = rec.__dict__
|
|
country = rec.__dict__.get('country_long')
|
|
region = rec.__dict__.get('region')
|
|
city = rec.__dict__.get('city')
|
|
self.result[domain]['ofac'] = False
|
|
if country and country.lower() in ofac_list:
|
|
self.result[domain]['ofac'] = True
|
|
elif region and region.lower() in ofac_list:
|
|
self.result[domain]['ofac'] = True
|
|
elif city and city.lower() in ofac_list:
|
|
self.result[domain]['ofac'] = True
|
|
else:
|
|
self.result[domain]['geolocation'] = None
|
|
except Exception:
|
|
logger.exception('Failed to Perform Geolocation')
|
|
finally:
|
|
if self.IP2Loc:
|
|
self.IP2Loc.close()
|
|
|
|
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] = {
|
|
'domain_or_url': domain,
|
|
'ip': 'N/A',
|
|
'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):
|
|
"""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)
|
|
|
|
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.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
|
|
and re.search('[a-zA-Z0-9]', checkeddom))):
|
|
return True
|
|
else:
|
|
return False
|
|
except Exception:
|
|
logger.exception('[ERROR] Verifying Domain')
|
|
return False
|
|
|
|
|
|
def sanitize_domain(domain):
|
|
"""Sanitize domain to be RFC1034 compliant."""
|
|
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:
|
|
if not urls:
|
|
return domains
|
|
|
|
for url in urls:
|
|
if not url:
|
|
continue
|
|
|
|
parse_uri = urlparse(url)
|
|
if not parse_uri.scheme:
|
|
url = '//' + url
|
|
parse_uri = urlparse(url)
|
|
domain = sanitize_domain(
|
|
'{uri.hostname}'.format(uri=parse_uri))
|
|
if verify_domain(domain):
|
|
domains.add(domain)
|
|
except Exception:
|
|
logger.exception('[ERROR] Extracting Domain form URL')
|
|
return domains
|