278 lines
9.8 KiB
Python
278 lines
9.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Whitelist management for the intrusion-detection system.
|
|
|
|
Owns provider CIDR constants, remote download/refresh logic, and
|
|
CIDR extraction/validation. Imported by `utils.py` which provides
|
|
the convenience lookups `whitelist_of()` and `is_whitelisted()`.
|
|
|
|
Whitelist loading order (per provider):
|
|
1. Local file in `data/<provider>/` (if present and readable)
|
|
2. Remote download (if network available)
|
|
3. Built-in hardcoded fallback
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import os
|
|
import re
|
|
from typing import Any
|
|
|
|
from .constants import (
|
|
CLOUDFLARE_URL_V4,
|
|
CLOUDFLARE_URL_V6,
|
|
CLOUDFLARE_V4_FILE,
|
|
CLOUDFLARE_V6_FILE,
|
|
PALOALTO_FILE,
|
|
PALOALTO_URL,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CIDR extraction regex
|
|
# ---------------------------------------------------------------------------
|
|
|
|
CIDR_RE = re.compile(
|
|
r"(?<![0-9A-Fa-f:.])([0-9A-Fa-f:.]+/\d{1,3})(?![0-9A-Fa-f:.])")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Local file paths (checked first, before remote download)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# (paths live in app/constants.py as CLOUDFLARE_V4_FILE,
|
|
# CLOUDFLARE_V6_FILE, PALOALTO_FILE; URLs as CLOUDFLARE_URL_V4, ...)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Built-in fallback CIDR lists
|
|
# ---------------------------------------------------------------------------
|
|
|
|
#: Init7 (Switzerland) Ltd. ranges — AS13030, verified via RIPE RDAP 2026-09-08.
|
|
#: Includes the Hybrid7 WAN /24 (router WAN 77.74.80.159), init7 DNS
|
|
#: infrastructure, and the PPPoE gateway. init7 publishes no geofeed, so
|
|
#: these are maintained here rather than downloaded.
|
|
INIT7_IPV4 = [
|
|
"77.74.80.0/24", # Hybrid7 WAN /24
|
|
"77.109.128.0/18", # init7 DNS / infrastructure
|
|
"213.144.128.0/19", # init7 DNS / infrastructure
|
|
"5.180.132.0/22", # init7 PPPoE gateway
|
|
]
|
|
|
|
#: init7 IPv6 allocation (contains delegated /48 2001:1620:549f::/48)
|
|
INIT7_IPV6 = ["2001:1620::/30"]
|
|
|
|
#: https://www.cloudflare.com/ips-v4
|
|
CLOUDFLARE_IPV4 = [
|
|
"173.245.48.0/20", "103.21.244.0/22", "103.22.200.0/22",
|
|
"103.31.4.0/22", "141.101.64.0/18", "108.162.192.0/18",
|
|
"190.93.240.0/20", "188.114.96.0/20", "197.234.240.0/22",
|
|
"198.41.128.0/17", "162.158.0.0/15", "104.16.0.0/13",
|
|
"104.24.0.0/14", "172.64.0.0/13", "131.0.72.0/22",
|
|
]
|
|
|
|
#: https://www.cloudflare.com/ips-v6
|
|
CLOUDFLARE_IPV6 = [
|
|
"2400:cb00::/32", "2606:4700::/32", "2803:f800::/32",
|
|
"2405:b500::/32", "2405:8100::/32", "2a06:98c0::/29",
|
|
"2c0f:f248::/32",
|
|
]
|
|
|
|
#: https://docs-cortex.paloaltonetworks.com/r/1/Cortex-Xpanse/Scanning-activity
|
|
PALOALTO_IPV4 = [
|
|
"35.203.210.0/23", "144.86.173.0/24", "147.185.132.0/23",
|
|
"162.216.149.0/24", "162.216.150.0/24", "172.105.147.0/24",
|
|
"198.235.24.0/24", "205.210.31.0/24", "216.25.88.0/21",
|
|
]
|
|
|
|
PALOALTO_IPV6 = [
|
|
"2604:a940:300:5b6::/64",
|
|
"2604:a940:301:225::/64",
|
|
"2604:a940:302:118::/64",
|
|
]
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Active whitelist state (mutated by refresh)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
#: address-list name -> CIDR ranges (v4 + v6 mixed, split at render time)
|
|
WHITELISTS: dict[str, list[str]] = {
|
|
"Init7": INIT7_IPV4 + INIT7_IPV6,
|
|
"Cloudflare": CLOUDFLARE_IPV4 + CLOUDFLARE_IPV6,
|
|
"PaloAlto": PALOALTO_IPV4 + PALOALTO_IPV6,
|
|
}
|
|
|
|
#: pre-parsed networks, so matching does not re-parse CIDRs per IP
|
|
WHITELIST_NETS: dict[str, tuple[Any, ...]] = {
|
|
name: tuple(ipaddress.ip_network(c) for c in cidrs)
|
|
for name, cidrs in WHITELISTS.items()
|
|
}
|
|
|
|
#: guard so refresh runs at most once per process
|
|
_refreshed = False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CIDR extraction / validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def extract_cidrs(text: str) -> list[str]:
|
|
"""Extract valid CIDR strings from plain text, CSV, or HTML."""
|
|
seen: set[str] = set()
|
|
cidrs: list[str] = []
|
|
for match in CIDR_RE.finditer(text):
|
|
candidate = match.group(1).strip()
|
|
try:
|
|
network = ipaddress.ip_network(candidate, strict=False)
|
|
except ValueError:
|
|
continue
|
|
rendered = str(network)
|
|
if rendered not in seen:
|
|
seen.add(rendered)
|
|
cidrs.append(rendered)
|
|
return cidrs
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Local file loading
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _read_local_file(path: str) -> list[str]:
|
|
"""Read a local whitelist file and extract valid CIDRs."""
|
|
try:
|
|
with open(path, "r", encoding="utf-8", errors="replace") as f:
|
|
return extract_cidrs(f.read())
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
def _load_local_whitelists() -> dict[str, list[str]]:
|
|
"""Load whitelist CIDRs from local data/ files.
|
|
|
|
Returns a dict of {name: [CIDR strings]}. Only keys with
|
|
successfully parsed CIDRs are included.
|
|
"""
|
|
out: dict[str, list[str]] = {}
|
|
# Init7 has no published geofeed; its ranges are built-in (INIT7_IPV4/6).
|
|
cf_v4 = _read_local_file(CLOUDFLARE_V4_FILE)
|
|
cf_v6 = _read_local_file(CLOUDFLARE_V6_FILE)
|
|
if cf_v4 or cf_v6:
|
|
out["Cloudflare"] = cf_v4 + cf_v6
|
|
cidrs = _read_local_file(PALOALTO_FILE)
|
|
if cidrs:
|
|
out["PaloAlto"] = cidrs
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# HTTP download helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _download_text(url: str) -> str:
|
|
"""Download a text file from a URL, return as string."""
|
|
import urllib.request
|
|
request = urllib.request.Request(
|
|
url,
|
|
headers={
|
|
"User-Agent": "Mozilla/5.0 (compatible; intrusion-detection/1.0)",
|
|
"Accept": "text/plain,text/csv,text/html;q=0.9,*/*;q=0.8",
|
|
},
|
|
)
|
|
with urllib.request.urlopen(request) as response:
|
|
return response.read().decode("utf-8", errors="replace")
|
|
|
|
|
|
def _download_cidrs(url: str) -> list[str]:
|
|
"""Download a whitelist source and extract valid CIDRs from it."""
|
|
return extract_cidrs(_download_text(url))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Refresh pipeline
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _download_whitelists() -> dict[str, list[str]]:
|
|
"""Download the latest whitelists from the Internet and save to local files.
|
|
|
|
Returns a dict of {name: [CIDR strings]}. Only keys with
|
|
successfully parsed CIDRs are included; failures are logged
|
|
and skipped so built-in fallbacks remain untouched.
|
|
Saves each download to the corresponding local file on success.
|
|
"""
|
|
out: dict[str, list[str]] = {}
|
|
# Init7 has no published geofeed; its ranges are built-in (INIT7_IPV4/6).
|
|
try:
|
|
text_v4 = _download_text(CLOUDFLARE_URL_V4)
|
|
text_v6 = _download_text(CLOUDFLARE_URL_V6)
|
|
cidrs = extract_cidrs(text_v4) + extract_cidrs(text_v6)
|
|
if cidrs:
|
|
out["Cloudflare"] = cidrs
|
|
_save_local_file(CLOUDFLARE_V4_FILE, text_v4)
|
|
_save_local_file(CLOUDFLARE_V6_FILE, text_v6)
|
|
except Exception as e:
|
|
print(f"Error downloading Cloudflare whitelist: {e}")
|
|
# Do not download PaloAlto whitelist for now, as it is not a reliable source of CIDRs and often returns HTML instead of plain text.
|
|
# try:
|
|
# text = _download_text(PALOALTO_URL)
|
|
# cidrs = extract_cidrs(text)
|
|
# if cidrs and len(cidrs) > 0:
|
|
# out["PaloAlto"] = cidrs
|
|
# _save_local_file(PALOALTO_FILE, text)
|
|
# except Exception as e:
|
|
# print(f"Error downloading PaloAlto whitelist: {e}")
|
|
return out
|
|
|
|
|
|
def _save_local_file(path: str, text: str) -> None:
|
|
"""Save downloaded text to a local file, creating parent dirs as needed."""
|
|
try:
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
f.write(text)
|
|
except Exception as e:
|
|
print(f"Error saving whitelist file {path}: {e}")
|
|
|
|
|
|
def update_whitelists() -> None:
|
|
"""Load whitelists (local files first, then remote, then built-in fallback).
|
|
|
|
Priority per provider:
|
|
1. Local file in data/<provider>/
|
|
2. Remote download (saves to local file on success)
|
|
3. Built-in hardcoded constants (always available)
|
|
"""
|
|
global WHITELISTS, WHITELIST_NETS
|
|
|
|
# Start from built-in fallbacks, then layer local, then remote
|
|
WHITELISTS = {
|
|
"Init7": INIT7_IPV4 + INIT7_IPV6,
|
|
"Cloudflare": CLOUDFLARE_IPV4 + CLOUDFLARE_IPV6,
|
|
"PaloAlto": PALOALTO_IPV4 + PALOALTO_IPV6,
|
|
}
|
|
|
|
# Layer 1: local files
|
|
local = _load_local_whitelists()
|
|
if local:
|
|
WHITELISTS.update(local)
|
|
|
|
# Layer 2: remote download (highest priority, saves local)
|
|
remote = _download_whitelists()
|
|
if remote:
|
|
WHITELISTS.update(remote)
|
|
|
|
# Rebuild parsed networks
|
|
WHITELIST_NETS = {
|
|
name: tuple(ipaddress.ip_network(c) for c in cidrs)
|
|
for name, cidrs in WHITELISTS.items()
|
|
}
|
|
|
|
|
|
def refresh_whitelists(force: bool = False) -> bool:
|
|
"""Refresh provider whitelists once per process, keeping built-in fallbacks.
|
|
|
|
Returns True if a refresh was attempted, False if already done.
|
|
"""
|
|
global _refreshed
|
|
if _refreshed and not force:
|
|
return False
|
|
update_whitelists()
|
|
_refreshed = True
|
|
return True
|