Add SSRF protection

aikido_firewall/__init__.py

@@ -40,5 +40,7 @@ def protect(module="any", server=True):
     import aikido_firewall.sinks.mysqlclient
     import aikido_firewall.sinks.pymongo
     import aikido_firewall.sinks.psycopg2
+    import aikido_firewall.sinks.http_client
+    import aikido_firewall.sinks.socket
 
     logger.info("Aikido python firewall started")

aikido_firewall/errors/__init__.py

@@ -21,3 +21,7 @@ class AikidoRateLimiting(AikidoException):
     def __init__(self, message="You are rate limited by Aikido firewall."):
         super().__init__(message)
         self.message = message
+
+
+class AikidoSSRF(AikidoException):
+    """Exception because of SSRF"""

aikido_firewall/helpers/get_port_from_url.py

@@ -0,0 +1,22 @@
+"""
+Helper function file, see function docstring
+"""
+
+from urllib.parse import urlparse
+
+
+def get_port_from_url(url):
+    """
+    Tries to retrieve a port number from the given url
+    """
+    parsed_url = urlparse(url)
+
+    # Check if the port is specified and is a valid integer
+    if parsed_url.port is not None:
+        return parsed_url.port
+
+    # Determine the default port based on the protocol
+    if parsed_url.scheme == "https":
+        return 443
+    elif parsed_url.scheme == "http":
+        return 80

aikido_firewall/helpers/get_port_from_url_test.py

@@ -0,0 +1,9 @@
+import pytest
+from .get_port_from_url import get_port_from_url
+
+
+def test_get_port_from_url():
+    assert get_port_from_url("http://localhost:4000") == 4000
+    assert get_port_from_url("http://localhost") == 80
+    assert get_port_from_url("https://localhost") == 443
+    assert get_port_from_url("ftp://localhost") is None

aikido_firewall/sinks/http_client.py

@@ -0,0 +1,28 @@
+"""
+Sink module for `http`
+"""
+
+import copy
+import importhook
+from aikido_firewall.helpers.logging import logger
+
+
+@importhook.on_import("http.client")
+def on_http_import(http):
+    """
+    Hook 'n wrap on `http.client.HTTPConnection.putrequest`
+    Our goal is to wrap the putrequest() function of the HTTPConnection class :
+    https://github.com/python/cpython/blob/372df1950817dfcf8b9bac099448934bf8657cf5/Lib/http/client.py#L1136
+    Returns : Modified http.client object
+    """
+    modified_http = importhook.copy_module(http)
+    former_putrequest = copy.deepcopy(http.HTTPConnection.putrequest)
+
+    def aik_new_putrequest(_self, method, url, *args, **kwargs):
+        logger.info("HTTP Request [%s] %s:%s %s", method, _self.host, _self.port, url)
+        return former_putrequest(_self, method, url, *args, **kwargs)
+
+    # pylint: disable=no-member
+    setattr(http.HTTPConnection, "putrequest", aik_new_putrequest)
+    logger.debug("Wrapped `http` module")
+    return modified_http

aikido_firewall/sinks/socket.py

@@ -0,0 +1,53 @@
+"""
+Sink module for `socket`
+"""
+
+import copy
+import importhook
+from aikido_firewall.helpers.logging import logger
+from aikido_firewall.vulnerabilities.ssrf.inspect_getaddrinfo_result import (
+    inspect_getaddrinfo_result,
+)
+
+SOCKET_OPERATIONS = [
+    "gethostbyname",
+    "gethostbyaddr",
+    "getaddrinfo",
+    "create_connection",
+]
+
+
+def generate_aikido_function(former_func, op):
+    """
+    Generates a new aikido function given a former function and op
+    """
+
+    def aik_new_func(*args, **kwargs):
+        logger.info("socket.%s() Hostname : `%s`;", op, args[0])
+        res = former_func(*args, **kwargs)
+        if op is "getaddrinfo":
+            inspect_getaddrinfo_result(dns_results=res, hostname=args[0], port=args[1])
+        logger.debug("Res %s", res)
+        return res
+
+    return aik_new_func
+
+
+@importhook.on_import("socket")
+def on_socket_import(socket):
+    """
+    Hook 'n wrap on `socket`
+    Our goal is to wrap the following socket functions that take a hostname :
+      -  gethostbyname() -- map a hostname to its IP number
+      -  gethostbyaddr() -- map an IP number or hostname to DNS info
+    https://github.com/python/cpython/blob/8f19be47b6a50059924e1d7b64277ad3cef4dac7/Lib/socket.py#L10
+    Returns : Modified http.client object
+    """
+    modified_socket = importhook.copy_module(socket)
+    for op in SOCKET_OPERATIONS:
+        former_func = copy.deepcopy(getattr(socket, op))
+        setattr(modified_socket, op, generate_aikido_function(former_func, op))
+        setattr(socket, op, generate_aikido_function(former_func, op))
+
+    logger.debug("Wrapped `http` module")
+    return modified_socket

