test(cron): close a blind spot in the kanban env drift guard

The AST invariant only matched `env["HERMES_KANBAN_X"] = ...` subscript
assignments, so a future dispatcher var added via `env.update({...})`,
`env.setdefault(...)`, or an annotated subscript would have slipped past
the guard and leaked into cron sessions unnoticed.

None of those shapes exist in _default_spawn today; this is about the
guard staying trustworthy as that function evolves.

Verified by injecting an unregistered var into _default_spawn one shape at
a time and requiring the guard to fail: subscript assign, annotated
assign, update(dict literal), setdefault(literal), and update(kwarg) are
all detected. Source restored byte-identical after probing.
tests/cron/ 410 passed; ruff clean.
This commit is contained in:
kshitij
2026-08-06 03:11:33 +05:30
committed by kshitij
parent 80f37e36ed
commit 145e777314
+31 -9
View File
@@ -396,16 +396,38 @@ def test_every_dispatcher_kanban_var_is_identity_gated():
injected = set()
for node in ast.walk(spawn):
if not isinstance(node, ast.Assign):
continue
for target in node.targets:
if not isinstance(target, ast.Subscript):
# env["HERMES_KANBAN_X"] = ... and the annotated form
if isinstance(node, (ast.Assign, ast.AnnAssign)):
targets = node.targets if isinstance(node, ast.Assign) else [node.target]
for target in targets:
if not isinstance(target, ast.Subscript):
continue
if ast.unparse(target.value) != "env":
continue
key = ast.unparse(target.slice).strip("\"'")
if key.startswith("HERMES_KANBAN_"):
injected.add(key)
# env.update({"HERMES_KANBAN_X": ...}) / env.setdefault("HERMES_KANBAN_X", ...)
elif isinstance(node, ast.Call):
func = ast.unparse(node.func)
if func not in ("env.update", "env.setdefault"):
continue
if ast.unparse(target.value) != "env":
continue
key = ast.unparse(target.slice).strip("\"'")
if key.startswith("HERMES_KANBAN_"):
injected.add(key)
literals = []
for arg in node.args:
if isinstance(arg, ast.Dict):
literals.extend(
k for k in arg.keys if isinstance(k, ast.Constant)
)
elif isinstance(arg, ast.Constant):
literals.append(arg)
for kw in node.keywords:
if kw.arg and kw.arg.startswith("HERMES_KANBAN_"):
injected.add(kw.arg)
for lit in literals:
if isinstance(lit.value, str) and lit.value.startswith(
"HERMES_KANBAN_"
):
injected.add(lit.value)
assert injected, "failed to parse dispatcher kanban env injection"