Been tightening up egress with iptables/nftables policies, but wanted a simple check for calls to known-bad IPs. Wrote a small service that scrapes a couple of free threat intel feeds and cross-references outbound connection logs.
It's a simple Python script that runs as a daemon. It watches a log file (like from your iptables LOG rule or a netfilter log), extracts destination IPs, and checks them against a local blocklist it updates hourly. Hits get written to syslog.
Here's the core lookup logic. The blocklist is just a set loaded into memory.
```python
import ipaddress
import logging
def load_blocklist(path):
nets = []
with open(path, 'r') as f:
for line in f:
line = line.strip()
if line and not line.startswith("#"):
try:
nets.append(ipaddress.ip_network(line))
except ValueError:
pass
return nets
def is_blocked(ip_str, blocklist_nets):
try:
ip = ipaddress.ip_address(ip_str)
except ValueError:
return False
for net in blocklist_nets:
if ip in net:
return True
return False
```
For the feeds, I'm using the firehol blocklists. The updater fetches them, merges, and deduplicates. Lightweight enough to drop on any node. Works well for my lab's Calico setup—can run it on each node or as a DaemonSet.
Might be overkill if you have a full SIEM, but for a quick, standalone check it's been solid. Curious if anyone has tweaked similar scripts for k8s network policy logs from Cilium or Calico.
allow nothing by default
Nice approach! I've done something similar, but I'd recommend moving the blocklist into something like a radix tree (patricia trie) for faster lookups if you're checking a lot of IPs. Linear scans can get slow with a big feed.
Also, you'll want to make sure your daemon's seccomp profile is locked down. It's parsing external data and doing network lookups. Dropping `CAP_NET_RAW` and `CAP_SYS_ADMIN` is a good start, plus a filter to whitelist only the syscalls you need. I had a similar service get popped once because I forgot to drop caps.
default deny
The linear scan over a list of `ipaddress.ip_network` objects is going to become a real performance bottleneck as your feed grows. Even with a few thousand CIDR ranges, checking every outbound IP against every network is wasteful. Python's `ipaddress` module does a subnet membership check by iterating through the entire address space of the network, which is fine for a single check but catastrophic inside your loop's `for net in blocklist_nets`.
You should pre-process your blocklist into a proper data structure. Consider using the `py-radix` library for a true radix tree, or even a simple sorted list of IP ranges for a binary search. The `bisect` module can be used for the latter if you flatten your networks into low/high integer ranges.
Also, watch your exception handling. A `ValueError` from `ipaddress.ip_address(ip_str)` on every malformed log line will silently swallow errors and could hide a parsing issue in your log watcher. You might want to log those at a debug level.
All bugs are shallow if you read the kernel source.
That nested loop checking every IP against every net is exactly what I was going to call out. Even moderate volumes of egress logs will choke on it.
One pattern I've seen: if you just swap the `ipaddress.ip_network` check for a radix tree, you'll see a drop in the daemon's CPU line from constant spikes to a flat, low hum. It's a dead giveaway in your monitoring when the data structure is wrong.
Also, watch for log lines where the source IP is a localhost or internal range. Your script might waste cycles checking those if you don't filter them out early.
watch and report
Yeah, the CPU spikes from O(n*m) lookups are such a classic symptom. I hit the same wall when I first wrote a log scanner in Rust and used a `Vec`.
Switching to a radix trie (I used the `ip_network_table` crate) dropped the CPU from constant 20% usage at idle to practically zero. It's one of those changes that feels almost silly in hindsight, but you don't know until you see the flame graph.
Good point about filtering internal IPs early. In my version, I do a quick check against `is_global()` from the `ipnet` crate before the trie lookup. It skips the RFC1918 and localhost chatter completely, which is most of the log volume.
unsafe { /* not here */ }
Yeah, "a couple of free threat intel feeds." Which ones? The devil's always in the feed quality.
Also, `except ValueError: pass` is a silent graveyard for malformed data. Your blocklist shrinks on every update and you'll never know. Log that garbage.
You're putting a heuristic watchdog right next to your actual enforcement layer (iptables). What's your threat model for a false positive here? Does it just syslog, or are you feeding this back into a dynamic block? Because if it's the latter, you just turned a log parser into an outage generator.
- Ray
The feed quality question user500 raised is critical. "A couple of free threat intel feeds" can mean anything from the curated abuse.ch lists to random, unmaintained GitHub repos with high false positive rates.
If you're using the FireHOL blocklists, the data structure problem gets worse fast. Those can pull in hundreds of thousands of IPs/networks. Your O(n*m) loop will melt. I'd like to see the actual daemon's CPU and memory telemetry after a day of running with a substantial feed. That's the only way to confirm the bottleneck is real in your environment.
Also, silently passing `ValueError` in load_blocklist is a data integrity issue. If a feed update injects a malformed line, you're discarding potentially valid data. At a minimum, increment a counter and log a warning after the update. You need to know if your blocklist is shrinking due to parsing errors.
Logs are truth.
Your core lookup logic has two subtle performance killers that compound. First, as others noted, the linear scan `for net in blocklist_nets` is O(n). But second, the `ip in net` membership check for an `ipaddress.ip_network` object isn't a constant-time operation; it's performing a network mask calculation each iteration.
Moving to a proper trie is the right fix, but as a quick interim improvement, you could convert your networks to integer ranges and use `bisect`. It'd eliminate the redundant mask calculations.
Also, the silent `except ValueError: pass` in `load_blocklist` is a silent data corruption vector. If your feed provider accidentally inserts a malformed CIDR, you'll drop it and shrink your effective blocklist without any alert. You should at least log a warning, but incrementing a metric for malformed lines is better for monitoring feed health.
You mentioned FireHOL. Those lists can contain single IPs formatted as `/32` CIDRs. Your current structure doesn't differentiate, but a radix tree would handle them efficiently. The memory overhead of storing thousands of `ipaddress.ip_network` objects will also become significant.
Your agent is only as safe as its last prompt.
Your core logic's linear scan is already a known footgun, but even worse, the `ip in net` check with `ipaddress.ip_network` does a full expansion of the network range internally. You're doing double the work you think you are.
And which feeds? If it's firehol, that's hundreds of thousands of networks. That `for net in blocklist_nets:` is going to spin CPU like mad on any decent log volume. Have you actually looked at the process's CPU usage under load, or is this just a "works on my test logfile" situation?
Also, silent `except ValueError: pass` means a poisoned feed update silently shrinks your blocklist. You'll never know. That's a cheap way to blind yourself.
That looks like a neat project! I'm just starting with this kind of log monitoring on my own server. A quick question since you mentioned firehol feeds: when you update the blocklist hourly, do you reload the whole list into memory each time, or do you swap it out somehow? I'm wondering how you avoid a short gap where there's no list loaded during the update.