feat(validation): server-side MO value validation — real-APIC 400 on bad values (v0.23.0)

Closes F10: the sim silently stored out-of-range/malformed scalar values
(e.g. vlan-9999, IP 999.888.777.x) that real APIC/NDO reject server-side.

New aci_sim/rest_aci/validators.py: table-driven RULES {(class,prop) -> validator}
(range/enum/IP·MAC-format/vlan-encap). Enforced in writes.py::_validate_planned
(APIC) and ndo/patch.py::apply_json_patch (NDO) before any store mutation.
Fail-safe default-allow: only registered (class,prop) validated; unknown pass.
Bad values now -> APIC Error 103 Malformed MO body. ~45 rules / ~18 classes.

Constraints sourced from vendored cisco.aci arg-specs + standard ACI ranges,
cross-checked against the real-machine var corpus for zero false-rejects
(regression: full create_tenant MS-TN1 + SF-TN1 chains stay green). +176 tests
(1098 passed). Referential integrity split out as F11 (needs real-APIC study).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017zdUTqU9fvCCvF3uVGZsu1
This commit is contained in:
dtzp555-max
2026-07-09 09:59:34 +10:00
co-authored by Claude Opus 4.8
parent ffaf9964c2
commit fe3d13221a
8 changed files with 1353 additions and 29 deletions
+79
View File
@@ -28,6 +28,8 @@ import hashlib
import re
from typing import Any
from aci_sim.rest_aci.validators import Validator, fmt, rng
#: Every collection key a schema template can carry. Real NDO always serves
#: these back as (at minimum) empty lists even when a caller's create/add
#: payload omitted them — e.g. `mso_schema_template.py`'s "schema does not
@@ -81,6 +83,77 @@ class PatchError(Exception):
"""Raised when a JSON-Patch op cannot be applied to the stored document."""
# ---------------------------------------------------------------------------
# F10 batch 3 — NDO value validation (design doc §3.3).
#
# NDO stores schema JSON, not APIC MOs, so values arrive as JSON-Patch
# add/replace op values rather than (class, prop) MO attrs — the RULES
# registry keying in aci_sim/rest_aci/validators.py doesn't apply verbatim.
# Instead NDO_RULES keys on (collection, field): *collection* is the name of
# the array a value is being written into/under (e.g. "subnets",
# "staticPorts" — the dict key one level up from the array itself), *field*
# is the attribute name within that array's object shape.
#
# The NDO value surface is much narrower than APIC's (design §3.3: "subnet
# ip, static-port vlan, a few vrf/bd enums") — this batch implements exactly
# the two the design doc calls out by name: subnet ip and static-port VLAN.
# The real `cisco.mso` field for the static-port VLAN is `portEncapVlan`
# (confirmed against this repo's own PR-12 test fixtures,
# tests/test_pr12_site_local.py — the design doc's pseudocode used the
# placeholder name "vlan"; `portEncapVlan` is the actual wire field).
# ---------------------------------------------------------------------------
NDO_RULES: dict[tuple[str, str], Validator] = {
("subnets", "ip"): fmt("ipv4_iface"),
("staticPorts", "portEncapVlan"): rng(1, 4094),
}
def _validate_ndo_value(collection: str, field: str, value: Any) -> None:
"""Look up *collection*/*field* in NDO_RULES and raise PatchError if
*value* fails it. Fail-safe default-allow, same policy as the APIC
registry: an unregistered (collection, field) or a None value is not
validated."""
validator = NDO_RULES.get((collection, field))
if validator is None or value is None:
return
try:
validator(collection, field, value, None)
except ValueError as reason:
raise PatchError(
f"Invalid value {value!r} for property {field!r} of {collection!r}: {reason}"
) from None
def _validate_ndo_op(tokens: list[str], container: Any, last: str, value: Any) -> None:
"""Validate a single JSON-Patch add/replace op's *value* against
NDO_RULES, given the already-resolved (container, last) from
``_resolve_container`` and the op's full *tokens* path.
Two value shapes reach here (design §3.3 "applied to each op's value"):
- Whole-object add/insert (``container`` is the collection list
itself, e.g. ``.../subnets/-`` or ``.../staticPorts/-``): *value* is
a dict of fields for the new item. The collection name is the token
that named this list, i.e. ``tokens[-2]`` (the list was reached by
indexing into ``doc[...][tokens[-2]]``). Validate every field of
*value* that NDO_RULES knows about.
- Single-field replace/add on an EXISTING item (``container`` is that
item's own dict, e.g. ``.../subnets/0/ip``): *last* IS the field
name and *value* is the scalar. The collection name is one level up
again — ``tokens[-3]`` (the list containing the item ``tokens[-2]``
indexes into).
"""
if isinstance(container, list):
collection = tokens[-2] if len(tokens) >= 2 else None
if collection is not None and isinstance(value, dict):
for field, sub_value in value.items():
_validate_ndo_value(collection, field, sub_value)
elif isinstance(container, dict):
collection = tokens[-3] if len(tokens) >= 3 else None
if collection is not None:
_validate_ndo_value(collection, last, value)
def _normalize_object(obj: dict, array_key: str, schema_id: str, template_name: str, anp_name: str = "") -> None:
"""Backfill *obj*'s missing collection-default keys for its container
array (e.g. array_key="bds" -> obj gets subnets/dhcpLabels defaults),
@@ -695,6 +768,12 @@ def apply_json_patch(doc: dict, ops: list[dict]) -> dict:
container, last = _resolve_container(doc, tokens)
if op in ("add", "replace"):
# F10 batch 3: validate before mutating (design §3.3) — a bad
# subnet ip / static-port vlan 400s instead of landing in the
# stored schema doc.
_validate_ndo_op(tokens, container, last, value)
if op == "add":
if isinstance(container, list):
if last == "-":
+387
View File
@@ -0,0 +1,387 @@
"""Server-side value validators for the aci-sim APIC write face (F10).
aci-sim is a faithful ACI management-plane MODEL: it stores/associates MOs
correctly but historically performed no APIC-style server-side attribute
validation. A real APIC 400-rejects out-of-range/malformed values (e.g.
``l3extRsPathL3OutAtt encap=vlan-9999``, ``l3extRsNodeL3OutAtt
rtrId=999.888.777.241``) even when the posting client (``cisco.aci.aci_rest``
raw MO bodies, in this campaign) does zero client-side validation of its own.
This module is the table-driven rule catalog that closes that gap — see
``_F10_VALIDATION_DESIGN.md`` (design doc, authoritative) for the full rule
table, sourcing, and false-reject risk analysis. Do not add rules here that
aren't in that table without updating the design doc first.
Architecture (design §3.1):
- Small composable validator *factories* (``rng``, ``one_of``, ``fmt``,
``all_of``) each return a ``Validator`` callable with signature
``(cls: str, prop: str, value, ctx) -> None`` that raises ``ValueError``
with a human-readable reason on an invalid value.
- ``RULES`` maps ``(class, prop) -> Validator``. A ``(class, prop)`` pair
with NO entry is intentionally unvalidated — see "Fail-safe" below.
Fail-safe / false-reject policy (design §3.6, load-bearing): **default-allow**.
Only pairs present in RULES are checked; every other (class, prop) passes
through untouched. The bias is hard toward "never reject a legal value"
every rule here is proven against the canonical corpus (design §6) before
being added, and any rule that would reject a canonical value must be
loosened to admit it (never loosened to admit a known-bad value instead).
"""
from __future__ import annotations
import ipaddress
import re
from typing import Callable
#: A validator inspects one (class, prop, value) triple and either returns
#: None (value is legal) or raises ValueError(reason) (value is illegal).
#: ``ctx`` is reserved for cross-field checks (Priority-3 in the design doc,
#: e.g. fvnsEncapBlk from<=to) — NOT implemented in this batch (P1+P2 only);
#: it is threaded through now so a later batch can add cross-field rules
#: without changing the RULES calling convention.
Validator = Callable[[str, str, object, object], None]
# ---------------------------------------------------------------------------
# Factories
# ---------------------------------------------------------------------------
def rng(lo: int, hi: int | None = None, *, transform: Callable[[object], object] | None = None) -> Validator:
"""Integer range check: ``lo <= int(value) <= hi``.
*hi=None* means "no upper bound" (used for the few design-doc rules that
only specify a floor, e.g. infraPortBlk ``fromPort/toPort/fromCard/
toCard`` >= 1). *transform* is applied to the raw value before the int()
conversion (e.g. stripping a "vlan-" prefix) — not currently used by any
RULES entry below (that parsing lives in the ``vlan_encap`` fmt kind
instead) but kept per the design §3.1 factory signature for forward use.
"""
def _validator(cls: str, prop: str, value: object, ctx: object = None) -> None:
raw = value if transform is None else transform(value)
try:
n = int(raw) # type: ignore[arg-type]
except (TypeError, ValueError):
raise ValueError(f"not an integer, got {value!r}") from None
if n < lo or (hi is not None and n > hi):
bound = f"[{lo},{hi}]" if hi is not None else f">= {lo}"
raise ValueError(f"must be {bound}, got {n}")
return _validator
def one_of(allowed: set[str] | frozenset[str], *, list_sep: str | None = None) -> Validator:
"""Enum check. Plain membership by default; with *list_sep* the MO's
string value is a token-list (real APIC stores multi-valued enum props
like ``scope``/``ctrl``/``aggregate``/``peerCtrl`` as a single
comma-joined string — confirmed against ``cisco.aci`` module source,
e.g. ``aci_bd_subnet.py``/``aci_l3out_extsubnet.py`` doing
``",".join(sorted(scope))``) — split on *list_sep*, and every non-empty
token must be a member of *allowed*. An empty value (no tokens at all)
is treated as vacuously valid — nothing to check, and rejecting it would
risk a false-reject on an unset/omitted enum (design §3.6 bias).
"""
allowed_set = set(allowed)
def _validator(cls: str, prop: str, value: object, ctx: object = None) -> None:
if not isinstance(value, str):
raise ValueError(f"expected a string, got {type(value).__name__}")
if list_sep is None:
if value not in allowed_set:
raise ValueError(f"must be one of {sorted(allowed_set)}, got {value!r}")
return
tokens = [t.strip() for t in value.split(list_sep) if t.strip()]
for token in tokens:
if token not in allowed_set:
raise ValueError(
f"token {token!r} not in allowed set {sorted(allowed_set)} (value={value!r})"
)
return _validator
_VLAN_ENCAP_RE = re.compile(r"^(vlan|vxlan)-(\d+)$")
_VLAN_MIN, _VLAN_MAX = 1, 4094 # NOT 4096 — see design doc §7 point 2:
# cisco.aci docs say "1 and 4096" but real APIC reserves 4095/4096; 4094 is
# the real server ceiling and comfortably admits every canonical value
# (max observed: 2611).
_VXLAN_MIN, _VXLAN_MAX = 1, 16777215 # F10 fix (independent review): the
# vlan-N 4094 ceiling does NOT apply to vxlan-N — VXLAN VNID is a 24-bit
# field (RFC 7348), max 2**24-1 = 16777215. The original single shared
# ceiling 400-rejected every legal vxlan-N value above 4094, a 100%
# false-reject on this prefix. Fail-safe bias (module docstring / design
# §3.6): take the widest legal range rather than guess a tighter one.
_MAC_COLON_RE = re.compile(r"^[0-9A-Fa-f]{2}(:[0-9A-Fa-f]{2}){5}$")
_MAC_DOT_RE = re.compile(r"^[0-9A-Fa-f]{4}\.[0-9A-Fa-f]{4}\.[0-9A-Fa-f]{4}$")
#: Named L4 ports real APIC/`cisco.aci` accept in place of a numeric port,
#: sourced verbatim from `aci-ansible-dev`'s `FILTER_PORT_MAPPING`
#: (module_utils/constants.py) values, plus the literal "unspecified".
_NAMED_PORTS = {"https", "smtp", "http", "dns", "pop3", "rtsp", "ftpData", "unspecified"}
def _check_vlan_encap(value: object, *, allow_unknown: bool) -> None:
if not isinstance(value, str):
raise ValueError(f"expected a string, got {type(value).__name__}")
if allow_unknown and value == "unknown":
return
m = _VLAN_ENCAP_RE.match(value)
if not m:
suffix = " or literal 'unknown'" if allow_unknown else ""
raise ValueError(f"expected 'vlan-N' or 'vxlan-N'{suffix}, got {value!r}")
prefix, n_str = m.group(1), m.group(2)
n = int(n_str)
lo, hi = (_VLAN_MIN, _VLAN_MAX) if prefix == "vlan" else (_VXLAN_MIN, _VXLAN_MAX)
if not (lo <= n <= hi):
raise ValueError(f"{prefix} id must be in [{lo},{hi}], got {n}")
def _check_ipv4_iface(value: object) -> None:
"""``a.b.c.d[/prefix]`` OR the IPv6 equivalent — real APIC uses the same
attribute (``ip``/``addr``) for either family on these props. Uses the
stdlib ``ipaddress`` module (never a regex) per design §3.1/§7 point 3.
Reason text is byte-identical to the pre-F10 hardcoded fvSubnet/
l3extSubnet ip check this replaces (design §3.2: "no behavior change,
one code path") — do not append the raw ``ipaddress`` exception text
here, existing regression tests + callers depend on the exact string."""
if not isinstance(value, str):
raise ValueError(f"expected a string, got {type(value).__name__}")
try:
ipaddress.ip_interface(value)
except ValueError:
raise ValueError("not a valid address[/prefix]") from None
def _check_ip_addr(value: object) -> None:
"""A bare IPv4 or IPv6 address, no ``/prefix`` (e.g. bgpPeerP.addr,
vnsRedirectDest.ip). Reason text is byte-identical to the pre-F10
hardcoded vnsRedirectDest ip check this replaces — see
``_check_ipv4_iface`` docstring."""
if not isinstance(value, str):
raise ValueError(f"expected a string, got {type(value).__name__}")
try:
ipaddress.ip_address(value)
except ValueError:
raise ValueError("not a valid IP address") from None
def _check_ipv4_addr(value: object) -> None:
"""IPv4-only dotted-quad (design table: l3extRsNodeL3OutAtt.rtrId is
explicitly "valid IPv4 dotted-quad (octet <=255)", not a dual-family
field like the iface/addr checks above)."""
if not isinstance(value, str):
raise ValueError(f"expected a string, got {type(value).__name__}")
try:
ipaddress.IPv4Address(value)
except ValueError as exc:
raise ValueError(f"not a valid IPv4 address: {exc}") from None
def _check_mac(value: object) -> None:
"""48-bit MAC, colon form (``HH:HH:HH:HH:HH:HH``) or Cisco dot form
(``HHHH.HHHH.HHHH``) — design §7 point 4."""
if not isinstance(value, str):
raise ValueError(f"expected a string, got {type(value).__name__}")
if _MAC_COLON_RE.match(value) or _MAC_DOT_RE.match(value):
return
raise ValueError(f"not a valid MAC address (colon or dot form), got {value!r}")
def _check_port(value: object) -> None:
"""int in [0,65535], or one of the named ports APIC accepts."""
if isinstance(value, str) and value in _NAMED_PORTS:
return
try:
n = int(value) # type: ignore[arg-type]
except (TypeError, ValueError):
raise ValueError(
f"not an integer or a named port {sorted(_NAMED_PORTS)}, got {value!r}"
) from None
if not (0 <= n <= 65535):
raise ValueError(f"port must be in [0,65535], got {n}")
_FMT_KINDS: dict[str, Callable[..., None]] = {
"ipv4_iface": _check_ipv4_iface,
"ip_addr": _check_ip_addr,
"ipv4_addr": _check_ipv4_addr,
"mac": _check_mac,
"port": _check_port,
}
def fmt(kind: str, **opts: object) -> Validator:
"""Format-check factory. *kind* selects the checker:
- ``vlan_encap``: ``vlan-N`` (N in [1,4094]) or ``vxlan-N`` (N in
[1,16777215], 24-bit VNID — NOT the vlan-N ceiling, see
``_VXLAN_MIN``/``_VXLAN_MAX``); pass ``allow_unknown=True`` for the
props whose design-table rule text explicitly says "or literal
unknown" (``fvRsPathAtt.primaryEncap``,
``fvRsDomAtt.encap``) — every other encap prop does NOT accept
"unknown" per its own row text (design §7 point 1: "needs a
per-rule unknown-ok flag").
- ``ipv4_iface``: address[/prefix], v4 or v6, via ``ipaddress.ip_interface``.
- ``ip_addr``: bare address, v4 or v6, via ``ipaddress.ip_address``.
- ``ipv4_addr``: v4-only dotted-quad, via ``ipaddress.IPv4Address``.
- ``mac``: 48-bit MAC, colon or dot form.
- ``port``: int in [0,65535] or a named L4 port.
"""
if kind == "vlan_encap":
allow_unknown = bool(opts.get("allow_unknown", False))
def _validator(cls: str, prop: str, value: object, ctx: object = None) -> None:
_check_vlan_encap(value, allow_unknown=allow_unknown)
return _validator
checker = _FMT_KINDS.get(kind)
if checker is None:
raise ValueError(f"unknown fmt kind {kind!r}")
def _validator(cls: str, prop: str, value: object, ctx: object = None) -> None:
checker(value)
return _validator
def mtu() -> Validator:
"""Literal ``inherit`` or int in [576,9216] — l3extRsPathL3OutAtt.mtu /
fvBD.mtu (design table row 2A "mtu")."""
_range = rng(576, 9216)
def _validator(cls: str, prop: str, value: object, ctx: object = None) -> None:
if isinstance(value, str) and value == "inherit":
return
_range(cls, prop, value, ctx)
return _validator
def all_of(*validators: Validator) -> Validator:
"""Compose several validators on one prop; all must pass."""
def _validator(cls: str, prop: str, value: object, ctx: object = None) -> None:
for v in validators:
v(cls, prop, value, ctx)
return _validator
# ---------------------------------------------------------------------------
# RULES — (class, prop) -> Validator
#
# Every entry below is sourced from a row in _F10_VALIDATION_DESIGN.md §2A
# (Priority 1 — server-only format/range) or §2B (Priority 2 — server-only
# enums). Comments cite the design-doc row. Cross-field / cardinality rules
# (§2C, e.g. infraPortBlk fromPort<=toPort, fvnsEncapBlk from<=to) are OUT OF
# SCOPE for this batch (P1+P2 only) — deliberately not implemented here.
# ---------------------------------------------------------------------------
RULES: dict[tuple[str, str], Validator] = {
# -- §2A Priority 1: server-only format/range --------------------------
# encap (vlan-N N in [1,4094] / vxlan-N N in [1,16777215] — see
# _VXLAN_MIN/_VXLAN_MAX) — "unknown" only where the row text says so.
("l3extRsPathL3OutAtt", "encap"): fmt("vlan_encap"),
("fvRsPathAtt", "encap"): fmt("vlan_encap"),
("fvRsPathAtt", "primaryEncap"): fmt("vlan_encap", allow_unknown=True),
("fvRsDomAtt", "encap"): fmt("vlan_encap", allow_unknown=True),
# fvnsEncapBlk from/to: same vlan-N format+range; from<=to cross-check is
# §2C (out of scope this batch).
("fvnsEncapBlk", "from"): fmt("vlan_encap"),
("fvnsEncapBlk", "to"): fmt("vlan_encap"),
# rtrId: IPv4-only dotted-quad.
("l3extRsNodeL3OutAtt", "rtrId"): fmt("ipv4_addr"),
# l3extRsPathL3OutAtt.addr: iface form (a.b.c.d[/p]), v4 or v6.
("l3extRsPathL3OutAtt", "addr"): fmt("ipv4_iface"),
# bgpPeerP.addr: iface form (a.b.c.d[/p]), v4 or v6 — F10 fix
# (independent review): real ACI subnet-based BGP peers legally carry a
# peer addr with an optional /prefix (e.g. "10.100.211.0/30"); the
# original bare-address-only ip_addr check 400-rejected those. Widened
# to ipv4_iface (ip_interface, same as l3extRsPathL3OutAtt.addr) which
# still accepts a bare address (defaults to /32 or /128) so the
# existing no-prefix canonical values keep passing too.
("bgpPeerP", "addr"): fmt("ipv4_iface"),
("bgpPeerP", "ttl"): rng(1, 255),
("bgpAsP", "asn"): rng(1, 4294967295),
("l3extRsPathL3OutAtt", "mtu"): mtu(),
("fvBD", "mtu"): mtu(),
("fvBD", "mac"): fmt("mac"),
("vnsRedirectDest", "mac"): fmt("mac"),
# infraPortBlk: int >= 1 (no stated upper bound; le-cross-check is §2C).
("infraPortBlk", "fromPort"): rng(1, None),
("infraPortBlk", "toPort"): rng(1, None),
("infraPortBlk", "fromCard"): rng(1, None),
("infraPortBlk", "toCard"): rng(1, None),
# vzEntry ports: int in [0,65535] or a named port.
("vzEntry", "dFromPort"): fmt("port"),
("vzEntry", "dToPort"): fmt("port"),
("vzEntry", "sFromPort"): fmt("port"),
("vzEntry", "sToPort"): fmt("port"),
# -- §2A Priority 1 (continued): ip already-validated (migrated, byte-
# identical semantics to the pre-F10 hardcoded branches in writes.py) --
("fvSubnet", "ip"): fmt("ipv4_iface"),
("l3extSubnet", "ip"): fmt("ipv4_iface"),
("vnsRedirectDest", "ip"): fmt("ip_addr"),
# -- §2B Priority 2: server-only enums ----------------------------------
("fvSubnet", "scope"): one_of({"private", "public", "shared"}, list_sep=","),
("fvSubnet", "ctrl"): one_of(
{"nd", "no-default-gateway", "querier", "unspecified"}, list_sep=","
),
("l3extSubnet", "scope"): one_of(
{
"import-security",
"export-rtctrl",
"import-rtctrl",
"shared-rtctrl",
"shared-security",
},
list_sep=",",
),
("l3extSubnet", "aggregate"): one_of(
{"export-rtctrl", "import-rtctrl", "shared-rtctrl"}, list_sep=","
),
("fvCtx", "pcEnfPref"): one_of({"enforced", "unenforced"}),
("fvCtx", "pcEnfDir"): one_of({"ingress", "egress"}),
("fvBD", "unkMacUcastAct"): one_of({"proxy", "flood"}),
("fvBD", "multiDstPktAct"): one_of({"bd-flood", "drop", "encap-flood"}),
("fvBD", "unkMcastAct"): one_of({"flood", "opt-flood"}),
("fvBD", "type"): one_of({"regular", "fc"}),
("fvBD", "epMoveDetectMode"): one_of({"", "garp"}),
("vzEntry", "etherT"): one_of(
{
"arp",
"fcoe",
"ip",
"ipv4",
"ipv6",
"mac_security",
"mpls_ucast",
"trill",
"unspecified",
}
),
("vzBrCP", "scope"): one_of({"context", "global", "tenant", "application-profile"}),
("fvRsDomAtt", "classPref"): one_of({"encap", "useg"}),
("fvRsDomAtt", "instrImedcy"): one_of({"immediate", "lazy"}),
# resImedcy (Resolution Immediacy) — F10 fix (independent review): real
# ACI's legal set is {immediate, lazy, pre-provision}; the original
# rule dropped "pre-provision", which is the value 100% of the
# canonical corpus's EPG-domain bindings actually use (every
# bind_epg_to_*_domain / migrate_existing_vlan* render), so the bug
# 400-rejected every one of them. Do NOT add "pre-provision" to
# instrImedcy (Deployment Immediacy) above — that enum really is just
# {immediate, lazy} per its own design-table row and 100% of the
# corpus uses "immediate".
("fvRsDomAtt", "resImedcy"): one_of({"immediate", "lazy", "pre-provision"}),
("bgpPeerP", "peerCtrl"): one_of(
{"bfd", "dis-conn-check", "nh-self", "allow-self-as", "send-com", "send-ext-com"},
list_sep=",",
),
("fvnsVlanInstP", "allocMode"): one_of({"dynamic", "static", "inherit"}),
("fvnsEncapBlk", "allocMode"): one_of({"dynamic", "static", "inherit"}),
}
+25 -28
View File
@@ -2,7 +2,6 @@
from __future__ import annotations
import ipaddress
import re
from aci_sim.build.fabric import loopback_ip, oob_ip
@@ -11,6 +10,7 @@ from aci_sim.build.neighbors import add_switch_adjacency
from aci_sim.mit.dn import pod_from_dn
from aci_sim.mit.mo import MO
from aci_sim.mit.store import MITStore
from aci_sim.rest_aci.validators import RULES
from aci_sim.topology.schema import Node
# Class -> RN template for children POSTed without an explicit ``dn``.
@@ -215,36 +215,33 @@ def _validate_node(cls: str, body: dict, path: str) -> None:
def _validate_planned(planned: list[tuple[str, dict]]) -> None:
"""Reject malformed property values a real APIC would 400 on (error-801
style) — e.g. ``fvSubnet ip=".1/24"`` or ``vnsRedirectDest ip="."``, the
exact garbage an unguarded J2 template renders when its input vars are
missing. The sim's contract is to FAIL the way real gear fails, not to
absorb it into the MIT. Only values actually PRESENT on a non-delete
write are validated (a delete needs nothing but the DN — that stays the
cleanup path for anything malformed that predates this check)."""
"""Reject malformed/out-of-range property values a real APIC would 400 on
(error-107/801 style) — e.g. ``fvSubnet ip=".1/24"``,
``l3extRsPathL3OutAtt encap="vlan-9999"``, ``bgpAsP asn="0"``. The sim's
contract is to FAIL the way real gear fails, not to absorb it into the
MIT. Only values actually PRESENT on a non-delete write are validated (a
delete needs nothing but the DN — that stays the cleanup path for
anything malformed that predates this check).
Table-driven (F10): every ``(class, prop)`` pair checked here comes from
``aci_sim.rest_aci.validators.RULES`` — see that module + design doc
``_F10_VALIDATION_DESIGN.md`` §3.2/§3.6 for the registry and its
fail-safe default-allow policy. A ``(class, prop)`` pair with NO entry in
RULES is intentionally NOT validated (unknown ⇒ allow, never reject)."""
for mo_cls, mo_attrs in planned:
if mo_attrs.get("status") == "deleted":
continue
if mo_cls in ("fvSubnet", "l3extSubnet"):
ip = mo_attrs.get("ip")
if ip is not None:
try:
ipaddress.ip_interface(ip)
except ValueError:
raise WriteValidationError(
f"Invalid value {ip!r} for property 'ip' of {mo_cls} "
f"{mo_attrs.get('dn', '')!r}: not a valid address[/prefix]"
) from None
elif mo_cls == "vnsRedirectDest":
ip = mo_attrs.get("ip")
if ip is not None:
try:
ipaddress.ip_address(ip)
except ValueError:
raise WriteValidationError(
f"Invalid value {ip!r} for property 'ip' of vnsRedirectDest "
f"{mo_attrs.get('dn', '')!r}: not a valid IP address"
) from None
for prop, value in mo_attrs.items():
validator = RULES.get((mo_cls, prop))
if validator is None or value is None:
continue # fail-safe: unregistered (class, prop) or unset value ⇒ allow
try:
validator(mo_cls, prop, value, mo_attrs)
except ValueError as reason:
raise WriteValidationError(
f"Invalid value {value!r} for property {prop!r} of {mo_cls} "
f"{mo_attrs.get('dn', '')!r}: {reason}"
) from None
def _plan_recursive(cls: str, attrs: dict, children: list, planned: list[tuple[str, dict]]) -> None: