Propose-a-change reads the staged policy, and the warning names the real direction (#402)

ai/propose.py

@@ -256,12 +256,24 @@ def _refuse(reason: str) -> Dict[str, Any]:
 
 
 def propose_change(
-    request: str, before_dir: "str | Path", host: "str | None" = None
+    request: str,
+    before_dir: "str | Path",
+    host: "str | None" = None,
+    policy: Any = None,
 ) -> Dict[str, Any]:
     """Propose ONE config change from plain English, and simulate it against
     a scratch copy of `before_dir` -- see the module docstring for the shape,
     the scope limit, and why neither step here calls a model.
 
+    `policy` (#402): the user's staged policy, or None to compare against
+    Netwise's built-in example rules. Passed straight through to
+    `analyse_change()`, which already installs and clears it -- this
+    function does no policy handling of its own. Before this parameter
+    existed, propose always used the built-in rules regardless of what the
+    caller had staged, and `policy_compliance`'s own evidence then correctly
+    -- but misleadingly -- reported "no policy file was supplied", because
+    nothing here had ever told it otherwise.
+
     Never raises. Every failure -- an unparseable request, an unknown
     device, an unreachable Batfish -- is a refusal, not an exception, the
     same convention `ai/query.py`'s `answer_question()` and
@@ -394,7 +406,7 @@ def propose_change(
         after_dir = Path(tmp) / "after"
         shutil.copytree(before_dir, after_dir)
         (after_dir / "configs" / cfg_path.name).write_text(new_cfg_text, encoding="utf-8")
-        impact = change_impact.analyse_change(before_dir, after_dir, host=host)
+        impact = change_impact.analyse_change(before_dir, after_dir, host=host, policy=policy)
 
     proposed_change = {"device": device, "filter": acl_name, "line": rule_line}
     # verified and warning are two DIFFERENT facts and must never be ANDed
@@ -408,15 +420,36 @@ def propose_change(
     # know" reads as LESS alarming than the truth. Gating warning on
     # verified silently swallowed the proof in that direction.
     verified = all(f["status"] != "error" for f in impact)
-    warning = any(
-        f["status"] == "found" and f["severity"] == "high" for f in impact
-    )
+    high_severity_found = [
+        f for f in impact if f["status"] == "found" and f["severity"] == "high"
+    ]
+    warning = bool(high_severity_found)
 
     if warning:
+        # The trigger stays severity-high regardless of direction (#402) --
+        # under-warning on a genuinely serious change is the dangerous
+        # direction, same reasoning as the verified/warning split above. What
+        # changes is the WORDING: "opens access" is only true when every
+        # triggering finding actually opens something. A high-severity
+        # POLICY finding can instead be a requirement breached, which CLOSES
+        # access that should stay open -- the opposite claim, found by
+        # @ARSH871-bot reviewing #389. Checking the real `direction` field
+        # (analysis/change_impact.py) rather than assuming severity implies
+        # it is what avoids repeating that mistake here.
+        if all(f.get("direction") == "opening" for f in high_severity_found):
+            warning_text = (
+                "Warning: simulating this change shows it newly opens access "
+                "that was previously blocked -- see the impact list."
+            )
+        else:
+            warning_text = (
+                "Warning: simulating this change has a serious effect that "
+                "is not simply opening access -- see the impact list for "
+                "exactly what changed."
+            )
         answer = (
             f"Generated: {rule_line}, on {device}'s {acl_name!r} filter. "
-            "Warning: simulating this change shows it newly opens access "
-            "that was previously blocked -- see the impact list."
+            f"{warning_text}"
         )
         if not verified:
             answer += (

analysis/change_impact.py

@@ -151,6 +151,36 @@ def _direction(line_action: str) -> str:
     return "opened" if str(line_action).upper() == "PERMIT" else "closed"
 
 
+#: One vocabulary for "direction" across all three finding producers here,
+#: carried on the finding as an extra key (see _attach_direction()) rather
+#: than left for a caller to infer from severity or parse out of `detail`.
+#:
+#: WHY THIS EXISTS (#402, raised by @ARSH871-bot reviewing #389)
+#:     For the filter and reachability producers, severity already IS a
+#:     faithful proxy for direction (_OPENING_SEVERITY/_CLOSING_SEVERITY are
+#:     assigned directly from "opened"/"closed"). For the POLICY producer it
+#:     is not: severity there tracks whether a breach was newly created, and
+#:     a newly-created breach can be a prohibition violated (access opens)
+#:     OR a requirement violated (access closes -- something that must flow
+#:     no longer does). A caller that assumes "severity high means opened"
+#:     -- ai/propose.py did -- gets the direction backwards for exactly the
+#:     second case. A real field removes the need to infer it at all.
+_DIRECTION_NORMAL_FORM = {
+    "opened": "opening",
+    "closed": "closing",
+    "changed": "changed",
+    "opening": "opening",
+    "tightening": "closing",
+}
+
+
+def _attach_direction(finding: Dict[str, Any], raw_direction: str) -> Dict[str, Any]:
+    """Set the normalised `direction` key on `finding` and return it, so
+    every append site reads as one line rather than three."""
+    finding["direction"] = _DIRECTION_NORMAL_FORM[raw_direction]
+    return finding
+
+
 # Which trace dispositions count as the flow having got through.
 #
 # DELIBERATELY WIDER THAN analysis/checks/routing.py'S SET, which excludes
@@ -239,23 +269,26 @@ def _filter_change_findings(
             severity = _CLOSING_SEVERITY
 
         results.append(
-            findings.make_finding(
-                check=CHECK_NAME,
-                severity=severity,
-                device=node,
-                summary=summary,
-                # Both lines, verbatim from Batfish. The reader needs the pair
-                # to see the change; one line alone is not a diff.
-                detail=(
-                    f"After:  {row['Line_Content']} (line {row['Line_Index']}, "
-                    f"{row['Line_Action']}). "
-                    f"Before: {row['Reference_Line_Content']} "
-                    f"(line {row['Reference_Line_Index']}). "
-                    "Both match at least one common flow and treat it differently."
+            _attach_direction(
+                findings.make_finding(
+                    check=CHECK_NAME,
+                    severity=severity,
+                    device=node,
+                    summary=summary,
+                    # Both lines, verbatim from Batfish. The reader needs the
+                    # pair to see the change; one line alone is not a diff.
+                    detail=(
+                        f"After:  {row['Line_Content']} (line {row['Line_Index']}, "
+                        f"{row['Line_Action']}). "
+                        f"Before: {row['Reference_Line_Content']} "
+                        f"(line {row['Reference_Line_Index']}). "
+                        "Both match at least one common flow and treat it differently."
+                    ),
+                    source=f"{node}:{filter_name}",
+                    status="found",
+                    number=number,
                 ),
-                source=f"{node}:{filter_name}",
-                status="found",
-                number=number,
+                direction,
             )
         )
         number += 1
@@ -297,19 +330,22 @@ def _reachability_change_findings(
             severity = _OPENING_SEVERITY
 
         results.append(
-            findings.make_finding(
-                check=CHECK_NAME,
-                severity=severity,
-                device=_device_of(row["Flow"]),
-                summary=summary,
-                detail=(
-                    f"Flow {row['Flow']}. "
-                    f"After:  {_first_trace(row['Snapshot_Traces'])}. "
-                    f"Before: {_first_trace(row['Reference_Traces'])}."
+            _attach_direction(
+                findings.make_finding(
+                    check=CHECK_NAME,
+                    severity=severity,
+                    device=_device_of(row["Flow"]),
+                    summary=summary,
+                    detail=(
+                        f"Flow {row['Flow']}. "
+                        f"After:  {_first_trace(row['Snapshot_Traces'])}. "
+                        f"Before: {_first_trace(row['Reference_Traces'])}."
+                    ),
+                    source="differentialReachability",
+                    status="found",
+                    number=number,
                 ),
-                source="differentialReachability",
-                status="found",
-                number=number,
+                direction,
             )
         )
         number += 1
@@ -436,25 +472,28 @@ def _policy_change_findings(bf: Any, number: int) -> Tuple[List[Dict[str, Any]],
             not_compared.append(f"{_rule_label(rule)}: more rule changes than finding numbers")
             continue
         results.append(
-            findings.make_finding(
-                check=CHECK_NAME,
-                severity=(rule.get("violation_severity", "high") if created else "low"),
-                device=rule["node"],
-                summary=(
-                    f"This change BREAKS {_rule_label(rule)}"
-                    if created else
-                    f"This change FIXES {_rule_label(rule)}"
-                ),
-                detail=(
-                    f"{rule['kind']} on {rule['node']} filter {rule['filter']}: "
-                    f"{'held before, breached after' if created else 'breached before, holds after'}. "
-                    f"Direction: {direction}. "
-                    + (f"Violation if applied: {rule.get('violation_summary')}." if created
-                       else "The breach this rule reports is gone in the proposed config.")
+            _attach_direction(
+                findings.make_finding(
+                    check=CHECK_NAME,
+                    severity=(rule.get("violation_severity", "high") if created else "low"),
+                    device=rule["node"],
+                    summary=(
+                        f"This change BREAKS {_rule_label(rule)}"
+                        if created else
+                        f"This change FIXES {_rule_label(rule)}"
+                    ),
+                    detail=(
+                        f"{rule['kind']} on {rule['node']} filter {rule['filter']}: "
+                        f"{'held before, breached after' if created else 'breached before, holds after'}. "
+                        f"Direction: {direction}. "
+                        + (f"Violation if applied: {rule.get('violation_summary')}." if created
+                           else "The breach this rule reports is gone in the proposed config.")
+                    ),
+                    source=f"{policy_label}; rule {rule.get('number', '?')}",
+                    status="found",
+                    number=number,
                 ),
-                source=f"{policy_label}; rule {rule.get('number', '?')}",
-                status="found",
-                number=number,
+                direction,
             )
         )
         number += 1

tests/test_change_against_policy.py

@@ -92,17 +92,33 @@ def test_a_rule_with_the_same_verdict_either_side_is_unchanged_not_a_finding(mon
     )
 
 
-@pytest.mark.parametrize("kind,before,after,direction", [
-    ("prohibition", False, True, "opening"),     # forbidden traffic now permitted
-    ("prohibition", True, False, "tightening"),
-    ("requirement", False, True, "tightening"),  # required traffic now denied
-    ("requirement", True, False, "opening"),
+@pytest.mark.parametrize("kind,before,after,direction,normalised", [
+    ("prohibition", False, True, "opening", "opening"),     # forbidden traffic now permitted
+    ("prohibition", True, False, "tightening", "closing"),
+    ("requirement", False, True, "tightening", "closing"),  # required traffic now denied
+    ("requirement", True, False, "opening", "opening"),
 ])
-def test_direction_is_stated_in_this_modules_own_terms(monkeypatch, kind, before, after, direction):
+def test_direction_is_stated_in_this_modules_own_terms(monkeypatch, kind, before, after, direction, normalised):
     found, _, _ = stage(monkeypatch, [rule(1, kind)], {1: (before, after)})
     assert f"Direction: {direction}" in found[0]["evidence"]["detail"]
 
 
+@pytest.mark.parametrize("kind,before,after,normalised", [
+    ("prohibition", False, True, "opening"),
+    ("prohibition", True, False, "closing"),
+    ("requirement", False, True, "closing"),
+    ("requirement", True, False, "opening"),
+])
+def test_direction_field_is_set_regardless_of_severity(monkeypatch, kind, before, after, normalised):
+    """#402: severity alone cannot be used to infer direction for a policy
+    finding (a high-severity REQUIREMENT breach closes access, it does not
+    open it) -- this is the field a caller should read instead."""
+    found, _, _ = stage(
+        monkeypatch, [rule(1, kind, severity="high")], {1: (before, after)},
+    )
+    assert found[0]["direction"] == normalised
+
+
 # --- F-4: could-not-ask is never unchanged ----------------------------------
 
 def test_a_rule_that_cannot_be_asked_is_an_error_never_unchanged(monkeypatch):

tests/test_change_impact.py

@@ -238,6 +238,7 @@ def test_reachability_findings_rate_an_opening_high():
 
     assert len(results) == 1
     assert results[0]["severity"] == "high"
+    assert results[0]["direction"] == "opening"
     assert "could not reach before" in results[0]["summary"]
 
 
@@ -250,9 +251,48 @@ def test_reachability_findings_rate_a_tightening_medium():
     assert results[0]["severity"] == "medium", (
         "a change that blocks traffic is not a new exposure"
     )
+    assert results[0]["direction"] == "closing"
     assert "no longer reaches" in results[0]["summary"]
 
 
+class _FakeSessionWithFilterRows:
+    def __init__(self, rows):
+        self.q = SimpleNamespace(compareFilters=lambda **kw: _FakeQuestion(rows))
+
+
+def _filter_row(line_action):
+    return {
+        "Node": "rtr-us5",
+        "Filter_Name": "acl_in",
+        "Line_Action": line_action,
+        "Line_Content": f"{line_action.lower()} tcp any any eq 443",
+        "Line_Index": 0,
+        "Reference_Line_Content": "deny tcp any any eq 443",
+        "Reference_Line_Index": 0,
+    }
+
+
+def test_filter_findings_carry_the_direction_opening():
+    bf = _FakeSessionWithFilterRows([_filter_row("PERMIT")])
+    results, _ = change_impact._filter_change_findings(bf, 1)
+
+    assert len(results) == 1
+    assert results[0]["severity"] == "high"
+    assert results[0]["direction"] == "opening"
+    assert "opens traffic" in results[0]["summary"]
+
+
+def test_filter_findings_carry_the_direction_closing():
+    """The mutation guard: hardcoding 'opening' here fails, not above."""
+    bf = _FakeSessionWithFilterRows([_filter_row("DENY")])
+    results, _ = change_impact._filter_change_findings(bf, 1)
+
+    assert len(results) == 1
+    assert results[0]["severity"] == "medium"
+    assert results[0]["direction"] == "closing"
+    assert "blocks traffic" in results[0]["summary"]
+
+
 def test_reachability_findings_number_sequentially_and_report_the_next():
     bf = _FakeSessionWithFlows([
         _flow_row("ACCEPTED", "DENIED_IN"),