OPNexus/backend/app/services/rule_analysis/cleanup.py

"""Cleanup candidates of one firewall: unused, shadowed, duplicate and long-disabled rules (read-only).

Nothing here changes a firewall. The list says *which rules are worth a look and how sure that is*; removing
them is a separate step (not part of this module).

Categories
* ``unused``       OPNexus has watched the rule's pf counters for at least N days and saw no hit
                   (services/rule_hit_service.py; N is a parameter, default 30).
* ``shadowed``     an earlier quick rule already decides everything the rule matches (rule_analysis/coverage.py).
* ``duplicate``    like shadowed, with the same action and the rule also covering the earlier one.
* ``disabled_long`` disabled, and OPNexus has seen it unchanged for N days or more (rule history).

Confidence (never higher than what is proven; what is not proven is never "sure")
* ``sure``      the evidence is complete: unused = the rule is loaded in pf, has no schedule, the counters were
                sampled recently and over the whole window; shadowed/duplicate = literal comparison, order from
                ``sort_order``, the covering rule is loaded; disabled_long = observed unchanged for N days.
* ``probable``  relies on a guess or on static alias content, or pf's loaded ruleset is not available, or the rule
                has a schedule (a hit might happen at another time), or (cluster) some node has no usage data for
                the rule. In a cluster "unused" is judged over all nodes (rule_hit_service.merge_cluster_hits):
                the CARP backup sees almost no traffic, so its zero alone never makes a rule unused.
* ``unknown``   there is a hint but no support: the rule is not loaded in pf (no counters can exist), or the
                counters are stale.

``removable`` says whether the rule could be offered for removal at all; the reason is a code:
``automatic`` (generated by OPNsense), ``legacy`` (not yet migrated, migrate it in OPNsense first),
``managed`` (OPNexus manages the rule: remove it through the normal flow of the rules page).
"""
from __future__ import annotations

from datetime import datetime, timedelta
from typing import Iterable

from app.services import rule_hit_service
from app.services.rule_analysis import coverage
from app.services.rule_analysis.context import Context
from app.services.rule_analysis.model import SECTION_ANTISPOOF, SECTION_INVALID, SECTION_NAT_AUTO, NormRule

CATEGORIES = ("unused", "shadowed", "duplicate", "disabled_long")
DEFAULT_UNUSED_DAYS = 30
MIN_DAYS, MAX_DAYS = 1, 3650


def removal_block(rule: NormRule, managed: dict | None) -> str | None:
    """Why the rule could not be offered for removal (None = it could)."""
    if rule.automatic or rule.section in (SECTION_ANTISPOOF, SECTION_NAT_AUTO):
        return "automatic"
    if rule.legacy:
        return "legacy"
    if managed is not None:
        return "managed"
    return None


def unused_confidence(rule: NormRule, hit, now: datetime) -> str:
    """Confidence of an 'unused' verdict (the caller already knows the window passed without a hit)."""
    if rule.loaded is False:
        return "unknown"   # pf did not load it (schedule, missing interface, gateway down): no counter can exist
    if hit.last_sampled_at is None or now - hit.last_sampled_at > rule_hit_service.SAMPLE_FRESH:
        return "unknown"   # stale counters say nothing about the present
    if rule.loaded is None or rule.sched:
        return "probable"  # pf's ruleset not available / the rule only works at certain times
    if getattr(hit, "peer_missing", False):
        return "probable"  # cluster: some node has no usage data for this rule (rule_hit_service.ClusterHit)
    return "sure"


def _name(rule: NormRule) -> str:
    return rule.description or rule.label or rule.uuid or (f"@{rule.pf_numbers[0]}" if rule.pf_numbers else "")


def build(rules: list[NormRule], ctx: Context, *, hits: dict, states: dict, managed: dict[str, dict],
          now: datetime, unused_days: int = DEFAULT_UNUSED_DAYS, order_guessed: bool = False) -> list[dict]:
    """Candidate rows, in evaluation order. ``hits``: RuleHit per rule label, ``states``: LiveRuleState per label,
    ``managed``: {label: {"kind", "id", "name"}} of the rules OPNexus deployed to this firewall."""
    unused_days = max(MIN_DAYS, min(MAX_DAYS, int(unused_days)))
    overlaps = {id(o.rule): o for o in coverage.analyse(rules, ctx, order_guessed=order_guessed)}
    rows: list[dict] = []
    for rule in rules:
        if rule.automatic or rule.section in (SECTION_ANTISPOOF, SECTION_INVALID, SECTION_NAT_AUTO):
            continue
        categories: list[dict] = []
        key = rule.uuid.lower()
        hit = hits.get(key)
        if hit is not None and rule_hit_service.classify(hit, now, unused_days) == "unused":
            category = {"kind": "unused", "confidence": unused_confidence(rule, hit, now), "days": unused_days}
            if getattr(hit, "peer_missing", False):
                category["notes"] = ["CLUSTER_PEER_USAGE_UNKNOWN"]
            categories.append(category)
        overlap = overlaps.get(id(rule))
        if overlap is not None:
            categories.append({
                "kind": overlap.kind, "confidence": overlap.confidence,
                "by": {"uuid": overlap.by.uuid, "description": _name(overlap.by), "section": overlap.by.section,
                       "automatic": overlap.by.automatic},
            })
        state = states.get(key)
        if (not rule.enabled and state is not None and state.present
                and now - state.changed_at >= timedelta(days=unused_days)):
            categories.append({"kind": "disabled_long", "confidence": "sure", "days": unused_days,
                               "since": state.changed_at})
        if not categories:
            continue
        block = removal_block(rule, managed.get(key))
        rows.append({
            **rule.summary(), "enabled": rule.enabled, "categories": categories,
            "hit_state": rule_hit_service.classify(hit, now, unused_days) if hit is not None else None,
            "last_hit_at": hit.last_hit_at if hit is not None else None,
            "watched_since": hit.first_seen_at if hit is not None else None,
            "removable": block is None, "not_removable_reason": block,
            "managed_by": managed.get(key),
        })
    return rows


def summarize(rows: Iterable[dict]) -> dict:
    counts = {kind: 0 for kind in CATEGORIES}
    for row in rows:
        for category in row["categories"]:
            counts[category["kind"]] += 1
    return counts