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]>

Author: 3 people

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 @@ def gelocation(self):
                 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 @@ def malware_check(self):
                                 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 @@ def maltrail_check(self):
                         '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