From 99cb0a51fd782ea8f08bf511255b11731c512524 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Wed, 22 Jul 2026 21:56:41 +0000 Subject: [PATCH] Reject unknown keys when loading a chaos scenario The scenario loader read every section and member through raw.get with a default, so a mistyped key was silently ignored and the block it belonged to stayed default-constructed while the YAML still looked correct on the page. Writing "assertion:" for "assertions:" disarmed a scenario's only assertions and left it unable to return a non-zero exit code; "link_flap:" for "link_flaps:" turned off the churn the scenario existed to inject, and a calm mesh then ran under the name of a chaos test. Keys are now checked at load time, at the top level and inside every section with a fixed schema, including the netem policy bodies, the {min, max} ranges and the assertion bodies themselves. The error names the offending key, the section it appeared in, and the keys that section does understand. Three mappings are deliberately exempt because their keys are names the author chooses rather than a schema, and two sub-trees are passed through whole; all five are listed at the key sets with the reason. Adding an assertion type now means registering it, so the next person to add one gets a rejection rather than a silent no-op. That coupling is the point and it is written down where they will see it. Verified against the seventeen scenarios in the tree, all of which still load, and against fixtures for each failure mode: a top-level typo, a nested typo, a typo inside an assertion body, and a section left empty, which previously raised an unrelated AttributeError from the next line of the loader. --- testing/chaos/sim/scenario.py | 119 ++++++++++++++++++++++++++++++++-- 1 file changed, 114 insertions(+), 5 deletions(-) diff --git a/testing/chaos/sim/scenario.py b/testing/chaos/sim/scenario.py index 79a7228..139c2e0 100644 --- a/testing/chaos/sim/scenario.py +++ b/testing/chaos/sim/scenario.py @@ -239,15 +239,90 @@ class Scenario: fips_overrides: dict = field(default_factory=dict) +# Keys each fixed-schema section understands. A key absent from these sets is a +# typo, and silently ignoring it leaves the block default-constructed while the +# YAML still looks correct — a mistyped "assertion:" disarms a scenario's only +# assertions, a mistyped "link_flap:" turns off the chaos it was meant to inject. +# +# Three mappings are deliberately NOT listed, because their keys are names the +# scenario author chooses rather than a schema: netem.mutation.policies, +# link_swap.policies and topology.transport_mix. Two sub-trees are passed through +# whole and are likewise not checked: fips_overrides and topology.params. +# +# NOTE: adding a new assertion type means adding it to _ASSERTION_KEYS below and +# giving it an entry here, or scenarios using it will be rejected at load. +_TOP_KEYS = { + "scenario", "topology", "netem", "link_flaps", "traffic", "node_churn", + "peer_churn", "bandwidth", "ingress", "link_swap", "assertions", "logging", + "fips_overrides", +} +_SECTION_KEYS = { + "scenario": {"name", "seed", "duration_secs"}, + "topology": { + "num_nodes", "algorithm", "params", "ensure_connected", "subnet", + "ip_start", "default_transport", "transport_mix", + }, + "netem": {"enabled", "default_policy", "link_policies", "mutation"}, + "netem.link_policies[]": {"edges", "policy", "policy_name"}, + "netem.mutation": {"interval_secs", "fraction", "policies"}, + "link_flaps": { + "enabled", "interval_secs", "max_down_links", "down_duration_secs", + "protect_connectivity", + }, + "traffic": { + "enabled", "max_concurrent", "interval_secs", "duration_secs", + "parallel_streams", + }, + "node_churn": { + "enabled", "interval_secs", "max_down_nodes", "down_duration_secs", + "protect_connectivity", + }, + "peer_churn": {"enabled", "interval_secs", "ephemeral_fraction"}, + "bandwidth": {"enabled", "tiers_mbps"}, + "ingress": {"enabled", "tiers_kbps", "burst_bytes"}, + "link_swap": {"enabled", "interval_secs", "policies", "edges"}, + "link_swap.edges[]": {"edge", "policy"}, + "assertions": {"bloom_send_rate", "min_parent_switches"}, + "logging": {"rust_log", "output_dir"}, +} +_ASSERTION_KEYS = { + "bloom_send_rate": {"window_secs", "max_per_node"}, + "min_parent_switches": {"min_total"}, +} +_NETEM_POLICY_KEYS = { + "delay_ms", "jitter_ms", "loss_pct", "duplicate_pct", "reorder_pct", + "corrupt_pct", +} +_RANGE_KEYS = {"min", "max"} + + +def _reject_unknown(data, known, context: str): + """Raise unless every key in `data` is one the loader understands.""" + if data is None: + raise ValueError( + f"{context}: section is present but empty; give it a body or remove it" + ) + if not isinstance(data, dict): + raise ValueError(f"{context}: expected a mapping, got {type(data).__name__}") + unknown = sorted(str(k) for k in set(data) - set(known)) + if unknown: + raise ValueError( + f"{context}: unknown key(s): {', '.join(unknown)} " + f"(known: {', '.join(sorted(known))})" + ) + + def _parse_range(data, name: str) -> Range: """Parse a {min, max} dict into a Range.""" if isinstance(data, dict): + _reject_unknown(data, _RANGE_KEYS, name) return Range(min=float(data["min"]), max=float(data["max"])) raise ValueError(f"{name}: expected {{min, max}} dict, got {type(data).__name__}") -def _parse_netem_policy(data: dict) -> NetemPolicy: +def _parse_netem_policy(data: dict, name: str = "netem policy") -> NetemPolicy: """Parse a netem policy from a dict with [min, max] lists or {min, max} dicts.""" + _reject_unknown(data, _NETEM_POLICY_KEYS, name) policy = NetemPolicy() for attr in ( "delay_ms", @@ -273,16 +348,22 @@ def load_scenario(path: str) -> Scenario: with open(path) as f: raw = yaml.safe_load(f) + if raw is None: + raise ValueError(f"{path}: file is empty") + _reject_unknown(raw, _TOP_KEYS, "(top level)") + s = Scenario() # Scenario section sc = raw.get("scenario", {}) + _reject_unknown(sc, _SECTION_KEYS["scenario"], "scenario") s.name = sc.get("name", os.path.splitext(os.path.basename(path))[0]) s.seed = int(sc.get("seed", 42)) s.duration_secs = int(sc.get("duration_secs", 120)) # Topology section tc = raw.get("topology", {}) + _reject_unknown(tc, _SECTION_KEYS["topology"], "topology") s.topology.num_nodes = int(tc.get("num_nodes", 10)) s.topology.algorithm = tc.get("algorithm", "random_geometric") s.topology.params = tc.get("params", {}) @@ -298,33 +379,43 @@ def load_scenario(path: str) -> Scenario: # Netem section nc = raw.get("netem", {}) + _reject_unknown(nc, _SECTION_KEYS["netem"], "netem") s.netem.enabled = nc.get("enabled", False) if "default_policy" in nc: - s.netem.default_policy = _parse_netem_policy(nc["default_policy"]) + s.netem.default_policy = _parse_netem_policy( + nc["default_policy"], "netem.default_policy" + ) if "link_policies" in nc: for lp_data in nc["link_policies"]: + _reject_unknown( + lp_data, _SECTION_KEYS["netem.link_policies[]"], "netem.link_policies[]" + ) override = LinkPolicyOverride( edges=lp_data.get("edges", []), ) if "policy" in lp_data: - override.policy = _parse_netem_policy(lp_data["policy"]) + override.policy = _parse_netem_policy( + lp_data["policy"], "netem.link_policies[].policy" + ) if "policy_name" in lp_data: override.policy_name = lp_data["policy_name"] s.netem.link_policies.append(override) if "mutation" in nc: mc = nc["mutation"] + _reject_unknown(mc, _SECTION_KEYS["netem.mutation"], "netem.mutation") s.netem.mutation.interval_secs = _parse_range( mc.get("interval_secs", {"min": 15, "max": 30}), "netem.mutation.interval_secs" ) s.netem.mutation.fraction = float(mc.get("fraction", 0.3)) if "policies" in mc: s.netem.mutation.policies = { - name: _parse_netem_policy(pdata) + name: _parse_netem_policy(pdata, f"netem.mutation.policies.{name}") for name, pdata in mc["policies"].items() } # Link flaps section lf = raw.get("link_flaps", {}) + _reject_unknown(lf, _SECTION_KEYS["link_flaps"], "link_flaps") s.link_flaps.enabled = lf.get("enabled", False) if "interval_secs" in lf: s.link_flaps.interval_secs = _parse_range(lf["interval_secs"], "link_flaps.interval_secs") @@ -337,6 +428,7 @@ def load_scenario(path: str) -> Scenario: # Traffic section tf = raw.get("traffic", {}) + _reject_unknown(tf, _SECTION_KEYS["traffic"], "traffic") s.traffic.enabled = tf.get("enabled", False) s.traffic.max_concurrent = int(tf.get("max_concurrent", 3)) if "interval_secs" in tf: @@ -347,6 +439,7 @@ def load_scenario(path: str) -> Scenario: # Node churn section nc2 = raw.get("node_churn", {}) + _reject_unknown(nc2, _SECTION_KEYS["node_churn"], "node_churn") s.node_churn.enabled = nc2.get("enabled", False) if "interval_secs" in nc2: s.node_churn.interval_secs = _parse_range(nc2["interval_secs"], "node_churn.interval_secs") @@ -359,6 +452,7 @@ def load_scenario(path: str) -> Scenario: # Peer churn section pc = raw.get("peer_churn", {}) + _reject_unknown(pc, _SECTION_KEYS["peer_churn"], "peer_churn") s.peer_churn.enabled = pc.get("enabled", False) if "interval_secs" in pc: s.peer_churn.interval_secs = _parse_range(pc["interval_secs"], "peer_churn.interval_secs") @@ -366,6 +460,7 @@ def load_scenario(path: str) -> Scenario: # Bandwidth section bw = raw.get("bandwidth", {}) + _reject_unknown(bw, _SECTION_KEYS["bandwidth"], "bandwidth") s.bandwidth.enabled = bw.get("enabled", False) if "tiers_mbps" in bw: tiers = bw["tiers_mbps"] @@ -375,6 +470,7 @@ def load_scenario(path: str) -> Scenario: # Ingress section ig = raw.get("ingress", {}) + _reject_unknown(ig, _SECTION_KEYS["ingress"], "ingress") s.ingress.enabled = ig.get("enabled", False) if "tiers_kbps" in ig: tiers = ig["tiers_kbps"] @@ -385,18 +481,22 @@ def load_scenario(path: str) -> Scenario: # Link swap section (deterministic asymmetric link-cost flapping). ls = raw.get("link_swap", {}) + _reject_unknown(ls, _SECTION_KEYS["link_swap"], "link_swap") s.link_swap.enabled = ls.get("enabled", False) if "interval_secs" in ls: s.link_swap.interval_secs = float(ls["interval_secs"]) if "policies" in ls: s.link_swap.policies = { - name: _parse_netem_policy(pdata) + name: _parse_netem_policy(pdata, f"link_swap.policies.{name}") for name, pdata in ls["policies"].items() } if "edges" in ls: for edata in ls["edges"]: if not isinstance(edata, dict): raise ValueError("link_swap.edges entries must be dicts") + _reject_unknown( + edata, _SECTION_KEYS["link_swap.edges[]"], "link_swap.edges[]" + ) edge = str(edata.get("edge", "")) policy = str(edata.get("policy", "")) if not edge or not policy: @@ -405,20 +505,29 @@ def load_scenario(path: str) -> Scenario: # Assertions section (post-run control-socket-based checks). asrt = raw.get("assertions", {}) + _reject_unknown(asrt, _SECTION_KEYS["assertions"], "assertions") if "bloom_send_rate" in asrt: bsr = asrt["bloom_send_rate"] + _reject_unknown( + bsr, _ASSERTION_KEYS["bloom_send_rate"], "assertions.bloom_send_rate" + ) s.assertions.bloom_send_rate = BloomSendRateAssertion( window_secs=int(bsr.get("window_secs", 30)), max_per_node=int(bsr.get("max_per_node", 30)), ) if "min_parent_switches" in asrt: mps = asrt["min_parent_switches"] + _reject_unknown( + mps, _ASSERTION_KEYS["min_parent_switches"], + "assertions.min_parent_switches", + ) s.assertions.min_parent_switches = MinParentSwitchesAssertion( min_total=int(mps.get("min_total", 1)), ) # Logging section lg = raw.get("logging", {}) + _reject_unknown(lg, _SECTION_KEYS["logging"], "logging") s.logging.rust_log = lg.get("rust_log", "info") s.logging.output_dir = lg.get("output_dir", "./sim-results")