Improve SSRF checks, strict path check for well_known_path

Author: ajinabraham

mobsf/MobSF/init.py

@@ -18,7 +18,7 @@
 
 logger = logging.getLogger(__name__)
 
-VERSION = '4.3.1'
+VERSION = '4.3.2'
 BANNER = r"""
   __  __       _    ____  _____       _  _    _____ 
  |  \/  | ___ | |__/ ___||  ___|_   _| || |  |___ / 

mobsf/MobSF/security.py

@@ -3,10 +3,13 @@
 import functools
 import logging
 import re
+import socket
+import ipaddress
 import sys
 from shutil import which
 from pathlib import Path
 from platform import system
+from urllib.parse import urlparse
 from concurrent.futures import ThreadPoolExecutor
 
 from mobsf.MobSF.utils import (
@@ -249,3 +252,62 @@ def sanitize_for_logging(filename: str, max_length: int = 255) -> str:
 
     # Truncate filename to the maximum allowed length
     return filename[:max_length]
+
+
+def valid_host(host):
+    """Check if host is valid, run SSRF checks."""
+    try:
+        if len(host) > 2083:  # Standard URL length limit
+            return False
+
+        prefixs = ('http://', 'https://')
+        if not host.startswith(prefixs):
+            host = f'http://{host}'
+        parsed = urlparse(host)
+        domain = parsed.netloc
+        hostname = parsed.hostname
+        path = parsed.path
+        query = parsed.query
+        params = parsed.params
+
+        # Check for hostname
+        if not hostname:
+            return False
+
+        # Check for URL credentials
+        if '@' in domain:
+            return False
+
+        # Detect parser escapes, only host is allowed
+        if len(path) > 0 or len(query) > 0 or len(params) > 0:
+            return False
+
+        # Resolve dns to get ipv4 or ipv6 address
+        ip_addresses = socket.getaddrinfo(hostname, None, timeout=5)
+        for ip in ip_addresses:
+            ip_obj = ipaddress.ip_address(ip[4][0])
+            if (ip_obj.is_private
+                or ip_obj.is_loopback
+                or ip_obj.is_link_local
+                or ip_obj.is_reserved
+                or ip_obj.is_multicast
+                    or ip_obj.is_unspecified):
+                return False
+
+            # Additional checks for specific IPv4 ranges
+            if isinstance(ip_obj, ipaddress.IPv4Address):
+                problematic_networks = [
+                    '127.0.0.0/8',
+                    '169.254.0.0/16',
+                    '172.16.0.0/12',
+                    '192.168.0.0/16',
+                    '10.0.0.0/8',
+                ]
+                for network in problematic_networks:
+                    if ip_obj in ipaddress.IPv4Network(network):
+                        return False
+
+        # If all checks pass, return True
+        return True
+    except Exception:
+        return False

mobsf/MobSF/utils.py

@@ -16,11 +16,9 @@
 import string
 import subprocess
 import stat
-import socket
 import sqlite3
 import unicodedata
 import threading
-from urllib.parse import urlparse
 from pathlib import Path
 from concurrent.futures import (
     ThreadPoolExecutor,
@@ -904,58 +902,6 @@ def id_generator(size=6, chars=string.ascii_uppercase + string.digits):
     return ''.join(random.choice(chars) for _ in range(size))
 
 
-def valid_host(host):
-    """Check if host is valid."""
-    try:
-        prefixs = ('http://', 'https://')
-        if not host.startswith(prefixs):
-            host = f'http://{host}'
-        parsed = urlparse(host)
-        domain = parsed.netloc
-        path = parsed.path
-        if len(domain) == 0:
-            # No valid domain
-            return False
-        if len(path) > 0:
-            # Only host is allowed
-            return False
-        if ':' in domain:
-            # IPv6
-            return False
-        # Local network
-        invalid_prefix = (
-            '100.64.',
-            '127.',
-            '192.',
-            '198.',
-            '10.',
-            '172.',
-            '169.',
-            '0.',
-            '203.0.',
-            '224.0.',
-            '240.0',
-            '255.255.',
-            'localhost',
-            '::1',
-            '64::ff9b::',
-            '100::',
-            '2001::',
-            '2002::',
-            'fc00::',
-            'fe80::',
-            'ff00::')
-        if domain.startswith(invalid_prefix):
-            return False
-        ip = socket.gethostbyname(domain)
-        if ip.startswith(invalid_prefix):
-            # Resolve dns to get IP
-            return False
-        return True
-    except Exception:
-        return False
-
-
 def append_scan_status(checksum, status, exception=None):
     """Append Scan Status to Database."""
     try:

mobsf/StaticAnalyzer/views/android/manifest_analysis.py

@@ -2,6 +2,7 @@
 # flake8: noqa
 """Module for android manifest analysis."""
 import logging
+from urllib.parse import urlparse
 
 import requests
 from concurrent.futures import ThreadPoolExecutor
@@ -10,8 +11,8 @@
     append_scan_status,
     is_number,
     upstream_proxy,
-    valid_host,
 )
+from mobsf.MobSF.security import valid_host
 from mobsf.StaticAnalyzer.views.android import (
     network_security,
 )
@@ -27,6 +28,8 @@
 ANDROID_9_0_LEVEL = 28
 ANDROID_10_0_LEVEL = 29
 ANDROID_MANIFEST_FILE = 'AndroidManifest.xml'
+WELL_KNOWN_PATH = '/.well-known/assetlinks.json'
+
 ANDROID_API_LEVEL_MAP = {
     '1': '1.0',
     '2': '1.1',
@@ -96,6 +99,14 @@ def _check_url(host, w_url):
             urls.add(f'https://{w_url[7:]}')
 
         for url in urls:
+            # Additional checks to ensure that
+            # the final path is WELL_KNOWN_PATH
+            purl = urlparse(url)
+            if (purl.path != WELL_KNOWN_PATH
+                or len(purl.query) > 0
+                    or len(purl.params) > 0):
+                logger.warning('Invalid Assetlinks URL: %s', url)
+                continue
             r = requests.get(url,
                              timeout=5,
                              allow_redirects=False,
@@ -134,7 +145,6 @@ def get_browsable_activities(node, ns):
         path_prefixs = []
         path_patterns = []
         well_known = {}
-        well_known_path = '/.well-known/assetlinks.json'
         catg = node.getElementsByTagName('category')
         for cat in catg:
             if cat.getAttribute(f'{ns}:name') == 'android.intent.category.BROWSABLE':
@@ -168,12 +178,13 @@ def get_browsable_activities(node, ns):
                             and host != '*'):
                         host = host.replace('*.', '').replace('#', '')
                         if not valid_host(host):
+                            logger.warning('Invalid Host: %s', host)
                             continue
                         shost = f'{scheme}://{host}'
                         if port and is_number(port):
-                            c_url = f'{shost}:{port}{well_known_path}'
+                            c_url = f'{shost}:{port}{WELL_KNOWN_PATH}'
                         else:
-                            c_url = f'{shost}{well_known_path}'
+                            c_url = f'{shost}{WELL_KNOWN_PATH}'
                         well_known[c_url] = shost
         schemes = [scheme + '://' for scheme in schemes]
         browse_dic['schemes'] = schemes

mobsf/StaticAnalyzer/views/common/firebase.py

@@ -5,8 +5,8 @@
 from mobsf.MobSF.utils import (
     append_scan_status,
     upstream_proxy,
-    valid_host,
 )
+from mobsf.MobSF.security import valid_host
 
 import requests
 
@@ -80,13 +80,12 @@ def firebase_analysis(checksum, code_an_dic):
 def open_firebase(checksum, url):
     # Detect Open Firebase Database
     try:
-        invalid = 'Invalid Firebase URL'
         if not valid_host(url):
-            logger.warning(invalid)
+            logger.warning('Invalid Host: %s', url)
             return url, False
         purl = urlparse(url)
-        if not purl.netloc.endswith('.firebaseio.com'):
-            logger.warning(invalid)
+        if not purl.netloc.lower().endswith('.firebaseio.com'):
+            logger.warning('Invalid Firebase URL')
             return url, False
         base_url = f'{purl.scheme}://{purl.netloc}/.json'
         proxies, verify = upstream_proxy('https')

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "mobsf"
-version = "4.3.1"
+version = "4.3.2"
 description = "Mobile Security Framework (MobSF) is an automated, all-in-one mobile application (Android/iOS/Windows) pen-testing, malware analysis and security assessment framework capable of performing static and dynamic analysis."
 keywords = ["mobsf", "mobile security framework", "mobile security", "security tool", "static analysis", "dynamic analysis", "malware analysis"]
 authors = ["Ajin Abraham <ajin@opensecurity.in>"]