HOTFIX: Maltrail Malware DB + Bug Fix #1606 (#1634)

Added support for Maltrail Malware DB
Fixes a bug in libsast #1606

DynamicAnalyzer/views/android/analysis.py

@@ -14,7 +14,7 @@
 
 from StaticAnalyzer.models import StaticAnalyzerAndroid
 
-from MalwareAnalyzer.views.domain_check import malware_check
+from MalwareAnalyzer.views.MalwareDomainCheck import MalwareDomainCheck
 
 logger = logging.getLogger(__name__)
 
@@ -48,7 +48,7 @@ def run_analysis(apk_dir, md5_hash, package):
         urls = []
     # Domain Extraction and Malware Check
     logger.info('Performing Malware Check on extracted Domains')
-    domains = malware_check(urls)
+    domains = MalwareDomainCheck().scan(urls)
 
     # Email Etraction Regex
     emails = []

LICENSES/maltrail_blacklist.txt

@@ -0,0 +1,21 @@
+The MIT License (MIT)
+
+Copyright (c) 2014-2020 Maltrail developers (https://github.com/stamparm/maltrail/)
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

MalwareAnalyzer/views/MalwareDomainCheck.py

@@ -0,0 +1,227 @@
+# -*- coding: utf_8 -*-
+# Module for Malware Analysis
+import io
+import logging
+import re
+from pathlib import Path
+from socket import (
+    gaierror,
+    gethostbyname,
+)
+from urllib.parse import urlparse
+
+from django.conf import settings
+
+import IP2Location
+
+from MobSF.utils import (
+    is_internet_available,
+    update_local_db,
+)
+
+logger = logging.getLogger(__name__)
+IP2Loc = IP2Location.IP2Location()
+
+
+class MalwareDomainCheck:
+
+    def __init__(self):
+        self.sig_dir = Path(settings.SIGNATURE_DIR)
+        self.malwaredomainlist = self.sig_dir / 'malwaredomainlist'
+        self.maltrail = self.sig_dir / 'maltrail-malware-domains.txt'
+        self.iplocbin = self.sig_dir / 'IP2LOCATION-LITE-DB5.IPV6.BIN'
+        self.result = {}
+        self.domainlist = None
+
+    def update_malware_db(self):
+        """Check for update in malware DB."""
+        try:
+            mal_db = self.malwaredomainlist
+            resp = update_local_db('Malware', settings.MALWARE_DB_URL, mal_db)
+            if not resp:
+                return
+            # DB needs update
+            # Check2: DB Syntax Changed
+            line = resp.decode('utf-8', 'ignore').split('\n')[0]
+            lst = line.split('",')
+            if len(lst) == 10:
+                # DB Format is not changed. Let's update DB
+                logger.info('Updating Malware Database')
+                with open(mal_db, 'wb') as wfp:
+                    wfp.write(resp)
+            else:
+                logger.warning('Unable to Update Malware DB')
+        except Exception:
+            logger.exception('[ERROR] Malware DB Update')
+
+    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)
+            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:
+            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
+                ip = None
+                try:
+                    ip = gethostbyname(domain)
+                except (gaierror, UnicodeError):
+                    pass
+                if ip:
+                    rec = IP2Loc.get_all(ip)
+                    self.result[domain]['geolocation'] = rec.__dict__
+                else:
+                    self.result[domain]['geolocation'] = None
+        except Exception:
+            logger.exception('Failed to Perform Geolocation')
+        finally:
+            if IP2Loc:
+                IP2Loc.close()
+
+    def malware_check(self):
+        try:
+            mal_db = self.malwaredomainlist
+            with io.open(mal_db,
+                         mode='r',
+                         encoding='utf8',
+                         errors='ignore') as flip:
+                entry_list = flip.readlines()
+            for entry in entry_list:
+                enlist = entry.split('","')
+                if len(enlist) > 5:
+                    details_dict = {}
+                    details_dict['domain_or_url'] = enlist[1]
+                    details_dict['ip'] = enlist[2]
+                    details_dict['desc'] = enlist[4]
+                    details_dict['bad'] = 'yes'
+                    dmn_url = details_dict['domain_or_url']
+                    for domain in self.domainlist:
+                        dmn_neturl = get_netloc(dmn_url)
+                        if (((dmn_neturl == domain or dmn_neturl == domain[4:])
+                                and (len(dmn_url) > 1))
+                                or details_dict['ip'].startswith(domain)):
+                            self.result[domain] = details_dict
+        except Exception:
+            logger.exception('[ERROR] Performing Malware Check')
+
+    def maltrail_check(self):
+        try:
+            mal_db = self.maltrail
+            with io.open(mal_db,
+                         mode='r',
+                         encoding='utf8',
+                         errors='ignore') as flip:
+                entry_list = flip.read().splitlines()
+            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 Exception:
+            logger.exception('[ERROR] Performing Maltrail Check')
+
+    def update(self):
+        if is_internet_available():
+            self.update_malware_db()
+            self.update_maltrail_db()
+        else:
+            logger.warning('Internet not available. '
+                           'Skipping Malware Database Update.')
+
+    def scan(self, urls):
+        if not settings.DOMAIN_MALWARE_SCAN:
+            logger.info('Domain Malware Check disabled in settings')
+            return self.result
+        self.domainlist = get_domains(urls)
+        if self.domainlist:
+            self.update()
+            self.malware_check()
+            self.maltrail_check()
+            self.gelocation()
+        return self.result
+
+
+# Helper Functions
+
+def verify_domain(checkeddom):
+    try:
+        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')
+
+
+def get_netloc(url):
+    """Get Domain."""
+    try:
+        domain = ''
+        parse_uri = urlparse(url)
+        if not parse_uri.scheme:
+            url = '//' + url
+            parse_uri = urlparse(url)
+        domain = '{uri.netloc}'.format(uri=parse_uri)
+        if verify_domain(domain):
+            return domain
+    except Exception:
+        logger.exception('[ERROR] Extracting Domain form URL')
+
+
+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])
+    return domain
+
+
+def get_domains(urls):
+    """Get Domains."""
+    try:
+        domains = set()
+        for url in urls:
+            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)
+        return domains
+    except Exception:
+        logger.exception('[ERROR] Extracting Domain form URL')