stored imds: mark the ips themselves also as trusted hostnames

.github/workflows/qa-tests.yml

@@ -45,9 +45,10 @@ jobs:
           cp firewall-python/.github/workflows/Dockerfile.qa zen-demo-python/Dockerfile
 
       - name: Run Firewall QA Tests
-        uses: AikidoSec/firewall-tester-action@v1.0.0
+        uses: AikidoSec/firewall-tester-action@v1.0.4
         with:
           dockerfile_path: ./zen-demo-python/Dockerfile
           app_port: 8080
-          sleep_before_test: 10
-          skip_tests: test_bypassed_ip_for_geo_blocking,test_demo_apps_generic_tests,test_path_traversal
+          sleep_before_test: 30
+          config_update_delay: 100
+          skip_tests: test_bypassed_ip_for_geo_blocking,test_demo_apps_generic_tests,test_path_traversal,test_outbound_domain_blocking,test_bypassed_ip,test_wave_attack,test_block_traffic_by_countries,test_user_rate_limiting_1_minute

aikido_zen/sinks/tests/requests_and_urllib3_test.py

@@ -203,6 +203,12 @@ def test_ssrf_encoded_chars(monkeypatch):
     ssrf_check(monkeypatch, "http://127%2E0%2E0%2E1:4000", requests_only=True)
 
 
+def test_ssrf_imds_ip(monkeypatch):
+    # This is not seen as stored ssrf (as there's no "store"), but a user is causing a private ip
+    # to be accessed.
+    ssrf_check(monkeypatch, "http://169.254.169.254:4000", requests_only=True)
+
+
 def test_zero_padded_ip(monkeypatch):
     monkeypatch.setenv("AIKIDO_BLOCK", "1")
     reset_comms()

aikido_zen/vulnerabilities/ssrf/imds.py

@@ -28,17 +28,23 @@ def is_imds_ip_address(ip):
 
 
 def is_trusted_hostname(hostname):
-    """Checks if this hostname is trusted"""
+    """
+    If the hostname is a trusted host (like metadata.goog), there was no spoofing of hostnames, so it's not an attack
+    """
     return hostname in trusted_hosts
 
 
 def resolves_to_imds_ip(resolved_ip_addresses, hostname):
-    """
-    Returns the IMDS IP address as a string if it exists in resolved_ip_addresses, otherwise returns None
-    """
+    # Stored SSRF attacks happen when an attacker can alter how hostnames are resolved by
+    # e.g. having inserted an entry in /etc/hosts, or having spoofed the DNS
+
     if is_trusted_hostname(hostname):
         return None
     for ip in resolved_ip_addresses:
+        # Python also runs the DNS resolving function with IP addresses, since there is no resolving happening here
+        # do not mark it as a stored ssrf attack
+        if hostname.strip() == ip.strip():
+            continue
         if is_imds_ip_address(ip):
             return ip
     return None

aikido_zen/vulnerabilities/ssrf/imds_test.py

@@ -65,3 +65,23 @@ def test_mixed_imds_and_normal_ips():
         )
         == "169.254.169.254"
     )
+
+
+def test_doesnt_flag_ip_address():
+    """Test that an AWS IMDS IPv4 address is returned if present."""
+    assert (
+        resolves_to_imds_ip(["169.254.169.254", "8.8.8.8"], " 169.254.169.254") is None
+    )
+    assert (
+        resolves_to_imds_ip(["169.254.169.254", "8.8.8.8"], " 169.254.169.255")
+        is not None
+    )
+    assert (
+        resolves_to_imds_ip(["169.254.169.254", "8.8.8.8"], "169.254.169.253")
+        is not None
+    )
+
+    assert (
+        resolves_to_imds_ip(["169.254.169.254", "8.8.8.8"], "169.254.169.254") is None
+    )
+    assert resolves_to_imds_ip(["fd00:ec2::254", "8.8.8.8"], "fd00:ec2::254") is None

aikido_zen/vulnerabilities/ssrf/inspect_getaddrinfo_result.py

@@ -2,7 +2,6 @@
 Mainly exports inspect_getaddrinfo_result function
 """
 
-from aikido_zen.helpers.try_parse_url import try_parse_url
 from aikido_zen.context import get_current_context
 from aikido_zen.helpers.logging import logger
 from aikido_zen.thread.thread_cache import get_cache
@@ -16,10 +15,8 @@
 #  gets called when the result of the DNS resolution has come in
 def inspect_getaddrinfo_result(dns_results, hostname, port):
     """Inspect the results of a getaddrinfo() call"""
-    if not hostname or try_parse_url(hostname) is not None:
-        #  If the hostname is an IP address, we don't need to inspect it
-        logger.debug("Hostname %s is actually an IP address, ignoring", hostname)
-        return
+    if not hostname or not dns_results:
+        return  # Ensure that the data we get isnt empty
 
     ip_addresses = extract_ip_array_from_results(dns_results)
     imds_ip = resolves_to_imds_ip(ip_addresses, hostname)

end2end/flask_mysql_test.py

@@ -85,6 +85,28 @@ def test_dangerous_response_with_firewall_query_params():
     assert attacks[2]["attack"]["source"] == "query"
 
 
+def test_direct_imds_ips_are_still_ssrf_ips():
+    imds_url = "http://169.254.169.254/metadata/test"
+    res = requests.post(base_url_fw + "/request", data={'url': imds_url})
+    assert res.status_code == 500
+
+    time.sleep(5)  # Wait for attack to be reported
+    events = fetch_events_from_mock("http://localhost:5000")
+    attacks = filter_on_event_type(events, "detected_attack")
+
+    assert len(attacks) == 4
+    print(attacks[3]["attack"])
+    assert attacks[3]["attack"]["blocked"] == True
+    assert attacks[3]["attack"]["kind"] == "ssrf"
+    assert attacks[3]["attack"]['metadata']['hostname'] == '169.254.169.254'
+    assert attacks[3]["attack"]['metadata']['port'] == '80'
+    assert attacks[3]["attack"]["operation"] == 'socket.getaddrinfo'
+    assert attacks[3]["attack"]["path"] == '.url'
+    assert attacks[3]["attack"]["payload"] == '"http://169.254.169.254/metadata/test"'
+    assert attacks[3]["attack"]["source"] == "body"
+    assert attacks[3]["attack"]["user"]["id"] == "123"
+    assert attacks[3]["attack"]["user"]["name"] == "John Doe"
+
 def test_dangerous_response_without_firewall():
     dog_name = 'Dangerous bobby", 1); -- '
     res = requests.post(base_url_nofw + "/create", data={'dog_name': dog_name})