webproxyserver-proxyserver/app/policy.py

"""Regel-Engine für White-/Blacklists auf Host- und Port-Ebene.

Reihenfolge der Auswertung:
  1. Loopback-Schutz (Einstellung block_loopback)
  2. Sperr-Regeln des Benutzers      -> blockiert
  3. Erlaubt-Regeln des Benutzers    -> erlaubt
  4. Globale Sperr-Regeln            -> blockiert
  5. Globale Erlaubt-Regeln          -> erlaubt
  6. Standardrichtlinie (des Benutzers, sonst global)

Host-Muster:
  ""  oder "*"         -> alle Ziele
  example.com          -> example.com und alle Subdomains
  *.example.com        -> Platzhalter (fnmatch), nur Subdomains
  10.0.0.0/8, 1.2.3.4  -> IP/Netz (gegen aufgelöste Ziel-IP geprüft)
  re:^ads?\\.          -> regulärer Ausdruck
Port-Muster: "" = alle, "25", "8000-9000", "25,465,587"
"""
import fnmatch
import ipaddress
import re
from dataclasses import dataclass

from .db import db

LOOPBACK = [ipaddress.ip_network("127.0.0.0/8"), ipaddress.ip_network("::1/128"),
            ipaddress.ip_network("0.0.0.0/32")]


def parse_ports(spec: str) -> list[tuple[int, int]] | None:
    """Gibt Liste von (von, bis) zurück; None = alle Ports. Wirft ValueError bei Fehlern."""
    spec = (spec or "").strip()
    if not spec or spec == "*":
        return None
    ranges = []
    for part in spec.split(","):
        part = part.strip()
        if not part:
            continue
        if "-" in part:
            a, b = (int(x) for x in part.split("-", 1))
        else:
            a = b = int(part)
        if not (0 < a <= 65535 and 0 < b <= 65535 and a <= b):
            raise ValueError(f"Ungültiger Port-Bereich: {part}")
        ranges.append((a, b))
    return ranges


class HostMatcher:
    def __init__(self, pattern: str):
        self.raw = (pattern or "").strip()
        p = self.raw.lower()
        self.any = p in ("", "*")
        self.net = None
        self.regex = None
        self.glob = None
        self.domain = None
        if self.any:
            return
        if p.startswith("re:"):
            self.regex = re.compile(self.raw[3:], re.IGNORECASE)
            return
        try:
            self.net = ipaddress.ip_network(p, strict=False)
            return
        except ValueError:
            pass
        if "*" in p or "?" in p:
            self.glob = p
        else:
            self.domain = p.lstrip(".").rstrip(".")

    def match(self, host: str, ip) -> bool:
        if self.any:
            return True
        if self.net is not None:
            return ip is not None and ip.version == self.net.version and ip in self.net
        h = (host or "").lower().rstrip(".")
        if self.regex is not None:
            return bool(self.regex.search(h))
        if self.glob is not None:
            return fnmatch.fnmatchcase(h, self.glob)
        return h == self.domain or h.endswith("." + self.domain)


def validate_rule(host: str, ports: str) -> None:
    HostMatcher(host)  # wirft re.error bei kaputtem Regex
    parse_ports(ports)


@dataclass
class Rule:
    id: int
    user_id: int | None
    action: str
    host: HostMatcher
    ports: list | None
    text: str

    def match(self, host: str, port: int, ip) -> bool:
        if self.ports is not None and not any(a <= port <= b for a, b in self.ports):
            return False
        return self.host.match(host, ip)


@dataclass
class Decision:
    allowed: bool
    reason: str
    rule_id: int | None = None


class Policy:
    def __init__(self):
        self.global_rules: list[Rule] = []
        self.user_rules: dict[int, list[Rule]] = {}
        self.settings: dict = {}
        self.anon_nets: list = []
        self.reload()

    def reload(self) -> None:
        self.settings = db.settings()
        g, u = [], {}
        for r in db.query("SELECT * FROM rules WHERE enabled=1 ORDER BY id"):
            try:
                rule = Rule(r["id"], r["user_id"], r["action"], HostMatcher(r["host"]),
                            parse_ports(r["ports"]), f'{r["action"]} {r["host"] or "*"}:{r["ports"] or "*"}')
            except (ValueError, re.error):
                continue
            (u.setdefault(r["user_id"], []) if r["user_id"] else g).append(rule)
        self.global_rules, self.user_rules = g, u
        self.anon_nets = parse_networks(self.settings.get("anonymous_networks", ""))

    def is_anonymous_allowed(self, client_ip: str) -> bool:
        return ip_in(client_ip, self.anon_nets)

    def decide(self, user: dict | None, host: str, port: int, ip=None) -> Decision:
        if ip is None:
            try:
                ip = ipaddress.ip_address(host.strip("[]"))
            except ValueError:
                ip = None
        if ip is not None and self.settings.get("block_loopback", "1") == "1":
            if getattr(ip, "ipv4_mapped", None):
                ip = ip.ipv4_mapped
            if any(ip.version == n.version and ip in n for n in LOOPBACK):
                return Decision(False, "Loopback-Ziel gesperrt")
        layers = []
        if user:
            layers.append(("Benutzer", self.user_rules.get(user["id"], [])))
        layers.append(("Global", self.global_rules))
        for label, rules in layers:
            for action in ("deny", "allow"):
                for r in rules:
                    if r.action == action and r.match(host, port, ip):
                        return Decision(action == "allow", f"{label}-Regel #{r.id}: {r.text}", r.id)
        policy = self.settings.get("default_policy", "allow")
        source = "Standardrichtlinie"
        if user and user.get("policy") in ("allow", "deny"):
            policy = user["policy"]
            source = "Benutzer-Standardrichtlinie"
        return Decision(policy == "allow", f"{source}: {'erlauben' if policy == 'allow' else 'sperren'}")


def parse_networks(text: str) -> list:
    nets = []
    for part in re.split(r"[\s,;]+", text or ""):
        if part:
            try:
                nets.append(ipaddress.ip_network(part, strict=False))
            except ValueError:
                pass
    return nets


def ip_in(ip: str, nets: list) -> bool:
    try:
        addr = ipaddress.ip_address(ip)
    except ValueError:
        return False
    if getattr(addr, "ipv4_mapped", None):
        addr = addr.ipv4_mapped
    return any(addr.version == n.version and addr in n for n in nets)


policy = Policy()