aikido_firewall/sources/flask.py

@@ -28,7 +28,7 @@
         context.set_as_current_context()
 
         response = call_next(request)
-        comms = get_comms() # get IPC facilitator
+        comms = get_comms()  # get IPC facilitator
 
         is_curr_route_useful = is_useful_route(
             response._status_code, context.route, context.method

aikido_firewall/vulnerabilities/ssrf/check_context_for_ssrf.py

@@ -0,0 +1,31 @@
+"""Exports check_context_for_ssrf"""
+
+from aikido_firewall.helpers.extract_strings_from_user_input import (
+    extract_strings_from_user_input,
+)
+from aikido_firewall.helpers.logging import logger
+from aikido_firewall.context import UINPUT_SOURCES as SOURCES
+from .find_hostname_in_userinput import find_hostname_in_userinput
+from .contains_private_ip_address import contains_private_ip_address
+
+
+def check_context_for_ssrf(hostname, port, operation, context):
+    """
+    This will check the context for SSRF
+    """
+    for source in SOURCES:
+        logger.debug("Checking source %s for SSRF", source)
+        if hasattr(context, source):
+            user_inputs = extract_strings_from_user_input(getattr(context, source))
+            for user_input, path in user_inputs.items():
+                found = find_hostname_in_userinput(user_input, hostname, port)
+                if found and contains_private_ip_address(hostname):
+                    return {
+                        "operation": operation,
+                        "kind": "ssrf",
+                        "source": source,
+                        "pathToPayload": path,
+                        "metadata": {},
+                        "payload": user_input,
+                    }
+    return {}

aikido_firewall/vulnerabilities/ssrf/contains_private_ip_address.py

@@ -0,0 +1,25 @@
+"""exports contains_private_ip_address"""
+
+from aikido_firewall.helpers.try_parse_url import try_parse_url
+from .is_private_ip import is_private_ip
+
+
+def contains_private_ip_address(hostname):
+    """
+    Checks if the hostname contains an IP that's private
+    """
+    if hostname == "localhost":
+        return True
+
+    # Attempt to parse the URL
+    url = try_parse_url(f"http://{hostname}")
+    if url is None:
+        return False
+
+    # Check for IPv6 addresses enclosed in square brackets
+    if url.hostname.startswith("[") and url.hostname.endswith("]"):
+        ipv6 = url.hostname[1:-1]  # Extract the IPv6 address
+        if is_private_ip(ipv6):
+            return True
+
+    return is_private_ip(url.hostname)

aikido_firewall/vulnerabilities/ssrf/contains_private_ip_address_test.py

@@ -0,0 +1,182 @@
+import pytest
+from .contains_private_ip_address import contains_private_ip_address
+
+
+public_ips = [
+    "44.37.112.180",
+    "46.192.247.73",
+    "71.12.102.112",
+    "101.0.26.90",
+    "111.211.73.40",
+    "156.238.194.84",
+    "164.101.185.82",
+    "223.231.138.242",
+    "::1fff:0.0.0.0",
+    "::1fff:10.0.0.0",
+    "::1fff:0:0.0.0.0",
+    "::1fff:0:10.0.0.0",
+    "2001:2:ffff:ffff:ffff:ffff:ffff:ffff",
+    "64:ff9a::0.0.0.0",
+    "64:ff9a::255.255.255.255",
+    "99::",
+    "99::ffff:ffff:ffff:ffff",
+    "101::",
+    "101::ffff:ffff:ffff:ffff",
+    "2000::",
+    "2000::ffff:ffff:ffff:ffff:ffff:ffff",
+    "2001:10::",
+    "2001:1f:ffff:ffff:ffff:ffff:ffff:ffff",
+    "2001:db7::",
+    "2001:db7:ffff:ffff:ffff:ffff:ffff:ffff",
+    "2001:db9::",
+    "fb00::",
+    "fbff:ffff:ffff:ffff:ffff:ffff:ffff:ffff",
+    "fec0::",
+]
+
+private_ips = [
+    "0.0.0.0",
+    "0000.0000.0000.0000",
+    "0.0.0.1",
+    "0.0.0.7",
+    "0.0.0.255",
+    "0.0.255.255",
+    "0.1.255.255",
+    "0.15.255.255",
+    "0.63.255.255",
+    "0.255.255.254",
+    "0.255.255.255",
+    "10.0.0.0",
+    "10.0.0.1",
+    "10.0.0.01",
+    "10.0.0.001",
+    "10.255.255.254",
+    "10.255.255.255",
+    "100.64.0.0",
+    "100.64.0.1",
+    "100.127.255.254",
+    "100.127.255.255",
+    "127.0.0.0",
+    "127.0.0.1",
+    "127.0.0.01",
+    "127.000.000.1",
+    "127.255.255.254",
+    "127.255.255.255",
+    "169.254.0.0",
+    "169.254.0.1",
+    "169.254.255.254",
+    "169.254.255.255",
+    "172.16.0.0",
+    "172.16.0.1",
+    "172.16.0.001",
+    "172.31.255.254",
+    "172.31.255.255",
+    "192.0.0.0",
+    "192.0.0.1",
+    "192.0.0.6",
+    "192.0.0.7",
+    "192.0.0.8",
+    "192.0.0.9",
+    "192.0.0.10",
+    "192.0.0.11",
+    "192.0.0.170",
+    "192.0.0.171",
+    "192.0.0.254",
+    "192.0.0.255",
+    "192.0.2.0",
+    "192.0.2.1",
+    "192.0.2.254",
+    "192.0.2.255",
+    "192.31.196.0",
+    "192.31.196.1",
+    "192.31.196.254",
+    "192.31.196.255",
+    "192.52.193.0",
+    "192.52.193.1",
+    "192.52.193.254",
+    "192.52.193.255",
+    "192.88.99.0",
+    "192.88.99.1",
+    "192.88.99.254",
+    "192.88.99.255",
+    "192.168.0.0",
+    "192.168.0.1",
+    "192.168.255.254",
+    "192.168.255.255",
+    "192.175.48.0",
+    "192.175.48.1",
+    "192.175.48.254",
+    "192.175.48.255",
+    "198.18.0.0",
+    "198.18.0.1",
+    "198.19.255.254",
+    "198.19.255.255",
+    "198.51.100.0",
+    "198.51.100.1",
+    "198.51.100.254",
+    "198.51.100.255",
+    "203.0.113.0",
+    "203.0.113.1",
+    "203.0.113.254",
+    "203.0.113.255",
+    "240.0.0.0",
+    "240.0.0.1",
+    "224.0.0.0",
+    "224.0.0.1",
+    "255.0.0.0",
+    "255.192.0.0",
+    "255.240.0.0",
+    "255.254.0.0",
+    "255.255.0.0",
+    "255.255.255.0",
+    "255.255.255.248",
+    "255.255.255.254",
+    "255.255.255.255",
+    "0000:0000:0000:0000:0000:0000:0000:0000",
+    "::",
+    "::1",
+    "::ffff:0.0.0.0",
+    "::ffff:127.0.0.1",
+    "fe80::",
+    "fe80::1",
+    "fe80::abc:1",
+    "febf:ffff:ffff:ffff:ffff:ffff:ffff:ffff",
+    "fc00::",
+    "fc00::1",
+    "fc00::abc:1",
+    "fdff:ffff:ffff:ffff:ffff:ffff:ffff:ffff",
+    "fd00:ec2::254",
+    "169.254.169.254",
+]
+
+invalid_ips = [
+    "100::ffff::",
+    "::ffff:0.0.255.255.255",
+    "::ffff:0.255.255.255.255",
+    "0000.0000",
+    "127.1",
+    "127.0.1",
+    "2130706433",
+    "0x7f000001",
+]
+
+
+def test_public_ips():
+    for ip in public_ips:
+        if ":" in ip:
+            ip = f"[{ip}]"  # IPv6 are enclosed in brackets
+        assert not contains_private_ip_address(ip), f"Expected {ip} to be public"
+
+
+def test_private_ips():
+    for ip in private_ips:
+        if ":" in ip:
+            ip = f"[{ip}]"  # IPv6 are enclosed in brackets
+        assert contains_private_ip_address(ip), f"Expected {ip} to be private"
+
+
+def test_invalid_ips():
+    for ip in invalid_ips:
+        if ":" in ip:
+            ip = f"[{ip}]"  # IPv6 are enclosed in brackets
+        assert not contains_private_ip_address(ip), f"Expected {ip} to be invalid"