host_firewall: ban firehol level2, bruteforceblocker, vxvault and cybercrime; greensnow is part of level2

host_firewall/files/nft-host-firehol

@@ -1,9 +1,12 @@
 #!/usr/bin/python3
 # Managed by Ansible (role host_firewall), do not edit.
-"""Rebuilds the FireHOL level1 set firehol_v4 of table inet host.
+"""Rebuilds the FireHOL sets of table inet host (IPv4 only, FireHOL has no IPv6 lists).
 
-A failed download or sanity check exits non-zero before anything changes, so
-the previous set stays in place.
+Every set is built from one or more lists of the FireHOL blocklist-ipsets
+(single addresses or CIDR networks, comment lines start with #). All sets go
+into one file, so one reload replaces them together. A failed download or
+sanity check of any list exits non-zero before anything changes, so the
+previous sets stay in place.
 """
 import ipaddress
 import json
@@ -16,11 +19,18 @@ import time
 import urllib.request
 
 PERSISTED = "/etc/nftables.d/host-firehol.nft"
-URL = "https://raw.githubusercontent.com/firehol/blocklist-ipsets/master/firehol_level1.netset"
+BASE_URL = "https://raw.githubusercontent.com/firehol/blocklist-ipsets/master/"
 USER_AGENT = "ruekov-host-firewall (nft-host-firehol)"
 
-# Lower bound that catches an empty or truncated answer.
-MIN_ENTRIES = 1000
+# set name -> (lists merged into it, lower bound for their entries together that
+# catches an empty or truncated answer; the lists hold about 4700 (level1),
+# 18700 (level2), 500 (bruteforceblocker) and 70 + 500 (vxvault, cybercrime)).
+SETS = {
+    "firehol_v4": (["firehol_level1.netset"], 1000),
+    "firehol_level2_v4": (["firehol_level2.netset"], 5000),
+    "bruteforceblocker_v4": (["bruteforceblocker.ipset"], 100),
+    "malware_v4": (["vxvault.ipset", "cybercrime.ipset"], 200),
+}
 # A prefix this wide is a parse error, not a bad network.
 MIN_PREFIXLEN = 8
 # Private, CGNAT (NetBird), loopback, link-local and multicast/reserved space:
@@ -36,17 +46,18 @@ def fail(message):
     sys.exit(1)
 
 
-def fetch():
+def fetch(name):
+    url = BASE_URL + name
     error = None
     for attempt in range(3):
         try:
-            request = urllib.request.Request(URL, headers={"User-Agent": USER_AGENT})
+            request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
             with urllib.request.urlopen(request, timeout=120) as response:
                 return response.read().decode("utf-8")
         except OSError as e:
             error = e
             time.sleep(5 * (attempt + 1))
-    fail(f"download of {URL} failed: {error}")
+    fail(f"download of {url} failed: {error}")
 
 
 def own_addresses():
@@ -57,20 +68,20 @@ def own_addresses():
             for a in link.get("addr_info", []) if a.get("family") == "inet"]
 
 
-def main():
+def build_set(lists, minimum, own):
     networks = []
-    for line in fetch().splitlines():
-        line = line.strip()
-        if not line or line.startswith("#"):
-            continue
-        try:
-            networks.append(ipaddress.IPv4Network(line))
-        except ValueError:
-            fail(f"unparsable line: {line!r}")
-    if len(networks) < MIN_ENTRIES:
-        fail(f"only {len(networks)} entries")
+    for name in lists:
+        for line in fetch(name).splitlines():
+            line = line.strip()
+            if not line or line.startswith("#"):
+                continue
+            try:
+                networks.append(ipaddress.IPv4Network(line))
+            except ValueError:
+                fail(f"unparsable line in {name}: {line!r}")
+    if len(networks) < minimum:
+        fail(f"only {len(networks)} entries in {', '.join(lists)}")
 
-    own = own_addresses()
     kept, excluded, own_skipped = [], 0, 0
     for network in networks:
         if any(network.overlaps(e) for e in EXCLUDED):
@@ -82,21 +93,28 @@ def main():
                 fail(f"suspiciously wide prefix {network}")
             kept.append(network)
     merged = list(ipaddress.collapse_addresses(kept))
-    print(f"firehol_v4: {len(networks)} entries, {excluded} in private or reserved space and "
-          f"{own_skipped} holding an own address skipped -> {len(merged)} merged")
+    return networks, excluded, own_skipped, merged
 
-    elements = ",\n\t\t".join(str(n) for n in merged) + ","
+
+def main():
+    own = own_addresses()
+    definitions = []
+    for set_name, (lists, minimum) in SETS.items():
+        networks, excluded, own_skipped, merged = build_set(lists, minimum, own)
+        print(f"{set_name}: {len(networks)} entries, {excluded} in private or reserved space and "
+              f"{own_skipped} holding an own address skipped -> {len(merged)} merged")
+        elements = ",\n\t\t".join(str(n) for n in merged) + ","
+        definitions.append(f"\tset {set_name} {{ type ipv4_addr; flags interval; auto-merge; elements = {{\n"
+                           f"\t\t{elements}\n\t}} }}\n")
     persisted = ("# Generated by nft-host-firehol, do not edit.\n"
-                 "table inet host {\n"
-                 "\tset firehol_v4 { type ipv4_addr; flags interval; auto-merge; elements = {\n"
-                 f"\t\t{elements}\n\t}} }}\n}}\n")
+                 "table inet host {\n" + "".join(definitions) + "}\n")
 
     # Checked under another table name: a dry run against the live table
     # name changes the live sets for its duration and unlisted addresses match.
     check_rules = persisted.replace("table inet host {", "table inet host_firehol_check {")
     checked = subprocess.run(["nft", "-c", "-f", "-"], input=check_rules, capture_output=True, text=True)
     if checked.returncode != 0:
-        fail(f"nft rejects the set: {checked.stderr.strip()}")
+        fail(f"nft rejects the sets: {checked.stderr.strip()}")
 
     with tempfile.TemporaryDirectory() as tmp:
         previous = None