diff --git a/.github/workflows/qa-tests.yml b/.github/workflows/qa-tests.yml index 21318a4f2..59aa7ea61 100644 --- a/.github/workflows/qa-tests.yml +++ b/.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 diff --git a/aikido_zen/sinks/tests/requests_and_urllib3_test.py b/aikido_zen/sinks/tests/requests_and_urllib3_test.py index a030d38dd..566633b3f 100644 --- a/aikido_zen/sinks/tests/requests_and_urllib3_test.py +++ b/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() diff --git a/aikido_zen/vulnerabilities/ssrf/imds.py b/aikido_zen/vulnerabilities/ssrf/imds.py index f04605a51..19ed890fc 100644 --- a/aikido_zen/vulnerabilities/ssrf/imds.py +++ b/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 diff --git a/aikido_zen/vulnerabilities/ssrf/imds_test.py b/aikido_zen/vulnerabilities/ssrf/imds_test.py index 969ddaa3a..63fb53f8d 100644 --- a/aikido_zen/vulnerabilities/ssrf/imds_test.py +++ b/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 diff --git a/aikido_zen/vulnerabilities/ssrf/inspect_getaddrinfo_result.py b/aikido_zen/vulnerabilities/ssrf/inspect_getaddrinfo_result.py index 67e52a5bb..43fa993f1 100644 --- a/aikido_zen/vulnerabilities/ssrf/inspect_getaddrinfo_result.py +++ b/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) diff --git a/end2end/flask_mysql_test.py b/end2end/flask_mysql_test.py index a14e48650..5662dbd40 100644 --- a/end2end/flask_mysql_test.py +++ b/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})