intrusion-detection/app/utils/whitelists.py

#!/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,
    OPENFACTORY_FILE,
    OPENFACTORY_URL,
    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 OPENFACTORY_FILE, CLOUDFLARE_V4_FILE,
#  CLOUDFLARE_V6_FILE, PALOALTO_FILE; URLs as OPENFACTORY_URL, ...)

# ---------------------------------------------------------------------------
# Built-in fallback CIDR lists
# ---------------------------------------------------------------------------

#: Openfactory AS58299 colocation IP range (https://openfactory.net/geofeed.csv)
OPENFACTORY_IPV4 = ["5.226.144.0/21"]

#: 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]] = {
    "Openfactory": OPENFACTORY_IPV4,
    "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]] = {}
    cidrs = _read_local_file(OPENFACTORY_FILE)
    if cidrs:
        out["Openfactory"] = cidrs
    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]] = {}
    try:
        text = _download_text(OPENFACTORY_URL)
        cidrs = extract_cidrs(text)
        if cidrs:
            out["Openfactory"] = cidrs
            _save_local_file(OPENFACTORY_FILE, text)
    except Exception as e:
        print(f"Error downloading Openfactory whitelist: {e}")
    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 = {
        "Openfactory": OPENFACTORY_IPV4,
        "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