From 457d1e61f3c8dccc3351b40cf7cc34fb764ac615 Mon Sep 17 00:00:00 2001 From: Saurabh Kokate Date: Mon, 15 Dec 2025 11:02:03 +0530 Subject: [PATCH] Improve error handling and logging in MalwareDomainCheck (#2555) * Improve error handling and logging in MalwareDomainCheck --------- Co-authored-by: Saurabh Co-authored-by: Ajin Abraham --- .../views/MalwareDomainCheck.py | 116 +++++++++++++++--- 1 file changed, 101 insertions(+), 15 deletions(-) diff --git a/mobsf/MalwareAnalyzer/views/MalwareDomainCheck.py b/mobsf/MalwareAnalyzer/views/MalwareDomainCheck.py index 634db006..45608d4f 100644 --- a/mobsf/MalwareAnalyzer/views/MalwareDomainCheck.py +++ b/mobsf/MalwareAnalyzer/views/MalwareDomainCheck.py @@ -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