diff --git a/.github/workflows/smoke-test-ffi.yml b/.github/workflows/smoke-test-ffi.yml index a6705062b..319bcdc6a 100644 --- a/.github/workflows/smoke-test-ffi.yml +++ b/.github/workflows/smoke-test-ffi.yml @@ -48,7 +48,7 @@ jobs: run: iex "& { $(iwr 'https://github.com/AikidoSec/safe-chain/releases/latest/download/install-safe-chain.ps1' -UseBasicParsing) } -ci" - name: Install dependencies - run: make install && cp -r .cache/binaries/* aikido_zen/libs + run: make install - name: Run tests run: poetry run pytest aikido_zen/vulnerabilities/sql_injection/init_test.py diff --git a/Makefile b/Makefile index 0dfd05eeb..4dce00482 100644 --- a/Makefile +++ b/Makefile @@ -5,8 +5,6 @@ BUILD_DIR := dist/ .PHONY: build build: check_binaries @echo "Building using poetry, creates $(BUILD_DIR) folder." - @echo "Copying binaries from $(CACHE_DIR) folder : " - cp -r .cache/binaries/* aikido_zen/libs poetry build .PHONY: clean clean: @@ -46,20 +44,27 @@ benchmark: # Binaries cache : -BASE_URL = https://github.com/AikidoSec/zen-internals/releases/download/v0.1.60 +BASE_URL = https://github.com/AikidoSec/zen-internals/releases/download/v0.1.75-beta.3 FILES = \ libzen_internals_aarch64-apple-darwin.dylib \ libzen_internals_aarch64-apple-darwin.dylib.sha256sum \ + libzen_internals_aarch64-pc-windows-msvc.dll \ + libzen_internals_aarch64-pc-windows-msvc.dll.sha256sum \ libzen_internals_aarch64-unknown-linux-gnu.so \ libzen_internals_aarch64-unknown-linux-gnu.so.sha256sum \ + libzen_internals_aarch64-unknown-linux-musl.so \ + libzen_internals_aarch64-unknown-linux-musl.so.sha256sum \ libzen_internals_x86_64-apple-darwin.dylib \ libzen_internals_x86_64-apple-darwin.dylib.sha256sum \ libzen_internals_x86_64-pc-windows-gnu.dll \ libzen_internals_x86_64-pc-windows-gnu.dll.sha256sum \ libzen_internals_x86_64-unknown-linux-gnu.so \ - libzen_internals_x86_64-unknown-linux-gnu.so.sha256sum + libzen_internals_x86_64-unknown-linux-gnu.so.sha256sum \ + libzen_internals_x86_64-unknown-linux-musl.so \ + libzen_internals_x86_64-unknown-linux-musl.so.sha256sum binaries: binaries_make_dir $(addprefix .cache/binaries/, $(FILES)) + echo "$(BASE_URL)" > $(CACHE_DIR)/.version binaries_make_dir: rm -rf .cache/binaries mkdir -p .cache/binaries/ @@ -67,14 +72,16 @@ binaries_make_dir: @echo "Downloading $*..." curl -L -o $@ $(BASE_URL)/$* +# Binaries from another release may not match the Python bindings, so download them again. .PHONY: check_binaries check_binaries: - @if [ -d "$(CACHE_DIR)" ]; then \ - echo "Directory $(CACHE_DIR) exists."; \ + @if [ "$$(cat $(CACHE_DIR)/.version 2>/dev/null)" = "$(BASE_URL)" ]; then \ + echo "Directory $(CACHE_DIR) has the binaries from $(BASE_URL)."; \ else \ - echo "Directory $(CACHE_DIR) is empty. Running 'make binaries'..."; \ + echo "Directory $(CACHE_DIR) is missing or outdated. Running 'make binaries'..."; \ $(MAKE) binaries; \ fi + cp -r $(CACHE_DIR)/* aikido_zen/libs # Replace version number automatically on publish : diff --git a/aikido_zen/background_process/cloud_connection_manager/update_firewall_lists.py b/aikido_zen/background_process/cloud_connection_manager/update_firewall_lists.py index ab54600a2..199ed8c8e 100644 --- a/aikido_zen/background_process/cloud_connection_manager/update_firewall_lists.py +++ b/aikido_zen/background_process/cloud_connection_manager/update_firewall_lists.py @@ -24,9 +24,9 @@ def update_firewall_lists(connection_manager): if not isinstance(blocked_user_agents, str): return - connection_manager.firewall_lists.set_blocked_ips(blocked_ips) - connection_manager.firewall_lists.set_allowed_ips(allowed_ips) - connection_manager.firewall_lists.set_blocked_user_agents(blocked_user_agents) + connection_manager.firewall_lists.update( + blocked_ips, allowed_ips, blocked_user_agents + ) except Exception as e: logger.debug("Exception in update_firewall_lists: %s", e) diff --git a/aikido_zen/background_process/cloud_connection_manager/update_firewall_lists_test.py b/aikido_zen/background_process/cloud_connection_manager/update_firewall_lists_test.py index dc8f79d46..e81216eb5 100644 --- a/aikido_zen/background_process/cloud_connection_manager/update_firewall_lists_test.py +++ b/aikido_zen/background_process/cloud_connection_manager/update_firewall_lists_test.py @@ -4,6 +4,9 @@ from aikido_zen.background_process.cloud_connection_manager import ( CloudConnectionManager, ) +from aikido_zen.background_process.commands.check_firewall_lists import ( + process_check_firewall_lists, +) from .update_firewall_lists import update_firewall_lists @@ -72,6 +75,27 @@ def fetch_firewall_lists(self, token): } +class MockApiInvalidNetwork: + """Mock API class to simulate a list with one network that is not valid.""" + + def __init__(self, invalid_network): + self.invalid_network = invalid_network + + def fetch_firewall_lists(self, token): + return { + "success": True, + "blockedIPAddresses": [ + { + "source": "new", + "description": "Example description", + "ips": ["10.0.0.0/8", self.invalid_network], + } + ], + "allowedIPAddresses": [], + "blockedUserAgents": "NewBot", + } + + @pytest.fixture def connection_manager(): """Fixture to create a CloudConnectionManager instance with a mock API.""" @@ -107,6 +131,93 @@ def test_update_firewall_lists_success(connection_manager): assert not connection_manager.firewall_lists.is_user_agent_blocked(None) +def test_update_firewall_lists_keeps_allowed_ips_when_rebuild_fails( + connection_manager, monkeypatch +): + update_firewall_lists(connection_manager) + assert not connection_manager.firewall_lists.is_allowed_ip("1.1.1.1") + + monkeypatch.setattr( + "aikido_zen.helpers.zen_internals._ip_matcher_create", lambda *args: None + ) + update_firewall_lists(connection_manager) + + assert not connection_manager.firewall_lists.is_allowed_ip("1.1.1.1") + assert connection_manager.firewall_lists.is_allowed_ip("192.168.2.50") + assert connection_manager.firewall_lists.is_blocked_ip("192.168.1.1") + + +@pytest.mark.parametrize("invalid_network", ["not-a-network", "\udc80.1.1.1"]) +def test_update_firewall_lists_skips_invalid_networks_and_logs_them( + connection_manager, caplog, invalid_network +): + update_firewall_lists(connection_manager) + + connection_manager.api = MockApiInvalidNetwork(invalid_network) + update_firewall_lists(connection_manager) + + assert "IP list new has 1 invalid network(s)" in caplog.text + assert connection_manager.firewall_lists.is_blocked_ip("10.0.0.1") + assert not connection_manager.firewall_lists.is_blocked_ip("192.168.1.1") + assert connection_manager.firewall_lists.is_allowed_ip("1.1.1.1") + assert connection_manager.firewall_lists.is_user_agent_blocked("NewBot") + assert not connection_manager.firewall_lists.is_user_agent_blocked("BadBot") + + +@pytest.mark.parametrize( + "allowed_entry", + [ + {"source": "list1", "description": "Example description"}, + {"source": "list1", "description": "Example description", "ips": None}, + {"source": "list1", "description": "Example description", "ips": []}, + ], +) +def test_update_firewall_lists_skips_an_allowlist_entry_without_ips( + connection_manager, allowed_entry +): + connection_manager.firewall_lists.update( + [{"source": "new", "description": "Example", "ips": ["10.0.0.0/8"]}], + [allowed_entry], + "NewBot", + ) + + assert connection_manager.firewall_lists.is_allowed_ip("8.8.8.8") + assert connection_manager.firewall_lists.is_blocked_ip("10.0.0.1") + assert connection_manager.firewall_lists.is_user_agent_blocked("NewBot") + + +@pytest.mark.parametrize( + "blocked_entry", + [ + {"source": "new", "ips": ["10.0.0.0/8"]}, + {"source": "new", "description": "", "ips": ["10.0.0.0/8"]}, + ], +) +def test_update_firewall_lists_blocks_ips_from_a_list_without_description( + connection_manager, blocked_entry +): + connection_manager.firewall_lists.update([blocked_entry], [], "") + + result = process_check_firewall_lists( + connection_manager, {"ip": "10.0.0.1", "user-agent": None} + ) + assert result == { + "blocked": True, + "type": "blocklist", + "reason": "a firewall list", + } + + +def test_update_firewall_lists_without_ip_matcher(connection_manager, monkeypatch): + monkeypatch.setattr("aikido_zen.helpers.zen_internals._ip_matcher_create", None) + + update_firewall_lists(connection_manager) + + assert not connection_manager.firewall_lists.is_blocked_ip("192.168.1.1") + assert connection_manager.firewall_lists.is_allowed_ip("1.1.1.1") + assert connection_manager.firewall_lists.is_user_agent_blocked("BadBot") + + def test_update_firewall_lists_no_ua(connection_manager): # Call the function to test connection_manager.api = MockApiNoUA() diff --git a/aikido_zen/background_process/service_config.py b/aikido_zen/background_process/service_config.py index 586e0b36f..9a867012c 100644 --- a/aikido_zen/background_process/service_config.py +++ b/aikido_zen/background_process/service_config.py @@ -61,7 +61,11 @@ def set_endpoints(self, endpoints): # Skip empty allowlist continue - endpoint["allowedIPAddresses"] = IPMatcher(endpoint["allowedIPAddresses"]) + allowed_ips = IPMatcher(endpoint["allowedIPAddresses"]) + if not allowed_ips.is_built(): + # An allowlist that failed to build would block the visitors it should allow. + continue + endpoint["allowedIPAddresses"] = allowed_ips def get_endpoints(self, route_metadata): """ diff --git a/aikido_zen/background_process/service_config_test.py b/aikido_zen/background_process/service_config_test.py index dd57e90a6..ebc738c3b 100644 --- a/aikido_zen/background_process/service_config_test.py +++ b/aikido_zen/background_process/service_config_test.py @@ -1,6 +1,9 @@ import pytest from .service_config import ServiceConfig from aikido_zen.helpers.ip_matcher import IPMatcher +from aikido_zen.sources.functions.ip_allowed_to_access_route import ( + ip_allowed_to_access_route, +) def test_service_config_outbound_blocking_initialization(): @@ -321,6 +324,30 @@ def test_service_config_with_empty_allowlist(): assert len(admin_endpoint["allowedIPAddresses"]) == 0 +def test_service_config_skips_allowlist_that_failed_to_build(monkeypatch): + monkeypatch.setattr("aikido_zen.helpers.zen_internals._ip_matcher_create", None) + endpoints = [ + { + "graphql": False, + "method": "GET", + "route": "/admin", + "rate_limiting": { + "enabled": False, + "max_requests": 10, + "window_size_in_ms": 1000, + }, + "allowedIPAddresses": ["1.2.3.4"], + "force_protection_off": False, + }, + ] + service_config = ServiceConfig(endpoints, "2023-10-01", [], [], True) + + route_metadata = {"method": "GET", "route": "/admin", "url": "/admin"} + admin_endpoints = service_config.get_endpoints(route_metadata) + assert admin_endpoints[0]["allowedIPAddresses"] == ["1.2.3.4"] + assert ip_allowed_to_access_route("8.8.8.8", route_metadata, admin_endpoints) + + def test_service_config_enabled_features(): """Test the update_enabled_features and is_feature_enabled methods""" config = ServiceConfig( diff --git a/aikido_zen/helpers/get_lib_path.py b/aikido_zen/helpers/get_lib_path.py deleted file mode 100644 index 72536409c..000000000 --- a/aikido_zen/helpers/get_lib_path.py +++ /dev/null @@ -1,36 +0,0 @@ -"""Exports get_binary_path""" - -import platform -import os - - -def get_binary_path(): - """Returns an absolute path for Rust binary file""" - current_dir = os.path.dirname(os.path.abspath(__file__)) - lib_path = os.path.join(current_dir, "../libs", get_file_name()) - return lib_path - - -def get_file_name(): - """Gives you the file name for the binary based on platform info""" - os_name = platform.system().lower() - machine = platform.machine().lower() - file_name = "libzen_internals_" - - # On macOS, platform.machine() returns "arm64" for Apple Silicon - # On Linux, platform.machine() returns "aarch64" for ARM64 - if "arm64" in machine or "aarch64" in machine: - file_name += "aarch64-" - # On macOS, platform.machine() returns "x86_64" for Intel - # On Linux, platform.machine() returns "x86_64" for AMD64 - elif "x86_64" in machine or "amd64" in machine: - file_name += "x86_64-" # x86_64 or AMD64 - - if os_name == "windows": - file_name += "pc-windows-gnu.dll" # Windows - elif os_name == "darwin": - file_name += "apple-darwin.dylib" # macOS - elif os_name in ["linux", "linux2"]: - file_name += "unknown-linux-gnu.so" # Linux - - return file_name diff --git a/aikido_zen/helpers/ip_matcher/__init__.py b/aikido_zen/helpers/ip_matcher/__init__.py index 98ca78a97..916cb5a3b 100644 --- a/aikido_zen/helpers/ip_matcher/__init__.py +++ b/aikido_zen/helpers/ip_matcher/__init__.py @@ -1,64 +1,45 @@ -import ipaddress - -try: - import pytricia - - PYTRICIA_AVAILABLE = True -except ImportError: - PYTRICIA_AVAILABLE = False - from aikido_zen.helpers.logging import logger - - logger.warning( - "pytricia is not available. This happens on windows devices where pytricia is not supported yet." - "Using fallback, this may result in slower performance." - "You can try to install pytricia for better performance: pip install pytricia" - ) - - -def preparse(network: str) -> str: - # Remove the brackets around IPv6 addresses if they are there. - network = network.strip("[]") - try: - ip = ipaddress.IPv6Address(network) - if ip.ipv4_mapped: - return str(ip.ipv4_mapped) - except ValueError: - pass - return network - - -if PYTRICIA_AVAILABLE: - - class IPMatcher: - def __init__(self, networks=None): - self.trie = pytricia.PyTricia(128) - if networks is not None: - for s in networks: - self._add(s) - # We freeze in constructor ensuring that after initialization the IPMatcher is always frozen. - self.trie.freeze() - - def has(self, network): - try: - return self.trie.get(preparse(network)) is not None - except ValueError: - return False - - def _add(self, network): - try: - self.trie[preparse(network)] = True - except ValueError: - pass - except SystemError: - # SystemError's have been known to occur in the PyTricia library (see issue #34 e.g.), - # best to play it safe and catch these errors. - pass - return self - - def is_empty(self): - return len(self.trie) == 0 - -else: - # Fallback to pure Python implementation - this happens on windows machines since pytricia is not - # fully supported there. - from aikido_zen.helpers.ip_matcher_fallback import IPMatcher # noqa: F401 +import weakref + +from aikido_zen.helpers import zen_internals + + +class IPMatcher: + def __init__(self, networks=None): + # json.loads can return lone surrogates such as "\udc80", and .encode() raises on + # them, which would fail the whole list update. Valid networks are ASCII. + # The library splits on newlines, so a newline would turn one entry into two. + networks = list(networks or []) + self._networks = [ + network + for network in networks + if isinstance(network, str) and network.isascii() and "\n" not in network + ] + self._skipped_count = len(networks) - len(self._networks) + self._handle = zen_internals.ip_matcher_create( + "\n".join(self._networks).encode() + ) + if self._handle is not None: + weakref.finalize(self, zen_internals.ip_matcher_free, self._handle) + + def is_built(self): + return self._handle is not None + + def rejected_count(self): + if self._handle is None: + return self._skipped_count + return self._skipped_count + zen_internals.ip_matcher_rejected_count( + self._handle + ) + + def has(self, network): + if self._handle is None: + return False + if not isinstance(network, str) or not network.isascii(): + return False + return zen_internals.ip_matcher_has(self._handle, network.encode()) + + def is_empty(self): + return len(self._networks) == 0 + + def __reduce__(self): + return self.__class__, (self._networks,) diff --git a/aikido_zen/helpers/ip_matcher/init_test.py b/aikido_zen/helpers/ip_matcher/init_test.py index e06a3ca4d..bfddaed13 100644 --- a/aikido_zen/helpers/ip_matcher/init_test.py +++ b/aikido_zen/helpers/ip_matcher/init_test.py @@ -1,3 +1,6 @@ +import gc +import pickle + import pytest from . import IPMatcher @@ -56,6 +59,9 @@ def test_with_invalid_ranges(): "123.123.123.123/1999", "", ",,,", + None, + "\udc80.1.1.1", + "1.2.3.4\n0.0.0.0/0", "192.168.0.124/32", "192.168.0.125/32", "192.168.0.170/32", @@ -70,8 +76,10 @@ def test_with_invalid_ranges(): assert matcher.has("192.168.0.222") == True assert matcher.has("192.168.0.1") == True assert matcher.has("10.0.0.1") == False + assert matcher.has("8.8.8.8") == False assert matcher.has("192.168.0.255") == True assert matcher.has("") == False + assert matcher.has(None) == False assert matcher.has("1") == False assert matcher.has("192.168.0.1/32") == True @@ -114,6 +122,14 @@ def test_with_ipv6_ranges(): assert matcher.has("2001:db9::abc") == True +def test_strips_zones_from_scoped_ipv6_addresses(): + matcher = IPMatcher(["fe80::/10", "::1/128"]) + assert matcher.has("fe80::1%eth0") == True + assert matcher.has("fe80::1%dummy0") == True + assert matcher.has("::1%lo") == True + assert matcher.has("fe80::1%") == False + + def test_mix_ipv4_and_ipv6(): input_list = ["2002:db8::/32", "10.0.0.0/8"] matcher = IPMatcher(input_list) @@ -123,6 +139,9 @@ def test_mix_ipv4_and_ipv6(): assert matcher.has("10.0.0.1") == True assert matcher.has("10.0.0.255") == True assert matcher.has("192.168.1.1") == False + assert matcher.has("a00::1") == False + assert IPMatcher(["205.30.0.0/16"]).has("cd1e::1") == False + assert IPMatcher(["205.30.0.0/16"]).has("cd1e:ffff::1") == False def test_add_ips_later(): @@ -144,10 +163,11 @@ def test_strange_ips(): matcher = IPMatcher(input_list) assert matcher.has("::ffff:0.0.0.0") == True assert matcher.has("::ffff:127.0.0.1") == True - assert matcher.has("::ffff:123") == True + assert matcher.has("::ffff:123") == False assert matcher.has("2001:db8::1") == False assert matcher.has("[::ffff:0.0.0.0]") == True assert matcher.has("::ffff:0:0:0:0") == True + assert IPMatcher(["127.0.0.0/8"]).has("::ffff:127.0.0.1") == True def test_different_cidr_ranges(): @@ -197,7 +217,54 @@ def test_allow_all_ips(): assert matcher.has("192.168.1.1") == True +def test_adjacent_4_ranges_at_end_of_address_space(): + matcher = IPMatcher(["224.0.0.0/4", "240.0.0.0/4"]) + assert matcher.has("224.0.0.1") == True + assert matcher.has("240.0.0.1") == True + assert matcher.has("255.255.255.255") == True + assert matcher.has("223.255.255.255") == False + + def test_edge_cases(): matcher1 = IPMatcher(["224.0.0.0/4"]) assert matcher1.has("224.0.0.1") == True assert matcher1.has("240.0.0.0") == False + + +def test_rejected_count(): + assert IPMatcher(["192.168.0.0/24", "2001:db8::/32"]).rejected_count() == 0 + assert ( + IPMatcher(["192.168.0.0/24", "foobar", "", None, "\udc80"]).rejected_count() + == 4 + ) + assert IPMatcher().rejected_count() == 0 + assert IPMatcher(ip for ip in ["192.168.0.0/24", "foobar"]).rejected_count() == 1 + + +def test_pickle_round_trip(): + matcher = pickle.loads(pickle.dumps(IPMatcher(["192.168.0.0/24", "2001:db8::/32"]))) + assert matcher.has("192.168.0.42") == True + assert matcher.has("2001:db8::1") == True + assert matcher.has("10.0.0.1") == False + + +def test_frees_the_matcher_when_it_is_dropped(monkeypatch): + freed = [] + monkeypatch.setattr( + "aikido_zen.helpers.zen_internals.ip_matcher_free", freed.append + ) + + matcher = IPMatcher(["192.168.0.0/24"]) + handle = matcher._handle + del matcher + gc.collect() + + assert freed == [handle] + + +def test_matches_nothing_without_the_library(monkeypatch): + monkeypatch.setattr("aikido_zen.helpers.zen_internals._ip_matcher_create", None) + + matcher = IPMatcher(["192.168.0.0/24"]) + assert matcher.is_built() == False + assert matcher.has("192.168.0.1") == False diff --git a/aikido_zen/helpers/ip_matcher_fallback/performance_test.py b/aikido_zen/helpers/ip_matcher/performance_test.py similarity index 100% rename from aikido_zen/helpers/ip_matcher_fallback/performance_test.py rename to aikido_zen/helpers/ip_matcher/performance_test.py diff --git a/aikido_zen/helpers/ip_matcher_fallback/__init__.py b/aikido_zen/helpers/ip_matcher_fallback/__init__.py deleted file mode 100644 index 254a1d7b3..000000000 --- a/aikido_zen/helpers/ip_matcher_fallback/__init__.py +++ /dev/null @@ -1,49 +0,0 @@ -""" -Based on https://github.com/demskie/netparser -MIT License - Copyright (c) 2019 alex -""" - -from .shared import parse_base_network, sort_networks, summarize_sorted_networks -from .sort import binary_search_for_insertion_index - - -class IPMatcher: - def __init__(self, networks=None): - self.sorted = [] - if networks is not None: - subnets = [] - for s in networks: - net = parse_base_network(s, False) - if net and net.is_valid(): - subnets.append(net) - sort_networks(subnets) - self.sorted = summarize_sorted_networks(subnets) - - def has(self, network): - """ - Checks if the given IP address is in the list of networks. - """ - net = parse_base_network(network, False) - if not net or not net.is_valid(): - return False - idx = binary_search_for_insertion_index(net, self.sorted) - if idx < 0: - return False - if idx < len(self.sorted) and self.sorted[idx].contains(net): - return True - if idx - 1 >= 0 and self.sorted[idx - 1].contains(net): - return True - return False - - def add(self, network): - net = parse_base_network(network, False) - if not net or not net.is_valid(): - return self - idx = binary_search_for_insertion_index(net, self.sorted) - if idx < len(self.sorted) and self.sorted[idx].compare(net) == 0: - return self - self.sorted.insert(idx, net) - return self - - def is_empty(self): - return len(self.sorted) == 0 diff --git a/aikido_zen/helpers/ip_matcher_fallback/address.py b/aikido_zen/helpers/ip_matcher_fallback/address.py deleted file mode 100644 index 529d4612b..000000000 --- a/aikido_zen/helpers/ip_matcher_fallback/address.py +++ /dev/null @@ -1,119 +0,0 @@ -""" -Based on https://github.com/demskie/netparser -MIT License - Copyright (c) 2019 alex -""" - -from aikido_zen.helpers.ip_matcher_fallback import parse - -# Constants -BEFORE = -1 -EQUALS = 0 -AFTER = 1 - - -class Address: - def __init__(self, address=None): - self.arr = [] - if address: - net = parse.network(address) - if net: - self.arr = net["bytes"] - - def bytes(self): - return self.arr if self.arr else [] - - def set_bytes(self, bytes_data): - if len(bytes_data) == 4 or len(bytes_data) == 16: - self.arr = bytes_data - else: - self.arr = [] - return self - - def destroy(self): - if self.is_valid(): - self.arr = [] - return self - - def is_valid(self): - return len(self.arr) > 0 - - def is_ipv4(self): - return len(self.arr) == 4 - - def is_ipv6(self): - return len(self.arr) == 16 - - def duplicate(self): - return Address().set_bytes(self.arr.copy()) - - def equals(self, address): - return self.compare(address) == EQUALS - - def compare(self, address): - if not self.is_valid() or not address.is_valid(): - return None - if self == address: - return EQUALS - if len(self.arr) < len(address.arr): - return BEFORE - if len(self.arr) > len(address.arr): - return AFTER - - for i, adr in enumerate(self.arr): - if adr < address.arr[i]: - return BEFORE - if adr > address.arr[i]: - return AFTER - - return EQUALS - - def apply_subnet_mask(self, cidr): - if not self.is_valid(): - return self - mask_bits = len(self.arr) * 8 - cidr - for i in range(len(self.arr) - 1, -1, -1): - mask = max(0, min(mask_bits, 8)) - if mask == 0: - return self - self.arr[i] &= ~((1 << mask) - 1) - mask_bits -= 8 - return self - - def is_base_address(self, cidr): - if not self.is_valid() or cidr < 0 or cidr > len(self.arr) * 8: - return False - if cidr == len(self.arr) * 8: - return True - mask_bits = len(self.arr) * 8 - cidr - for i in range(len(self.arr) - 1, -1, -1): - mask = max(0, min(mask_bits, 8)) - if mask == 0: - return True - if self.arr[i] != (self.arr[i] & ~((1 << mask) - 1)): - return False - mask_bits -= 8 - return True - - def increase(self, cidr): - if self.is_valid(): - self.offset_address(cidr, True) - else: - self.destroy() - return self - - def offset_address(self, cidr, forwards, throw_errors=False): - target_byte = (cidr - 1) // 8 - if self.is_valid() and 0 <= target_byte < len(self.arr): - increment = 2 ** (8 - (cidr - target_byte * 8)) - self.arr[target_byte] += increment * (1 if forwards else -1) - if target_byte >= 0: - if self.arr[target_byte] < 0: - self.arr[target_byte] = 256 + (self.arr[target_byte] % 256) - self.offset_address(target_byte * 8, forwards, throw_errors) - elif self.arr[target_byte] > 255: - self.arr[target_byte] %= 256 - self.offset_address(target_byte * 8, forwards, throw_errors) - else: - self.destroy() - else: - self.destroy() diff --git a/aikido_zen/helpers/ip_matcher_fallback/address_test.py b/aikido_zen/helpers/ip_matcher_fallback/address_test.py deleted file mode 100644 index 3eabe5274..000000000 --- a/aikido_zen/helpers/ip_matcher_fallback/address_test.py +++ /dev/null @@ -1,49 +0,0 @@ -import pytest -from .address import Address - - -def test_is_ipv4(): - assert Address("1.2.3.5").is_ipv4() == True - assert Address("::1").is_ipv4() == False - - -def test_is_ipv6(): - assert Address("::1").is_ipv6() == True - - -def test_compare(): - assert Address("1.2.3.4").compare(Address("1.2.3.5")) == -1 - assert Address("1.2.3.4").compare(Address("1.2.3.4")) == 0 - assert Address("1.2.3.4").compare(Address("1.2.3.4").duplicate()) == 0 - - # edge cases - assert Address().compare(Address()) is None - assert Address("1.2.3.4").compare(Address()) is None - - -def test_bytes_method(): - assert Address("1.2.3.4").bytes() == [1, 2, 3, 4] - assert Address().bytes() == [] - - -def test_set_bytes_method(): - assert Address("1.2.3.4").set_bytes([3]).bytes() == [] - - -def test_equals_method(): - assert Address("1.2.3.4").equals(Address("1.2.3.4")) == True - assert Address("1.2.3.4").equals(Address("1.2.3.5")) == False - - -def test_apply_subnet_mask(): - assert Address().apply_subnet_mask(0) is not None - - -def test_increase_method(): - assert Address().increase(0).bytes() == [] - assert Address().bytes() == [] - - -def test_self_comparison(): - a = Address("3.4.5.6") - assert a.compare(a) == 0 diff --git a/aikido_zen/helpers/ip_matcher_fallback/init_test.py b/aikido_zen/helpers/ip_matcher_fallback/init_test.py deleted file mode 100644 index c0b78e625..000000000 --- a/aikido_zen/helpers/ip_matcher_fallback/init_test.py +++ /dev/null @@ -1,204 +0,0 @@ -import pytest -from . import IPMatcher - - -def test_single_ipv4s(): - input_list = [ - "192.168.0.0/32", - "192.168.0.3/32", - "192.168.0.24/32", - "192.168.0.52/32", - "192.168.0.123/32", - "192.168.0.124/32", - "192.168.0.125/32", - "192.168.0.170/32", - "192.168.0.171/32", - "192.168.0.222/32", - "192.168.0.234/32", - "192.168.0.255/32", - ] - matcher = IPMatcher(input_list) - assert matcher.has("192.168.0.254") == False - assert matcher.has("192.168.0.1") == False - assert matcher.has("192.168.0.255") == True - assert matcher.has("192.168.0.24") == True - - -def test_with_ranges(): - input_list = [ - "192.168.0.0/24", - "192.168.0.3/32", - "192.168.0.24/32", - "192.168.0.52/32", - "192.168.0.123/32", - "192.168.0.124/32", - "192.168.0.125/32", - "192.168.0.170/32", - "192.168.0.171/32", - "192.168.0.222/32", - "192.168.0.234/32", - "192.168.0.255/32", - ] - matcher = IPMatcher(input_list) - assert matcher.has("192.168.0.254") == True # Now included because of the /24 range - assert matcher.has("10.0.0.1") == False - assert matcher.has("192.168.0.234") == True - - -def test_with_invalid_ranges(): - input_list = [ - "192.168.0.0/24", - "192.168.0.3/32", - "192.168.0.24/32", - "192.168.0.52/32", - "foobar", - "0.a.0.0/32", - "123.123.123.123/1999", - "", - ",,,", - "192.168.0.124/32", - "192.168.0.125/32", - "192.168.0.170/32", - "192.168.0.171/32", - "192.168.0.222/32", - "192.168.0.234/32", - "192.168.0.255", - ] - matcher = IPMatcher(input_list) - assert matcher.has("192.168.0.254") == True - assert matcher.has("foobar") == False - assert matcher.has("192.168.0.222") == True - assert matcher.has("192.168.0.1") == True - assert matcher.has("10.0.0.1") == False - assert matcher.has("192.168.0.255") == True - assert matcher.has("") == False - assert matcher.has("1") == False - assert matcher.has("192.168.0.1/32") == True - - -def test_with_empty_ranges(): - input_list = [] - matcher = IPMatcher(input_list) - assert matcher.has("192.168.2.1") == False - assert matcher.has("foobar") == False - - -def test_with_ipv6_ranges(): - input_list = [ - "2002:db8::/32", - "2001:db8::1/128", - "2001:db8::2/128", - "2001:db8::3/128", - "2001:db8::4/128", - "2001:db8::5/128", - "2001:db8::6/128", - "2001:db8::7/128", - "2001:db8::8/128", - "2001:db8::9/128", - "2001:db8::a/128", - "2001:db8::b/128", - "2001:db8::c/128", - "2001:db8::d/128", - "2001:db8::e/128", - "[2001:db8::f]", - "2001:db9::abc", - ] - matcher = IPMatcher(input_list) - assert matcher.has("2001:db8::1") == True - assert matcher.has("2001:db8::0") == False - assert matcher.has("2001:db8::f") == True - assert matcher.has("[2001:db8::f]") == True - assert matcher.has("2001:db8::10") == False - assert matcher.has("2002:db8::1") == True - assert matcher.has("2002:db8::2f:2") == True - assert matcher.has("2001:db9::abc") == True - - -def test_mix_ipv4_and_ipv6(): - input_list = ["2002:db8::/32", "10.0.0.0/8"] - matcher = IPMatcher(input_list) - assert matcher.has("2001:db8::1") == False - assert matcher.has("2001:db8::0") == False - assert matcher.has("2002:db8::1") == True - assert matcher.has("10.0.0.1") == True - assert matcher.has("10.0.0.255") == True - assert matcher.has("192.168.1.1") == False - - -def test_add_ips_later(): - input_list = ["2002:db8::/32", "10.0.0.0/8"] - matcher = IPMatcher() - assert matcher.has("2001:db8::0") == False - assert matcher.has("2002:db8::1") == False - for ip in input_list: - matcher.add(ip) - assert matcher.has("2001:db8::1") == False - assert matcher.has("2001:db8::0") == False - assert matcher.has("2002:db8::1") == True - assert matcher.has("10.0.0.1") == True - assert matcher.has("10.0.0.255") == True - assert matcher.has("192.168.1.1") == False - - -def test_strange_ips(): - input_list = ["::ffff:0.0.0.0", "::ffff:0:0:0:0", "::ffff:127.0.0.1"] - matcher = IPMatcher(input_list) - assert matcher.has("::ffff:0.0.0.0") == True - assert matcher.has("::ffff:127.0.0.1") == True - assert matcher.has("::ffff:123") == False - assert matcher.has("2001:db8::1") == False - assert matcher.has("[::ffff:0.0.0.0]") == True - assert matcher.has("::ffff:0:0:0:0") == True - - -def test_different_cidr_ranges(): - assert IPMatcher(["123.2.0.2/0"]).has("1.1.1.1") == True - assert IPMatcher(["123.2.0.2/1"]).has("1.1.1.1") == True - assert IPMatcher(["123.2.0.2/2"]).has("1.1.1.1") == False - assert IPMatcher(["123.2.0.2/3"]).has("123.3.0.1") == True - assert IPMatcher(["123.2.0.2/4"]).has("123.3.0.1") == True - assert IPMatcher(["123.2.0.2/5"]).has("123.3.0.1") == True - assert IPMatcher(["123.2.0.2/6"]).has("123.3.0.1") == True - assert IPMatcher(["123.2.0.2/7"]).has("123.3.0.1") == True - assert IPMatcher(["123.2.0.2/8"]).has("123.3.0.1") == True - assert IPMatcher(["123.2.0.2/9"]).has("123.3.0.1") == True - assert IPMatcher(["123.2.0.2/10"]).has("123.3.0.1") == True - assert IPMatcher(["123.2.0.2/11"]).has("123.3.0.1") == True - assert IPMatcher(["123.2.0.2/12"]).has("123.3.0.1") == True - assert IPMatcher(["123.2.0.2/13"]).has("123.3.0.1") == True - assert IPMatcher(["123.2.0.2/14"]).has("123.3.0.1") == True - assert IPMatcher(["123.2.0.2/15"]).has("123.3.0.1") == True - assert IPMatcher(["123.2.0.2/16"]).has("123.3.0.1") == False - assert IPMatcher(["123.2.0.2/17"]).has("123.2.0.1") == True - assert IPMatcher(["123.2.0.2/18"]).has("123.2.0.1") == True - assert IPMatcher(["123.2.0.2/19"]).has("123.2.0.1") == True - assert IPMatcher(["123.2.0.2/20"]).has("123.2.0.1") == True - assert IPMatcher(["123.2.0.2/21"]).has("123.2.0.1") == True - assert IPMatcher(["123.2.0.2/22"]).has("123.2.0.1") == True - assert IPMatcher(["123.2.0.2/23"]).has("123.2.0.1") == True - assert IPMatcher(["123.2.0.2/24"]).has("123.2.0.1") == True - assert IPMatcher(["123.2.0.2/25"]).has("123.2.0.1") == True - assert IPMatcher(["123.2.0.2/26"]).has("123.2.0.1") == True - assert IPMatcher(["123.2.0.2/27"]).has("123.2.0.1") == True - assert IPMatcher(["123.2.0.2/29"]).has("123.2.0.1") == True - assert IPMatcher(["123.2.0.2/30"]).has("123.2.0.1") == True - assert IPMatcher(["123.2.0.2/31"]).has("123.2.0.1") == False - assert IPMatcher(["123.2.0.2/32"]).has("123.2.0.2") == True - - -def test_allow_all_ips(): - matcher = IPMatcher(["0.0.0.0/0", "::/0"]) - assert matcher.has("1.2.3.4") == True - assert matcher.has("::1") == True - assert matcher.has("::ffff:1234") == True - assert matcher.has("1.1.1.1") == True - assert matcher.has("2002:db8::1") == True - assert matcher.has("10.0.0.1") == True - assert matcher.has("10.0.0.255") == True - assert matcher.has("192.168.1.1") == True - - -def test_edge_cases(): - matcher1 = IPMatcher(["224.0.0.0/4"]) - assert matcher1.has("224.0.0.1") == True - assert matcher1.has("240.0.0.0") == False diff --git a/aikido_zen/helpers/ip_matcher_fallback/network.py b/aikido_zen/helpers/ip_matcher_fallback/network.py deleted file mode 100644 index dd1e9f689..000000000 --- a/aikido_zen/helpers/ip_matcher_fallback/network.py +++ /dev/null @@ -1,111 +0,0 @@ -""" -Based on https://github.com/demskie/netparser -MIT License - Copyright (c) 2019 alex -""" - -import math -import aikido_zen.helpers.ip_matcher_fallback.parse as parse -from .address import Address - -BEFORE = -1 -EQUALS = 0 -AFTER = 1 - - -class Network: - def __init__(self, network=None): - self.addr = Address() - self.netbits = -1 - if network: - net = parse.network(network) - if net: - self.addr.set_bytes(net["bytes"]) - self.netbits = net["cidr"] - - def destroy(self): - self.addr.destroy() - self.netbits = -1 - return self - - def cidr(self): - if self.is_valid(): - return self.netbits - return float("nan") - - def is_valid(self): - return self.addr.is_valid() and self.netbits != -1 - - def duplicate(self): - network = Network() - if self.is_valid(): - network.addr.set_bytes(self.addr.bytes().copy()) - network.netbits = self.netbits - return network - - def next(self): - self.addr.increase(self.netbits) - return self - - def set_cidr(self, cidr): - if not self.addr.is_valid(): - self.destroy() - else: - cidr = math.floor(cidr) - if 0 <= cidr <= len(self.addr.bytes()) * 8: - self.netbits = cidr - else: - self.destroy() - return self - - def compare(self, network): - if not self.is_valid() or not network.is_valid(): - return None - cmp = self.addr.compare(network.addr) - if cmp != EQUALS: - return cmp - if self.netbits < network.netbits: - return BEFORE - if self.netbits > network.netbits: - return AFTER - return EQUALS - - def contains(self, network): - if not self.is_valid() or not network.is_valid(): - return False - if len(self.addr.bytes()) != len(network.addr.bytes()): - return False - if self.netbits == 0: - return True - if network.netbits == 0: - return False - if self.addr.compare(network.addr) == AFTER: - return False - next_network = self.duplicate().next() - other_next = network.duplicate().next() - if not next_network.is_valid(): - return True - if next_network.addr.compare(other_next.addr) == BEFORE: - return False - return True - - def adjacent(self, network): - if not self.is_valid() or not network.is_valid(): - return False - if len(self.addr.bytes()) != len(network.addr.bytes()): - return False - if self.netbits == 0 or network.netbits == 0: - return True - cmp = self.addr.compare(network.addr) - if cmp == EQUALS: - return False - if cmp == BEFORE: - alpha = self.duplicate().next() - bravo = network - else: - alpha = network.duplicate().next() - bravo = self - if not alpha.is_valid(): - return False - if alpha.addr.compare(bravo.addr) == EQUALS: - return True - return False diff --git a/aikido_zen/helpers/ip_matcher_fallback/network_test.py b/aikido_zen/helpers/ip_matcher_fallback/network_test.py deleted file mode 100644 index 152805b5e..000000000 --- a/aikido_zen/helpers/ip_matcher_fallback/network_test.py +++ /dev/null @@ -1,44 +0,0 @@ -import pytest -import math -from .network import Network - - -def test_empty_network_cidr(): - result = Network().cidr() - assert math.isnan(result) - - -def test_set_cidr_zero_invalid(): - assert not Network().set_cidr(0).is_valid() - - -def test_valid_network_is_valid(): - assert Network("192.168.2.1/24").is_valid() - - -def test_set_invalid_cidr_invalidates_network(): - assert not Network("192.168.2.1/24").set_cidr(-1).is_valid() - - -def test_compare_empty_networks(): - assert Network().compare(Network()) is None - - -def test_compare_valid_and_empty_networks(): - assert Network("192.168.2.1/24").compare(Network()) is None - - -def test_empty_network_does_not_contain_empty_network(): - assert not Network().contains(Network()) - - -def test_valid_network_does_not_contain_empty_network(): - assert not Network("192.168.2.1/24").contains(Network()) - - -def test_network_contains_subnetwork(): - assert Network("192.168.2.1/24").contains(Network("192.168.2.1/32")) - - -if __name__ == "__main__": - pytest.main([__file__]) diff --git a/aikido_zen/helpers/ip_matcher_fallback/parse.py b/aikido_zen/helpers/ip_matcher_fallback/parse.py deleted file mode 100644 index fee95da94..000000000 --- a/aikido_zen/helpers/ip_matcher_fallback/parse.py +++ /dev/null @@ -1,181 +0,0 @@ -""" -Based on https://github.com/demskie/netparser -MIT License - Copyright (c) 2019 alex -""" - -import math -import re - - -def network(s): - s = s.strip() - parts = s.split("/") - if len(parts) == 0 or len(parts) > 2: - return None - is_ipv4 = looks_like_ipv4(s) - if is_ipv4 is None: - return None - cidr = 32 if is_ipv4 else 128 - if len(parts) == 2: - x = parse_int_range(parts[1], 0, cidr) - if x is None: - return None - cidr = x - bytes_data = v4_addr_to_bytes(parts[0]) if is_ipv4 else v6_addr_to_bytes(parts[0]) - if bytes_data is None: - return None - return {"bytes": bytes_data, "cidr": cidr} - - -def looks_like_ipv4(s): - for c in s: - if c == ".": - return True - if c == ":": - return False - return None - - -def parse_int_range(old, min_val, max_val): - s = "" - for old_i in old: - if not old_i.isdigit(): - break - s += old_i - if not s: - return None - x = int(s) - if min_val <= x <= max_val: - return x - return None - - -def v4_addr_to_bytes(old): - bytes_data = [0] * 4 - parts = old.split(".") - if len(parts) == 4: - for i in range(4): - try: - x = int(parts[i]) - except ValueError: - return None - if 0 <= x <= 255: - bytes_data[i] = x - else: - return None - return bytes_data - return None - - -def v6_addr_to_bytes(s): - bytes_data = [0] * 16 - if not s: - return None - s = remove_brackets(s) - if s == "::": - return bytes_data - halves = s.split("::") - if len(halves) == 0 or len(halves) > 2: - return None - left_byte_index = parse_left_half(bytes_data, halves[0]) - if left_byte_index is None: - return None - if len(halves) == 2: - right_byte_index = parse_right_half(bytes_data, halves[1], left_byte_index) - if right_byte_index is None: - return None - return bytes_data - - -def remove_brackets(s): - if s.startswith("["): - for i in range(len(s) - 1, -1, -1): - if s[i] == "]": - return s[1:i] - return s - - -def parse_hextet(s): - s = s.strip() - if len(s) < 1 or len(s) > 4: - return float("nan") - val = 0 - for i, s_i in enumerate(s): - try: - x = int(s_i, 16) - except ValueError: - return float("nan") - val += x * (2 ** (4 * (len(s) - i - 1))) - return val - - -def parse_left_half(bytes_data, left_half): - left_byte_index = 0 - if left_half != "": - left_parts = left_half.split(":") - for left_part in left_parts: - if left_byte_index >= 16: - return None - ipv4_parts = left_part.split(".") - if len(ipv4_parts) == 0: - return None - if len(ipv4_parts) != 4: - x = parse_hextet(left_part) - if isinstance(x, float) and math.isnan(x): - return None - if x < 0 or x > 65535: - return None - bytes_data[left_byte_index] = x // 256 - bytes_data[left_byte_index + 1] = x % 256 - left_byte_index += 2 - else: - for ipv4_part in ipv4_parts: - try: - x = int(ipv4_part) - except ValueError: - return None - if x < 0 or x > 255: - return None - bytes_data[left_byte_index] = x - left_byte_index += 1 - return left_byte_index - - -def parse_right_half(bytes_data, right_half, left_byte_index): - right_byte_index = 15 - if right_half != "": - right_parts = right_half.split(":") - for i in range(len(right_parts) - 1, -1, -1): - if not right_parts[i].strip(): - return None - if left_byte_index > right_byte_index: - return None - ipv4_parts = right_parts[i].split(".") - if len(ipv4_parts) == 0: - return None - if len(ipv4_parts) != 4: - if i == len(right_parts) - 1: - right_parts[i] = remove_port_info(right_parts[i]) - x = parse_hextet(right_parts[i]) - if isinstance(x, float) and math.isnan(x): - return None - if x < 0 or x > 65535: - return None - bytes_data[right_byte_index] = x % 256 - bytes_data[right_byte_index - 1] = x // 256 - right_byte_index -= 2 - else: - for j in range(len(ipv4_parts) - 1, -1, -1): - try: - x = int(ipv4_parts[j]) - except ValueError: - return None - if x < 0 or x > 255: - return None - bytes_data[right_byte_index] = x - right_byte_index -= 1 - return right_byte_index - - -def remove_port_info(s): - return re.sub(r"(#|p|\.).*", "", s).strip() diff --git a/aikido_zen/helpers/ip_matcher_fallback/parse_test.py b/aikido_zen/helpers/ip_matcher_fallback/parse_test.py deleted file mode 100644 index b263adb26..000000000 --- a/aikido_zen/helpers/ip_matcher_fallback/parse_test.py +++ /dev/null @@ -1,24 +0,0 @@ -import pytest -from .parse import network - - -def test_ipv4_valid_cidr(): - result = network("192.168.2.1/24") - expected = {"bytes": [192, 168, 2, 1], "cidr": 24} - assert result == expected - - -def test_ipv4_invalid_cidr(): - result = network("192.168.2.1/abcde") - assert result is None - - -def test_ipv4_additional_text(): - result = network("192.168.2.1/24/test") - assert result is None - - -def test_ipv6_valid(): - result = network("::1") - expected = {"bytes": [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1], "cidr": 128} - assert result == expected diff --git a/aikido_zen/helpers/ip_matcher_fallback/shared.py b/aikido_zen/helpers/ip_matcher_fallback/shared.py deleted file mode 100644 index 2ba00afb6..000000000 --- a/aikido_zen/helpers/ip_matcher_fallback/shared.py +++ /dev/null @@ -1,67 +0,0 @@ -""" -Based on https://github.com/demskie/netparser -MIT License - Copyright (c) 2019 alex -""" - -import math -from .sort import native_sort -from .network import Network - - -def sort_networks(networks): - return native_sort(networks) - - -def increase_size_by_one_bit(network): - current_cidr = network.cidr() - if math.isnan(current_cidr) or current_cidr <= 0: - return network # Can't increase size further, or network is invalid - wider = network.set_cidr(current_cidr - 1) - wider.addr.apply_subnet_mask(wider.cidr()) - return wider - - -def summarize_sorted_networks(sorted_networks): - if not sorted_networks: - return [] - summarized = sorted_networks[:1] # Copy the first element - for idx in range(1, len(sorted_networks)): - current_network = sorted_networks[idx] - if len(summarized) == 0: - summarized.append(current_network) - continue - if summarized[-1].contains(current_network): - continue - summarized.append(current_network) - while len(summarized) >= 2: - a = summarized[-2] - b = summarized[-1] - a_cidr = a.cidr() - b_cidr = b.cidr() - # Handle NaN cases for cidr() and ensure CIDRs are equal and valid - if ( - math.isnan(a_cidr) - or math.isnan(b_cidr) - or a_cidr != b_cidr - or a_cidr <= 0 # Can't decrease CIDR if it's already <= 0 - or not a.addr.is_base_address(a_cidr - 1) - or not a.adjacent(b) - ): - break - increase_size_by_one_bit(a) - summarized.pop() - return summarized - - -def parse_base_network(s, strict=False): - net = Network(s) - if not net.is_valid(): - return None - if not strict: - net.addr.apply_subnet_mask(net.cidr()) - else: - original = net.addr.duplicate() - net.addr.apply_subnet_mask(net.cidr()) - if not net.addr.equals(original): - return None - return net diff --git a/aikido_zen/helpers/ip_matcher_fallback/shared_test.py b/aikido_zen/helpers/ip_matcher_fallback/shared_test.py deleted file mode 100644 index f61f2e229..000000000 --- a/aikido_zen/helpers/ip_matcher_fallback/shared_test.py +++ /dev/null @@ -1,25 +0,0 @@ -from aikido_zen.helpers.ip_matcher_fallback.network import Network -from aikido_zen.helpers.ip_matcher_fallback.shared import summarize_sorted_networks - - -def test_summarize_sorted_networks_empty(): - result = summarize_sorted_networks([]) - assert result == [] - - -def test_summarize_sorted_networks_single(): - networks = [Network("192.168.1.0/24")] - result = summarize_sorted_networks(networks) - assert len(result) == 1 - assert result[0].addr.bytes() == [192, 168, 1, 0] - assert result[0].cidr() == 24 - - -def test_summarize_sorted_networks_no_merge(): - networks = [Network("192.168.1.0/24"), Network("192.168.3.0/24")] - result = summarize_sorted_networks(networks) - assert len(result) == 2 - assert result[0].addr.bytes() == [192, 168, 1, 0] - assert result[0].cidr() == 24 - assert result[1].addr.bytes() == [192, 168, 3, 0] - assert result[1].cidr() == 24 diff --git a/aikido_zen/helpers/ip_matcher_fallback/sort.py b/aikido_zen/helpers/ip_matcher_fallback/sort.py deleted file mode 100644 index 8f3ef8597..000000000 --- a/aikido_zen/helpers/ip_matcher_fallback/sort.py +++ /dev/null @@ -1,60 +0,0 @@ -""" -Based on https://github.com/demskie/netparser -MIT License - Copyright (c) 2019 alex -""" - -import math -from functools import cmp_to_key - -BEFORE = -1 -EQUALS = 0 -AFTER = 1 - - -def compare_networks(a, b): - a_bytes = a.addr.bytes() - b_bytes = b.addr.bytes() - if len(a_bytes) != len(b_bytes): - return len(a_bytes) - len(b_bytes) - for i, a_byte in enumerate(a_bytes): - if a_byte != b_bytes[i]: - return a_byte - b_bytes[i] - a_cidr = a.cidr() - b_cidr = b.cidr() - if math.isnan(a_cidr) and math.isnan(b_cidr): - return 0 - if math.isnan(a_cidr): - return -1 - if math.isnan(b_cidr): - return 1 - return a_cidr - b_cidr - - -def native_sort(networks): - networks.sort(key=cmp_to_key(compare_networks)) - return networks - - -def binary_search_for_insertion_index(network, sorted_networks): - if not sorted_networks or len(sorted_networks) == 0: - return 0 - left = 0 - right = len(sorted_networks) - 1 - while left < right: - middle = left + (right - left) // 2 - cmp_result = sorted_networks[middle].compare(network) - if cmp_result is None: - return middle + 1 - if cmp_result == EQUALS: - return middle + 1 - if cmp_result == BEFORE: - left = middle + 1 - else: # AFTER - right = middle - 1 - # Now left == right - cmp_result = sorted_networks[left].compare(network) - if cmp_result is None: - return left + 1 - if cmp_result == BEFORE: - return left + 1 - return left diff --git a/aikido_zen/helpers/ip_matcher_fallback/sort_test.py b/aikido_zen/helpers/ip_matcher_fallback/sort_test.py deleted file mode 100644 index 003eac3b6..000000000 --- a/aikido_zen/helpers/ip_matcher_fallback/sort_test.py +++ /dev/null @@ -1,160 +0,0 @@ -import pytest -import math -from .network import Network -from .sort import native_sort, binary_search_for_insertion_index - - -def test_native_sort_empty_list(): - networks = [] - result = native_sort(networks) - assert result == [] - - -def test_native_sort_single_network(): - networks = [Network("192.168.1.1/24")] - result = native_sort(networks) - assert len(result) == 1 - assert result[0].addr.bytes() == [192, 168, 1, 1] - assert result[0].cidr() == 24 - - -def test_native_sort_different_byte_lengths(): - networks = [ - Network("::1/64"), # IPv6 - Network("192.168.1.1/24"), # IPv4 - ] - result = native_sort(networks) - assert len(result) == 2 - assert len(result[0].addr.bytes()) == 4 # IPv4 - assert len(result[1].addr.bytes()) == 16 # IPv6 - - -def test_native_sort_same_byte_length_different_bytes(): - networks = [ - Network("192.168.2.1/24"), - Network("192.168.1.1/24"), - ] - result = native_sort(networks) - assert len(result) == 2 - assert result[0].addr.bytes() == [192, 168, 1, 1] - assert result[1].addr.bytes() == [192, 168, 2, 1] - - -def test_native_sort_same_bytes_different_cidrs(): - networks = [ - Network("192.168.1.1/24"), - Network("192.168.1.1/16"), - ] - result = native_sort(networks) - assert len(result) == 2 - assert result[0].cidr() == 16 - assert result[1].cidr() == 24 - - -def test_native_sort_invalid_networks(): - invalid_net = Network().destroy() - valid_net = Network("192.168.1.1/24") - networks = [invalid_net, valid_net] - result = native_sort(networks) - assert len(result) == 2 - assert not result[0].is_valid() - assert result[1].is_valid() - - -def test_native_sort_already_sorted(): - networks = [ - Network("192.168.1.1/16"), - Network("192.168.1.1/24"), - Network("192.168.2.1/24"), - ] - original_bytes = [net.addr.bytes() for net in networks] - original_cidrs = [net.cidr() for net in networks] - result = native_sort(networks) - assert len(result) == 3 - for i in range(3): - assert result[i].addr.bytes() == original_bytes[i] - assert result[i].cidr() == original_cidrs[i] - - -def test_native_sort_reverse_order(): - networks = [ - Network("192.168.2.1/24"), - Network("192.168.1.1/24"), - Network("192.168.1.1/16"), - ] - result = native_sort(networks) - assert len(result) == 3 - assert result[0].addr.bytes() == [192, 168, 1, 1] - assert result[0].cidr() == 16 - assert result[1].addr.bytes() == [192, 168, 1, 1] - assert result[1].cidr() == 24 - assert result[2].addr.bytes() == [192, 168, 2, 1] - assert result[2].cidr() == 24 - - -def test_binary_search_empty_list(): - network = Network("192.168.1.1/24") - assert binary_search_for_insertion_index(network, []) == 0 - - -def test_binary_search_insert_at_beginning(): - networks = [ - Network("192.168.2.1/24"), - Network("192.168.3.1/24"), - ] - new_network = Network("192.168.1.1/24") - assert binary_search_for_insertion_index(new_network, networks) == 0 - - -def test_binary_search_insert_at_end(): - networks = [ - Network("192.168.1.1/24"), - Network("192.168.2.1/24"), - ] - new_network = Network("192.168.3.1/24") - assert binary_search_for_insertion_index(new_network, networks) == 2 - - -def test_binary_search_insert_in_middle(): - networks = [ - Network("192.168.1.1/24"), - Network("192.168.3.1/24"), - ] - new_network = Network("192.168.2.1/24") - assert binary_search_for_insertion_index(new_network, networks) == 1 - - -def test_binary_search_mixed_ipv4_ipv6(): - networks = [ - Network("192.168.1.1/24"), - Network("::1/64"), - ] - new_network1 = Network("192.168.0.1/24") - assert binary_search_for_insertion_index(new_network1, networks) == 0 - new_network2 = Network("192.168.2.1/24") - assert binary_search_for_insertion_index(new_network2, networks) == 1 - new_network3 = Network("::0/64") - assert binary_search_for_insertion_index(new_network3, networks) == 1 - new_network4 = Network("::2/64") - assert binary_search_for_insertion_index(new_network4, networks) == 2 - - -def test_binary_search_mixed_multiple_networks(): - networks = [ - Network("192.168.1.1/24"), - Network("192.168.2.1/24"), - Network("::1/64"), - Network("::2/64"), - ] - new_network1 = Network("192.168.0.1/24") - assert binary_search_for_insertion_index(new_network1, networks) == 0 - new_network2 = Network("192.168.1.2/24") - assert binary_search_for_insertion_index(new_network2, networks) == 1 - new_network3 = Network("192.168.3.1/24") - assert binary_search_for_insertion_index(new_network3, networks) == 2 - new_network4 = Network("::0/64") - assert binary_search_for_insertion_index(new_network4, networks) == 2 - new_network5 = Network("::1.5/64") - assert binary_search_for_insertion_index(new_network5, networks) == 3 - new_network6 = Network("::3/64") - assert binary_search_for_insertion_index(new_network6, networks) == 4 diff --git a/aikido_zen/helpers/zen_internals.py b/aikido_zen/helpers/zen_internals.py index a706a6c51..bddef0bf7 100644 --- a/aikido_zen/helpers/zen_internals.py +++ b/aikido_zen/helpers/zen_internals.py @@ -6,30 +6,74 @@ import ctypes import os import platform +import sysconfig -from aikido_zen.helpers.get_lib_path import get_binary_path from aikido_zen.helpers.logging import logger +_TARGET = f"{platform.system().lower()}-{platform.machine().lower()}" + + +def _binary_path(): + os_name = platform.system().lower() + machine = platform.machine().lower() + is_arm = "arm64" in machine or "aarch64" in machine + file_name = "libzen_internals_" + + # On macOS, platform.machine() returns "arm64" for Apple Silicon + # On Linux, platform.machine() returns "aarch64" for ARM64 + if is_arm: + file_name += "aarch64-" + # On macOS, platform.machine() returns "x86_64" for Intel + # On Linux, platform.machine() returns "x86_64" for AMD64 + elif "x86_64" in machine or "amd64" in machine: + file_name += "x86_64-" + + if os_name == "windows": + file_name += "pc-windows-msvc.dll" if is_arm else "pc-windows-gnu.dll" + elif os_name == "darwin": + file_name += "apple-darwin.dylib" + elif os_name in ["linux", "linux2"]: + # The library must use the same libc as this Python, which records it at build + # time (for example "x86_64-pc-linux-musl"). platform.libc_ver() and files under + # /lib describe the system, which can have both glibc and musl installed. + host = sysconfig.get_config_var("HOST_GNU_TYPE") or "" + libc = "musl" if "musl" in host else "gnu" + file_name += f"unknown-linux-{libc}.so" + + return os.path.join( + os.path.dirname(os.path.abspath(__file__)), "../libs", file_name + ) + + def _load(): - binary_path = get_binary_path() - target = f"{platform.system().lower()}-{platform.machine().lower()}" + binary_path = _binary_path() if not os.path.exists(binary_path): logger.warning( - "Zen will NOT block SQL injection attacks. Cannot find the zen-internals library for %s. " + "Cannot find the zen-internals library for %s. " "Request support: https://github.com/AikidoSec/firewall-python/issues", - target, + _TARGET, ) return None try: return ctypes.CDLL(binary_path) except OSError as e: - logger.warning( - "Zen will NOT block SQL injection attacks. Failed to load the zen-internals library for %s: %s", - target, - e, - ) - return None + error = e + # A musl Python built without HOST_GNU_TYPE picks the glibc build, which musl + # cannot load. + musl_path = os.path.join( + os.path.dirname(binary_path), + os.path.basename(binary_path).replace("-gnu.so", "-musl.so"), + ) + if musl_path != binary_path and os.path.exists(musl_path): + try: + return ctypes.CDLL(musl_path) + except OSError: + pass + logger.warning( + "Failed to load the zen-internals library for %s: %s", _TARGET, error + ) + return None def _bind(internals_lib, name, argtypes, restype): @@ -39,7 +83,9 @@ def _bind(internals_lib, name, argtypes, restype): try: function = getattr(internals_lib, name) except AttributeError: - logger.debug("The zen-internals library has no function %s", name) + logger.warning( + "The zen-internals library for %s has no %s function.", _TARGET, name + ) return None function.argtypes = argtypes function.restype = restype @@ -60,6 +106,8 @@ def _bind(internals_lib, name, argtypes, restype): ], ctypes.c_int, ) +if _detect_sql_injection is None: + logger.warning("Zen will NOT block SQL injection attacks.") def detect_sql_injection(query: bytes, user_input: bytes, dialect: int) -> int: @@ -74,3 +122,53 @@ def detect_sql_injection(query: bytes, user_input: bytes, dialect: int) -> int: return _detect_sql_injection( query_buffer, len(query), user_input_buffer, len(user_input), dialect ) + + +_ip_matcher_create = _bind( + _internals_lib, + "ip_matcher_create", + [ctypes.c_char_p, ctypes.c_size_t], + ctypes.c_void_p, +) +_ip_matcher_has = _bind( + _internals_lib, + "ip_matcher_has", + [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_size_t], + ctypes.c_int, +) +_ip_matcher_free = _bind(_internals_lib, "ip_matcher_free", [ctypes.c_void_p], None) +_ip_matcher_rejected_count = _bind( + _internals_lib, "ip_matcher_rejected_count", [ctypes.c_void_p], ctypes.c_size_t +) + + +def has_ip_matcher() -> bool: + return None not in (_ip_matcher_create, _ip_matcher_has, _ip_matcher_free) + + +if not has_ip_matcher(): + logger.warning("Zen will NOT apply IP lists or block SSRF attacks.") + + +def ip_matcher_create(networks: bytes): + """Builds a matcher from newline-separated networks. Returns None on failure.""" + # A handle is only useful when has and free exist too, so ip_matcher_has and + # ip_matcher_free never need their own check. + if not has_ip_matcher(): + return None + return _ip_matcher_create(networks, len(networks)) + + +def ip_matcher_has(handle, network: bytes) -> bool: + return _ip_matcher_has(handle, network, len(network)) == 1 + + +def ip_matcher_free(handle) -> None: + _ip_matcher_free(handle) + + +def ip_matcher_rejected_count(handle) -> int: + """Returns how many networks the library could not parse, or 0 when not loaded.""" + if _ip_matcher_rejected_count is None: + return 0 + return _ip_matcher_rejected_count(handle) diff --git a/aikido_zen/helpers/zen_internals_test.py b/aikido_zen/helpers/zen_internals_test.py index 483e532b2..205782dac 100644 --- a/aikido_zen/helpers/zen_internals_test.py +++ b/aikido_zen/helpers/zen_internals_test.py @@ -1,4 +1,10 @@ import ctypes +import os +import shutil +import subprocess +import sys + +import pytest from aikido_zen.helpers import zen_internals @@ -37,30 +43,80 @@ def test_detect_sql_injection_returns_error_when_function_is_not_loaded(monkeypa def test_load_returns_none_when_library_file_is_missing(monkeypatch, caplog): monkeypatch.setattr( - "aikido_zen.helpers.zen_internals.get_binary_path", + "aikido_zen.helpers.zen_internals._binary_path", lambda: "/nonexistent/libzen_internals.so", ) assert zen_internals._load() is None - assert "Zen will NOT block SQL injection attacks" in caplog.text assert "Cannot find the zen-internals library" in caplog.text + assert "Request support" in caplog.text def test_load_returns_none_when_library_fails_to_load(monkeypatch, caplog): monkeypatch.setattr( - "aikido_zen.helpers.zen_internals.get_binary_path", lambda: __file__ + "aikido_zen.helpers.zen_internals._binary_path", lambda: __file__ ) assert zen_internals._load() is None - assert "Zen will NOT block SQL injection attacks" in caplog.text assert "Failed to load the zen-internals library" in caplog.text +def test_load_falls_back_to_the_musl_build_when_the_glibc_build_fails( + monkeypatch, tmp_path +): + libs = tmp_path / "app-gnu.so" / "libs" + libs.mkdir(parents=True) + gnu_path = libs / "libzen_internals_x86_64-unknown-linux-gnu.so" + gnu_path.write_text("not a library") + shutil.copy( + zen_internals._binary_path(), libs / gnu_path.name.replace("-gnu", "-musl") + ) + monkeypatch.setattr( + "aikido_zen.helpers.zen_internals._binary_path", lambda: str(gnu_path) + ) + + assert zen_internals._load() is not None + + +def test_ip_matcher_round_trip(): + handle = zen_internals.ip_matcher_create(b"10.0.0.0/8\n2001:db8::/32\nfoobar") + assert handle is not None + assert zen_internals.ip_matcher_has(handle, b"10.1.2.3") is True + assert zen_internals.ip_matcher_has(handle, b"2001:db8::1") is True + assert zen_internals.ip_matcher_has(handle, b"8.8.8.8") is False + assert zen_internals.ip_matcher_rejected_count(handle) == 1 + zen_internals.ip_matcher_free(handle) + + +def test_ip_matcher_rejected_count_returns_0_when_function_is_not_loaded( + monkeypatch, +): + handle = zen_internals.ip_matcher_create(b"foobar") + monkeypatch.setattr( + "aikido_zen.helpers.zen_internals._ip_matcher_rejected_count", None + ) + + assert zen_internals.ip_matcher_rejected_count(handle) == 0 + zen_internals.ip_matcher_free(handle) + + +@pytest.mark.parametrize( + "function", ["_ip_matcher_create", "_ip_matcher_has", "_ip_matcher_free"] +) +def test_ip_matcher_create_returns_none_when_function_is_not_loaded( + monkeypatch, function +): + monkeypatch.setattr(f"aikido_zen.helpers.zen_internals.{function}", None) + + assert zen_internals.has_ip_matcher() is False + assert zen_internals.ip_matcher_create(b"10.0.0.0/8") is None + + def test_bind_returns_none_when_library_is_not_loaded(): assert zen_internals._bind(None, "detect_sql_injection", [], ctypes.c_int) is None -def test_bind_returns_none_when_function_is_missing(): +def test_bind_returns_none_when_function_is_missing(caplog): class LibraryWithoutFunctions: pass @@ -70,3 +126,110 @@ class LibraryWithoutFunctions: ) is None ) + assert "has no detect_sql_injection function" in caplog.text + + +def _warnings_at_import(cdll_patch): + script = ( + "import ctypes\n" f"{cdll_patch}\n" "import aikido_zen.helpers.zen_internals\n" + ) + result = subprocess.run( + [sys.executable, "-c", script], capture_output=True, text=True, check=True + ) + return result.stdout + result.stderr + + +def test_warns_once_per_disabled_feature_when_library_fails_to_load(): + output = _warnings_at_import( + "def fail(*args, **kwargs):\n" + " raise OSError('cannot open shared object file')\n" + "ctypes.CDLL = fail" + ) + + assert "Failed to load the zen-internals library" in output + assert "cannot open shared object file" in output + assert "Zen will NOT block SQL injection attacks." in output + assert "Zen will NOT apply IP lists or block SSRF attacks." in output + + +def test_warns_about_the_missing_function_when_library_is_loaded(): + output = _warnings_at_import( + "real_cdll = ctypes.CDLL\n" + "class WithoutSql:\n" + " def __init__(self, lib):\n" + " self._lib = lib\n" + " def __getattr__(self, name):\n" + " if name == 'detect_sql_injection':\n" + " raise AttributeError(name)\n" + " return getattr(self._lib, name)\n" + "ctypes.CDLL = lambda path, *args, **kwargs: WithoutSql(real_cdll(path))" + ) + + assert "has no detect_sql_injection function" in output + assert "Zen will NOT block SQL injection attacks." in output + assert "Failed to load the zen-internals library" not in output + + +@pytest.mark.parametrize( + "system, machine, host, file_name", + [ + ("Darwin", "arm64", "", "libzen_internals_aarch64-apple-darwin.dylib"), + ("Darwin", "x86_64", "", "libzen_internals_x86_64-apple-darwin.dylib"), + ( + "Linux", + "aarch64", + "aarch64-unknown-linux-gnu", + "libzen_internals_aarch64-unknown-linux-gnu.so", + ), + ( + "Linux", + "x86_64", + "x86_64-pc-linux-gnu", + "libzen_internals_x86_64-unknown-linux-gnu.so", + ), + ( + "Linux", + "aarch64", + "aarch64-alpine-linux-musl", + "libzen_internals_aarch64-unknown-linux-musl.so", + ), + ( + "Linux", + "x86_64", + "x86_64-pc-linux-musl", + "libzen_internals_x86_64-unknown-linux-musl.so", + ), + ("Linux", "x86_64", None, "libzen_internals_x86_64-unknown-linux-gnu.so"), + ("Windows", "AMD64", None, "libzen_internals_x86_64-pc-windows-gnu.dll"), + ("Windows", "ARM64", None, "libzen_internals_aarch64-pc-windows-msvc.dll"), + ], +) +def test_binary_path_picks_the_build_for_the_platform( + monkeypatch, system, machine, host, file_name +): + monkeypatch.setattr("platform.system", lambda: system) + monkeypatch.setattr("platform.machine", lambda: machine) + monkeypatch.setattr("sysconfig.get_config_var", lambda name: host) + + path = zen_internals._binary_path() + + assert os.path.basename(path) == file_name + assert os.path.dirname(os.path.normpath(path)).endswith( + os.path.join("aikido_zen", "libs") + ) + + +def test_binary_path_ignores_what_the_system_reports_about_libc(monkeypatch): + monkeypatch.setattr("platform.system", lambda: "Linux") + monkeypatch.setattr("platform.machine", lambda: "x86_64") + monkeypatch.setattr("sysconfig.get_config_var", lambda name: "x86_64-pc-linux-gnu") + monkeypatch.setattr("platform.libc_ver", lambda: ("", "")) + monkeypatch.setattr("glob.glob", lambda pattern: ["/lib/ld-musl-x86_64.so.1"]) + + assert os.path.basename(zen_internals._binary_path()) == ( + "libzen_internals_x86_64-unknown-linux-gnu.so" + ) + + +def test_binary_path_exists_for_this_machine(): + assert os.path.exists(zen_internals._binary_path()) diff --git a/aikido_zen/storage/firewall_lists.py b/aikido_zen/storage/firewall_lists.py index 9381e82c7..64f631790 100644 --- a/aikido_zen/storage/firewall_lists.py +++ b/aikido_zen/storage/firewall_lists.py @@ -1,4 +1,6 @@ +from aikido_zen.helpers import zen_internals from aikido_zen.helpers.ip_matcher import IPMatcher +from aikido_zen.helpers.logging import logger import regex as re from aikido_zen.helpers.net.is_private_ip import is_private_ip @@ -10,6 +12,33 @@ def __init__(self): self.allowed_ips = [] self.blocked_user_agent_regex = None + def update(self, blocked_ip_entries, allowed_ip_entries, blocked_user_agents): + """Replaces all lists together, so lists from different updates never mix.""" + blocked_ips = list(map(parse_ip_entry, blocked_ip_entries)) + # An allowlist without IPs would block every visitor, so it is skipped. + allowed_ips = [ + entry + for entry in map(parse_ip_entry, allowed_ip_entries) + if not entry["iplist"].is_empty() + ] + # Aikido only sends valid networks, so an invalid one means the data is broken. + for entry in blocked_ips + allowed_ips: + rejected_count = entry["iplist"].rejected_count() + if rejected_count: + logger.error( + "IP list %s has %s invalid network(s), which Zen skipped.", + entry["source"], + rejected_count, + ) + # An allowlist that failed to build would block the visitors it should allow. + if not all(entry["iplist"].is_built() for entry in allowed_ips): + if zen_internals.has_ip_matcher(): + return + allowed_ips = [] + self.blocked_ips = blocked_ips + self.allowed_ips = allowed_ips + self.set_blocked_user_agents(blocked_user_agents) + def set_blocked_ips(self, blocked_ip_entries): self.blocked_ips = list(map(parse_ip_entry, blocked_ip_entries)) @@ -56,7 +85,7 @@ def parse_ip_entry(entry): Converts ip entry: {"source": "example", "description": "Example description", "ips": []} """ return { - "source": entry["source"], - "description": entry["description"], - "iplist": IPMatcher(entry["ips"]), + "source": entry.get("source"), + "description": entry.get("description") or "a firewall list", + "iplist": IPMatcher(entry.get("ips")), } diff --git a/aikido_zen/vulnerabilities/sql_injection/init_test.py b/aikido_zen/vulnerabilities/sql_injection/init_test.py index d017bc442..e13ec4b3a 100644 --- a/aikido_zen/vulnerabilities/sql_injection/init_test.py +++ b/aikido_zen/vulnerabilities/sql_injection/init_test.py @@ -1,9 +1,11 @@ import os import pytest +from aikido_zen.helpers import zen_internals from aikido_zen.vulnerabilities.sql_injection import ( detect_sql_injection, should_return_early, ) +from .map_dialect_to_rust_int import map_dialect_to_rust_int """Removed tests : -> `I'm writting you` : Invalid SQL @@ -86,6 +88,17 @@ def is_not_sql_injection(sql, input, dialect="all"): ), f"Expected no SQL injection for SQL: {sql} and input: {input} in {current} dialect" +def is_tokenize_error(sql, input, dialect): + result = zen_internals.detect_sql_injection( + sql.lower().encode(), + input.lower().strip().encode(), + map_dialect_to_rust_int(dialect), + ) + assert ( + result == 3 + ), f"Expected tokenize error for SQL: {sql} and input: {input} in {dialect} dialect" + + def test_should_return_early(): # Test cases where the function should return True @@ -218,7 +231,7 @@ def test_check_string_safely_escaped(): ) # Positive example of same query : is_sql_injection( - "SELECT * FROM comments WHERE comment = 'I'm writting you--'", + "SELECT * FROM comments WHERE comment = 'I'm writting you-- '", "I'm writting you--", ) is_sql_injection( @@ -415,11 +428,29 @@ def test_function_calls_as_sql_injections(): def test_trimmed_user_input_bypass(): is_sql_injection( - "INSERT INTO pets (name, owner) VALUES ('x', 'dummy'), ('injected', 'hacker'); --', 'owner')", + "INSERT INTO pets (name, owner) VALUES ('x', 'dummy'), ('injected', 'hacker'); -- ', 'owner')", "x', 'dummy'), ('injected', 'hacker'); -- ", ) +def test_mysql_does_not_detect_dash_dash_without_space(): + for sql, input in [ + ( + "SELECT * FROM comments WHERE comment = 'I'm writting you--'", + "I'm writting you--", + ), + ( + "INSERT INTO pets (name, owner) VALUES ('x', 'dummy'), ('injected', 'hacker'); --', 'owner')", + "x', 'dummy'), ('injected', 'hacker'); -- ", + ), + ]: + is_tokenize_error(sql, input, "mysql") + is_not_sql_injection(sql, input, "mysql") + is_sql_injection(sql, input, "generic") + is_sql_injection(sql, input, "postgres") + is_sql_injection(sql, input, "sqlite") + + def test_returns_false_when_library_cannot_be_loaded(monkeypatch): monkeypatch.setattr("aikido_zen.helpers.zen_internals._detect_sql_injection", None) diff --git a/benchmarks/ip_matcher_benchmark/ip_matcher_benchmark.py b/benchmarks/ip_matcher_benchmark/ip_matcher_benchmark.py index c9a08781d..8ec737582 100644 --- a/benchmarks/ip_matcher_benchmark/ip_matcher_benchmark.py +++ b/benchmarks/ip_matcher_benchmark/ip_matcher_benchmark.py @@ -10,12 +10,8 @@ def generate_random_ipv4_networks(n): def generate_random_ipv6_networks(n): return [f"{random.randint(0, 65535):04x}:{random.randint(0, 65535):04x}:{random.randint(0, 65535):04x}:{random.randint(0, 65535):04x}:{random.randint(0, 65535):04x}:{random.randint(0, 65535):04x}:{random.randint(0, 65535):04x}:{random.randint(0, 65535):04x}/{random.randint(16, 128)}" for _ in range(n)] -def benchmark_add(networks): - matcher = IPMatcher() - def add_networks(): - for net in networks: - matcher._add(net) - return timeit.timeit(add_networks, number=1) +def benchmark_build(networks): + return timeit.timeit(lambda: IPMatcher(networks), number=1) def benchmark_has(matcher, networks): def check_networks(): @@ -28,8 +24,8 @@ def check_networks(): ipv6_networks = generate_random_ipv6_networks(1000) # Benchmarking -ipv4_add_time = benchmark_add(ipv4_networks) -ipv6_add_time = benchmark_add(ipv6_networks) +ipv4_build_time = benchmark_build(ipv4_networks) +ipv6_build_time = benchmark_build(ipv6_networks) ipv4_matcher = IPMatcher(ipv4_networks) ipv6_matcher = IPMatcher(ipv6_networks) @@ -37,7 +33,7 @@ def check_networks(): ipv4_has_time = benchmark_has(ipv4_matcher, ipv4_networks) ipv6_has_time = benchmark_has(ipv6_matcher, ipv6_networks) -print(f"IPv4 Add Time: {ipv4_add_time:.6f} seconds") +print(f"IPv4 Build Time: {ipv4_build_time:.6f} seconds") print(f"IPv4 Has Time: {ipv4_has_time:.6f} seconds") -print(f"IPv6 Add Time: {ipv6_add_time:.6f} seconds") +print(f"IPv6 Build Time: {ipv6_build_time:.6f} seconds") print(f"IPv6 Has Time: {ipv6_has_time:.6f} seconds") diff --git a/poetry.lock b/poetry.lock index a8bc4f57e..61bc8ef6d 100644 --- a/poetry.lock +++ b/poetry.lock @@ -3045,18 +3045,6 @@ files = [ [package.dependencies] six = ">=1.5" -[[package]] -name = "pytricia" -version = "1.3.0" -description = "An efficient IP address storage and lookup module for Python." -optional = false -python-versions = "*" -groups = ["main"] -markers = "sys_platform != \"win32\"" -files = [ - {file = "pytricia-1.3.0.tar.gz", hash = "sha256:1c3a3d6909e10d4c9c2f0fe4542a2481e109d29aab99cc027ca7fe93f8c8853f"}, -] - [[package]] name = "pytz" version = "2025.2" @@ -4238,4 +4226,4 @@ cffi = ["cffi (>=1.17,<2.0) ; platform_python_implementation != \"PyPy\" and pyt [metadata] lock-version = "2.1" python-versions = ">=3.8,<3.15" -content-hash = "c77b3d119232b6bf287feec1ce02bee84defa04c192f3d5e2a7179358591a3ef" +content-hash = "44d57ee8927d1a9884cac495447da931913a330ed272a2bc4a1f4cac659a8594" diff --git a/pyproject.toml b/pyproject.toml index 7ea7eb753..2f85776d5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -71,7 +71,6 @@ wrapt = [ { version = "^1.17.2", python = "<3.13" }, { version = "^2.4.0", python = ">=3.13" }, ] -pytricia = { version = "^1.3.0", markers = "sys_platform != 'win32'" } [tool.poetry.group.dev.dependencies] black = "^24.4.2"