Repository navigation
Use the zen-internals IP matcher - #757
iacobdaniel wants to merge 17 commits into
Conversation
| 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()) |
There was a problem hiding this comment.
🟠 High - Missing native matcher disables SSRF and IP-list enforcement
When the bundled zen-internals library is missing or cannot be loaded, IPMatcher has no handle and its has() method returns false for every address. The static IMDS/private-IP matchers and configured global or route IP lists then produce no matches, allowing DNS resolutions to metadata/private addresses and bypassing configured IP blocklists and allowlists.
Show fix
Retain a functional pure-Python matcher fallback for load failures, or fail closed and make the application refuse to enable these security controls unless all required native matcher symbols are available. Ensure every supported deployment target ships and loads the matching artifact before accepting IP-based security configuration.
More info - Reply on this comment to give feedback or ignore the issue.
There was a problem hiding this comment.
On platforms that have no zen-internals build (for example 32-bit Linux), Zen fails open and logs a warning, the same way SQL injection detection already behaves there on main.
There was a problem hiding this comment.
🔴 We were not able to ignore this issue because of the following reason:
You do not have the permission to ignore issues.
|
Reminder: this points at the v0.1.75-beta.2 pre-release of zen-internals. Switch BASE_URL to the first stable release that contains the IP matcher. |
|
|
||
| endpoint["allowedIPAddresses"] = IPMatcher(endpoint["allowedIPAddresses"]) | ||
| allowed_ips = IPMatcher(endpoint["allowedIPAddresses"]) | ||
| if not allowed_ips.is_built(): |
There was a problem hiding this comment.
any way we can avoid this? e.g. use a function create_ip_matcher that returns None if no IPMatcher could be created?
There was a problem hiding this comment.
I tried it. create_ip_matcher() returning None removes the if for route allowlists, but every other place that uses a matcher then needs a None check: the blocklist, bypass list and both SSRF lists. That's 5 checks instead of 2, and any new caller has to remember it. So I would keep it as it is.
There was a problem hiding this comment.
Do those call sites need to check whether the IP matcher was built?
| [network.encode() for network in self._networks] | ||
| ) | ||
| if self._handle is not None: | ||
| weakref.finalize(self, zen_internals.ip_matcher_free, self._handle) |
There was a problem hiding this comment.
The finalizer may receive None as its callback when the library lacks ip_matcher_free; registering it can cause a runtime exception during finalization.
Details
✨ AI Reasoning
The create binding can be present while the free binding is absent. In that case, construction succeeds and registers a finalizer with a nullable callback, which can fail when the object is finalized.
🔧 How do I fix it?
Check values for null before use with guards, early returns, or null-coalescing. Declare nullable return types for methods that can return null, and avoid silencing nullable warnings with the null-forgiving operator (!) unless the value's non-nullness is guaranteed.
Reply on this comment to give feedback or ignore the issue.
More info
There was a problem hiding this comment.
Fixed. Without ip_matcher_free, ip_matcher_create returns None.
| for network in networks or [] | ||
| if isinstance(network, str) and network.isascii() | ||
| ] | ||
| self._handle = zen_internals.ip_matcher_create( |
There was a problem hiding this comment.
libzen v0.1.75-beta.2 adds ip_matcher_rejected_count, which allows you to check how many networks were rejected at creation. This allows fine-grained decisions about whether creation succeeded.
Networks data provided to Zen by Aikido are expected to be well-formed so the rejection of any networks is considered an exception. A tolerable rejection rate may be preferable, depending on the source of the networks data.
There was a problem hiding this comment.
Thanks, added it. Invalid networks are now logged with the list they came from, and the valid ones still apply. The lists are also updated together now, so updates never mix.
| binary_path = get_binary_path() | ||
| target = f"{platform.system().lower()}-{platform.machine().lower()}" | ||
| binary_path = _binary_path() | ||
| if not os.path.exists(binary_path): |
There was a problem hiding this comment.
Should we also check whether the musl_path file exists here, to avoid returning early if a musl Python built without HOST_GNU_TYPE picks the glibc build, or are the musl and glibc builds of the libzen included together?
There was a problem hiding this comment.
No need: the glibc and musl builds are always shipped together, so if the glibc file is missing, the musl one is too. If the glibc file exists but fails to load, we already retry with the musl build.
| if not all(entry["iplist"].is_built() for entry in allowed_ips): | ||
| allowed_ips = [] | ||
| self.allowed_ips = allowed_ips |
There was a problem hiding this comment.
What if the allowed IPs are currently set? Should a currently failing attempt to set the allowed IPs override a previously successful attempt to set the allowed IPs by clearing the allowed IPs entirely?
There was a problem hiding this comment.
Right, good point. Pushed a fix to keep previous allowlist.
| def set_allowed_ips(self, allowed_ip_entries): | ||
| self.allowed_ips = list(map(parse_ip_entry, allowed_ip_entries)) | ||
| allowed_ips = list(map(parse_ip_entry, allowed_ip_entries)) | ||
| # An allowlist that failed to build would block the visitors it should allow. |
There was a problem hiding this comment.
An IP list that fails to build clears the allowed list. Does an empty allowed IPs list not block visitors that it should allow?
There was a problem hiding this comment.
No. An empty allowed list means allow mode is off, so all visitors are allowed (is_allowed_ip returns True when the list is empty).
| "ip_matcher_has", | ||
| [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_size_t], | ||
| ctypes.c_int, | ||
| ) |
There was a problem hiding this comment.
Add the ip_matcher_rejected_count binding, if you wish to use it.
| networks = list(networks or []) | ||
| self._networks = [ | ||
| network | ||
| for network in networks | ||
| if isinstance(network, str) and network.isascii() | ||
| ] | ||
| self._skipped_count = len(networks) - len(self._networks) |
There was a problem hiding this comment.
networks is rebound to a new list, obscuring the original input value during initialization.
Show fix
| networks = list(networks or []) | |
| self._networks = [ | |
| network | |
| for network in networks | |
| if isinstance(network, str) and network.isascii() | |
| ] | |
| self._skipped_count = len(networks) - len(self._networks) | |
| normalized_networks = list(networks or []) | |
| self._networks = [ | |
| network | |
| for network in normalized_networks | |
| if isinstance(network, str) and network.isascii() | |
| ] | |
| self._skipped_count = len(normalized_networks) - len(self._networks) |
Details
✨ AI Reasoning
The constructor first replaces the supplied networks value with a new list, then uses that name for subsequent processing. This obscures the distinction between the caller's input and the normalized local value; keeping the original parameter intact would make that distinction clearer.
Reply on this comment to give feedback or ignore the issue.
More info
There was a problem hiding this comment.
not relevant, the original value isn't needed after this line, so we just reuse the name
There was a problem hiding this comment.
⛔ You don't have permission to ignore issues.
| blocked_ips = list(map(parse_ip_entry, blocked_ip_entries)) | ||
| allowed_ips = list(map(parse_ip_entry, allowed_ip_entries)) |
There was a problem hiding this comment.
🟡 Medium - Malformed allowlist entry suppresses valid blocklist refresh
A successful firewall-list response with valid blocked IP entries but a malformed allowed-list entry causes FirewallLists.update to fail while parsing the allowlist (Trigger). The new atomic update builds both lists before assigning either, and the caller swallows the exception, so the valid blocklist is not installed (Mechanism). Requests from IPs that the response intended to block then pass the global blocklist until a later successful refresh, or indefinitely when this occurs during initial setup (Consequence).
Show fix
| blocked_ips = list(map(parse_ip_entry, blocked_ip_entries)) | |
| allowed_ips = list(map(parse_ip_entry, allowed_ip_entries)) | |
| blocked_ips = list(map(parse_ip_entry, blocked_ip_entries)) | |
| try: | |
| allowed_ips = list(map(parse_ip_entry, allowed_ip_entries)) | |
| except (KeyError, TypeError, ValueError) as e: | |
| logger.debug("Exception parsing allowed IP lists: %s", e) | |
| self.blocked_ips = blocked_ips | |
| self.set_blocked_user_agents(blocked_user_agents) | |
| return |
More info - Reply on this comment to give feedback or ignore the issue.
There was a problem hiding this comment.
Valid, fixed differently: a missing ips or description no longer raises, so the update still applies everything together.
| blocked_ips = list(map(parse_ip_entry, blocked_ip_entries)) | ||
| allowed_ips = list(map(parse_ip_entry, allowed_ip_entries)) |
There was a problem hiding this comment.
🟡 Medium - Missing IP field installs an empty global allowlist
A firewall-list response with an allowlist entry whose ips field is missing or null is accepted by parse_ip_entry. The resulting empty IPMatcher is still treated as a built, non-empty allowlist entry, so is_allowed_ip rejects every public client address; the same normalization on a malformed blocked entry can also replace a previously effective blocklist with a matcher that matches nothing.
Show fix
| blocked_ips = list(map(parse_ip_entry, blocked_ip_entries)) | |
| allowed_ips = list(map(parse_ip_entry, allowed_ip_entries)) | |
| blocked_ips = ( | |
| self.blocked_ips | |
| if any( | |
| not isinstance(entry, dict) | |
| or not isinstance(entry.get("ips"), list) | |
| for entry in blocked_ip_entries | |
| ) | |
| else list(map(parse_ip_entry, blocked_ip_entries)) | |
| ) | |
| allowed_ips = [ | |
| parse_ip_entry(entry) | |
| for entry in allowed_ip_entries | |
| if isinstance(entry, dict) and isinstance(entry.get("ips"), list) | |
| ] |
More info - Reply on this comment to give feedback or ignore the issue.
There was a problem hiding this comment.
Valid for the allowlist, fixed: allowlist entries without IPs are skipped, so they can't block every visitor.
Replaces pytricia and the Python fallback with the Rust matcher from zen-internals.