mirror of
https://github.com/hansjone/netx.git
synced 2026-10-11 15:30:45 +08:00
130 lines
6.2 KiB
Python
130 lines
6.2 KiB
Python
"""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)
|