fix(biz-compare): validate templates and protect comparison inputs

netx_api/biz_state/compare_validation.py

@@ -0,0 +1,130 @@
+"""Validate template writes without silently dropping parts of the comparison."""
+from __future__ import annotations
+
+import math
+import re
+from typing import Any
+
+from fastapi import HTTPException
+
+FILTER_OPS = {"eq", "==", "ci_eq", "ne", "!=", "in", "not_in", "nin", "contains", "empty", "not_empty", "nonempty", "regex", "age_timer"}
+COMPARE_MODES = {"", "eq", "ignore", "skip", "off", "numeric", "number", "int", "float", "percent", "pct", "rel"}
+NORMALIZE_MODES = {"", "none", "strip", "lower", "upper", "mac", "empty_as_blank"}
+
+
+def invalid(path: str, reason: str) -> None:
+    raise HTTPException(status_code=400, detail={"error": "invalid_template", "path": path, "reason": reason})
+
+
+def _strings(value: Any, path: str, *, required: bool = False) -> None:
+    if not isinstance(value, list) or any(not isinstance(x, str) or not x.strip() for x in value):
+        invalid(path, "nonempty_strings_required")
+    if required and not value:
+        invalid(path, "key_fields_required")
+    if len({x.strip() for x in value}) != len(value):
+        invalid(path, "duplicate_field")
+
+
+def validate_filters(value: Any, path: str, depth: int = 0) -> None:
+    if not isinstance(value, list):
+        invalid(path, "list_required")
+    if depth > 12:
+        invalid(path, "filter_nesting_too_deep")
+    for i, node in enumerate(value):
+        here = f"{path}[{i}]"
+        if not isinstance(node, dict) or not node:
+            invalid(here, "filter_required")
+        groups = [key for key in ("any", "all") if key in node]
+        if groups:
+            if len(groups) != 1 or any(node.get(key) is not None for key in ("field", "op", "value")):
+                invalid(here, "group_or_leaf_required")
+            kids = node[groups[0]]
+            if not isinstance(kids, list) or not kids:
+                invalid(here, "nonempty_group_required")
+            validate_filters(kids, f"{here}.{groups[0]}", depth + 1)
+            continue
+        if not isinstance(node.get("field"), str) or not node["field"].strip():
+            invalid(here, "filter_field_required")
+        op = str(node.get("op") or "eq").strip().lower()
+        if op not in FILTER_OPS:
+            invalid(here, "unknown_filter_operator")
+        expect = node.get("value")
+        if op in ("in", "not_in", "nin"):
+            if not isinstance(expect, list) or not expect or any(not isinstance(x, (str, int, float, bool)) for x in expect):
+                invalid(here, "nonempty_value_list_required")
+        elif op not in ("empty", "not_empty", "nonempty", "age_timer"):
+            if not isinstance(expect, (str, int, float, bool)):
+                invalid(here, "scalar_value_required")
+            if op in ("contains", "regex") and not str(expect).strip():
+                invalid(here, "value_required")
+        if op == "regex":
+            try:
+                re.compile(str(expect), re.I)
+            except re.error:
+                invalid(here, "invalid_regex")
+
+
+def validate_sheet(sheet: Any, path: str, *, partial: bool = False) -> None:
+    if not isinstance(sheet, dict):
+        invalid(path, "sheet_required")
+    if not partial or "metric_id" in sheet:
+        if not isinstance(sheet.get("metric_id"), str) or not sheet["metric_id"].strip():
+            invalid(path, "metric_id_required")
+    for key in ("key_fields", "iface_fields", "compare_fields", "display_fields", "ignore_fields"):
+        if key in sheet or (key == "key_fields" and not partial):
+            _strings(sheet.get(key), f"{path}.{key}", required=key == "key_fields")
+    if "ignore_port_changes" in sheet and sheet["ignore_port_changes"] is not None and not isinstance(sheet["ignore_port_changes"], bool):
+        invalid(path, "boolean_port_policy_required")
+    if "row_filters" in sheet:
+        validate_filters(sheet["row_filters"], f"{path}.row_filters")
+    if "field_rules" in sheet:
+        rules = sheet["field_rules"]
+        if not isinstance(rules, list):
+            invalid(path, "field_rules_list_required")
+        seen = set()
+        for i, rule in enumerate(rules):
+            here = f"{path}.field_rules[{i}]"
+            if not isinstance(rule, dict) or not isinstance(rule.get("field"), str) or not rule["field"].strip():
+                invalid(here, "rule_field_required")
+            field = rule["field"].strip()
+            if field in seen:
+                invalid(here, "duplicate_field_rule")
+            seen.add(field)
+            if str(rule.get("compare") or "").strip().lower() not in COMPARE_MODES:
+                invalid(here, "unknown_compare_mode")
+            if str(rule.get("normalize") or "").strip().lower() not in NORMALIZE_MODES:
+                invalid(here, "unknown_normalize_mode")
+            tol = rule.get("tolerance")
+            if tol is not None:
+                if isinstance(tol, bool) or not isinstance(tol, (int, float)) or not math.isfinite(tol) or tol < 0:
+                    invalid(here, "finite_nonnegative_tolerance_required")
+
+
+def validate_template_body(body: dict[str, Any], *, partial: bool = False) -> None:
+    metrics = body.get("metrics")
+    if metrics is not None:
+        if not isinstance(metrics, list) or not metrics:
+            invalid("metrics", "metrics_required")
+        seen = set()
+        for i, sheet in enumerate(metrics):
+            path = f"metrics[{i}]"
+            validate_sheet(sheet, path)
+            sid = str(sheet.get("sheet_id") or sheet["metric_id"]).strip()
+            if sid in seen:
+                invalid(path, "duplicate_sheet_id")
+            seen.add(sid)
+    else:
+        validate_sheet(body, "template", partial=partial)
+    if "iface_normalize_rules" in body and body["iface_normalize_rules"] is not None:
+        rules = body["iface_normalize_rules"]
+        if not isinstance(rules, list):
+            invalid("iface_normalize_rules", "list_required")
+        seen = set()
+        for i, rule in enumerate(rules):
+            path = f"iface_normalize_rules[{i}]"
+            if not isinstance(rule, dict) or any(not isinstance(rule.get(k), str) or not rule[k].strip() for k in ("from", "to")):
+                invalid(path, "alias_pair_required")
+            key = rule["from"].strip().lower()
+            if key in seen:
+                invalid(path, "duplicate_alias")
+            seen.add(key)