Audit: triage report + 3 CRITICAL fixes (read-only bypass, credential leak, Junos false-blocked)

backend/netsecops/adapters/policies.py

@@ -409,7 +409,12 @@ def _cmds(*patterns: str) -> tuple[CommandRule, ...]:
         HttpRule(
             "GET",
             "/api/",
-            reason="XML API reads are issued as GET where the vendor permits it",
+            reason="XML API reads are issued as GET where the vendor permits it; the "
+            "same type/action/cmd that a POST carries in its body travel in the query "
+            "string over GET, and the API is fully functional over GET (action=set, "
+            "type=commit), so the GET side needs the identical predicate or a write "
+            "refused as POST is permitted as GET (SRS §8.1.3d)",
+            body_predicate="panos_read_only",
         ),
         HttpRule(
             "POST",

backend/netsecops/adapters/session.py

@@ -12,6 +12,7 @@
 
 from __future__ import annotations
 
+import re
 import time
 import uuid
 from abc import ABC, abstractmethod
@@ -22,10 +23,28 @@
 
 from netsecops.adapters.readonly import ReadOnlyGuard
 from netsecops.core.errors import ReadOnlyViolationError
-from netsecops.core.logging import get_logger
+from netsecops.core.logging import REDACTED, get_logger
 
 log = get_logger(__name__)
 
+#: Query-string parameters that carry a credential. PAN-OS keygen-by-password puts the
+#: administrator's password and, elsewhere, the issued API key straight in the URL, so
+#: the recorded command — which is `METHOD path` and lands verbatim in the tamper-evident
+#: audit log's command_text (and is served by GET /api/v1/audit-log) — would otherwise
+#: persist and expose the cleartext secret. The value is masked before the path is ever
+#: recorded or logged; the real value still goes on the wire to the device.
+_SECRET_QUERY_PARAM = re.compile(
+    r"(?i)\b(password|passwd|pwd|secret|api[_-]?key|apikey|key|token)=([^&\s]*)"
+)
+
+
+def redact_query_secrets(path: str) -> str:
+    """Mask credential-bearing query-string values in a request path for the audit log."""
+    base, sep, query = path.partition("?")
+    if not sep:
+        return path
+    return base + sep + _SECRET_QUERY_PARAM.sub(lambda m: f"{m.group(1)}={REDACTED}", query)
+
 
 @dataclass(frozen=True, slots=True)
 class CommandResult:
@@ -224,19 +243,20 @@ async def run_all(self, commands: Sequence[str]) -> list[CommandResult]:
     async def request(
         self, method: str, path: str, *, body: Any = None, timeout: int | None = None
     ) -> HttpResult:
+        safe_path = redact_query_secrets(path)
         try:
             self.guard.check_request(method, path, body=body)
         except ReadOnlyViolationError as violation:
             await self.recorder.record_violation(
-                command=f"{method.upper()} {path}",
+                command=f"{method.upper()} {safe_path}",
                 device_id=self.device_id,
                 reason=str(violation),
                 context=violation.extra,
             )
             log.error(
                 "readonly.violation",
                 method=method,
-                path=path,
+                path=safe_path,
                 device_id=str(self.device_id) if self.device_id else None,
                 platform=self.guard.policy.platform,
             )
@@ -257,7 +277,7 @@ async def request(
             duration_ms=duration_ms,
         )
         await self.recorder.record(
-            command=f"{result.method} {path}",
+            command=f"{result.method} {safe_path}",
             device_id=self.device_id,
             succeeded=result.succeeded,
             duration_ms=duration_ms,

backend/netsecops/parsers/juniper/junos.py

@@ -561,62 +561,124 @@ def add(name: str, value: str | None, members: list[str], line: int) -> None:
                     elif rest[3] not in existing.members:
                         existing.members.append(rest[3])
 
+    @staticmethod
+    def _apply_policy_tail(rule: SecurityRule, tail: list[str], inactive: bool) -> None:
+        """Apply one `match ...` or `then ...` fragment of a policy to its rule."""
+        if tail[:1] == ["match"] and len(tail) > 2:
+            field, values = tail[1], _values(tail[2:])
+            if field == "source-address":
+                rule.src.extend(values)
+            elif field == "destination-address":
+                rule.dst.extend(values)
+            elif field == "application":
+                rule.applications.extend(values)
+        elif tail[:1] == ["then"] and len(tail) > 1:
+            action = tail[1]
+            if action in {"permit", "deny", "reject"}:
+                rule.action = "allow" if action == "permit" else action
+            elif action == "log":
+                rule.log_start = "session-init" in tail
+                rule.log_end = "session-close" in tail
+        if inactive:
+            rule.enabled = False
+
     def _policies(self, statements: list[tuple[int, list[str], bool]], result: ParseResult) -> None:
         ncm = result.ncm
         rules: dict[tuple[str, str, str], SecurityRule] = {}
-        lines: dict[tuple[str, str, str], int] = {}
+        global_rules: dict[str, SecurityRule] = {}
+        lines: dict[object, int] = {}
+        default_action: str | None = None
+        default_line: int | None = None
 
         for line, tokens, inactive in self._under(statements, "security", "policies"):
             # `from-zone trust to-zone untrust policy allow-web match source-address any`
-            if tokens[:1] != ["from-zone"] or len(tokens) < 6:
-                continue
-            if tokens[2] != "to-zone" or tokens[4] != "policy":
-                continue
-
-            src_zone, dst_zone, name = tokens[1], tokens[3], tokens[5]
-            key = (src_zone, dst_zone, name)
-
-            rule = rules.get(key)
-            if rule is None:
-                rule = SecurityRule(
-                    order=len(rules) + 1,
-                    name=name,
-                    rulebase=f"{src_zone}->{dst_zone}",
-                    src_zones=[src_zone],
-                    dst_zones=[dst_zone],
-                    # Junos has no implicit permit: a policy with no `then` is not a
-                    # permit, so the default here is the safe one and is overwritten
-                    # only by an explicit action below.
-                    action="deny",
-                    enabled=not inactive,
-                )
-                rules[key] = rule
-                lines[key] = line
-
-            tail = tokens[6:]
-            if tail[:1] == ["match"] and len(tail) > 2:
-                field, values = tail[1], _values(tail[2:])
-                if field == "source-address":
-                    rule.src.extend(values)
-                elif field == "destination-address":
-                    rule.dst.extend(values)
-                elif field == "application":
-                    rule.applications.extend(values)
-            elif tail[:1] == ["then"] and len(tail) > 1:
-                action = tail[1]
-                if action in {"permit", "deny", "reject"}:
-                    rule.action = "allow" if action == "permit" else action
-                elif action == "log":
-                    rule.log_start = "session-init" in tail
-                    rule.log_end = "session-close" in tail
-            if inactive:
-                rule.enabled = False
-
-        for key, rule in rules.items():
+            if (
+                tokens[:1] == ["from-zone"]
+                and len(tokens) >= 6
+                and tokens[2] == "to-zone"
+                and tokens[4] == "policy"
+            ):
+                src_zone, dst_zone, name = tokens[1], tokens[3], tokens[5]
+                key = (src_zone, dst_zone, name)
+                rule = rules.get(key)
+                if rule is None:
+                    rule = SecurityRule(
+                        order=len(rules) + 1,
+                        name=name,
+                        rulebase=f"{src_zone}->{dst_zone}",
+                        src_zones=[src_zone],
+                        dst_zones=[dst_zone],
+                        # Junos has no implicit permit: a policy with no `then` is not a
+                        # permit, so the default here is the safe one and is overwritten
+                        # only by an explicit action below.
+                        action="deny",
+                        enabled=not inactive,
+                    )
+                    rules[key] = rule
+                    lines[key] = line
+                self._apply_policy_tail(rule, tokens[6:], inactive)
+
+            # `global policy allow-any match ... then permit` — applies to any zone pair
+            # and is evaluated after the zone-specific rulebases. Empty src/dst zones mean
+            # "any zone" to the matcher (firewall/analysis.py:559), which is the semantics.
+            elif tokens[:1] == ["global"] and len(tokens) >= 3 and tokens[1] == "policy":
+                name = tokens[2]
+                rule = global_rules.get(name)
+                if rule is None:
+                    rule = SecurityRule(
+                        order=0,  # assigned below, after the zone-pair rules
+                        name=name,
+                        rulebase="global",
+                        src_zones=[],
+                        dst_zones=[],
+                        action="deny",
+                        enabled=not inactive,
+                    )
+                    global_rules[name] = rule
+                    lines[("global", name)] = line
+                self._apply_policy_tail(rule, tokens[3:], inactive)
+
+            # `default-policy permit-all` / `default-policy deny-all` — the device-wide
+            # action when nothing else matches. Dropping it made a `permit-all` default
+            # read as an implicit deny, so a query over otherwise-unmatched traffic
+            # returned a false `blocked` (audit CRITICAL).
+            elif tokens[:1] == ["default-policy"] and len(tokens) >= 2:
+                default_action = tokens[1]
+                default_line = line
+
+        order = 0
+
+        def _emit(rule: SecurityRule, at: int | None) -> None:
+            nonlocal order
+            order += 1
+            rule.order = order
             ncm.firewall.security_rules.append(rule)
             result.record(
                 f"firewall.security_rules.{len(ncm.firewall.security_rules) - 1}",
-                line=lines.get(key),
+                line=at,
+            )
+
+        # Zone-specific first, then global, then the device default — Junos evaluation
+        # order, which is what a first-match connectivity walk depends on.
+        for key, rule in rules.items():
+            _emit(rule, lines.get(key))
+        for gname, rule in global_rules.items():
+            _emit(rule, lines.get(("global", gname)))
+        if default_action is not None:
+            _emit(
+                SecurityRule(
+                    order=0,
+                    name="default-policy",
+                    rulebase="default",
+                    src_zones=[],
+                    dst_zones=[],
+                    src=["any"],
+                    dst=["any"],
+                    applications=["any"],
+                    action="allow" if default_action == "permit-all" else "deny",
+                    enabled=True,
+                ),
+                default_line,
             )
 
     # ──────────────────────────── version ───────────────────────────────

backend/tests/test_device_session.py

@@ -14,7 +14,7 @@
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from netsecops.adapters.policies import CISCO_IOS
+from netsecops.adapters.policies import CISCO_IOS, PANOS
 from netsecops.adapters.readonly import ReadOnlyGuard
 from netsecops.adapters.recorder import AuditingRecorder
 from netsecops.adapters.session import DeviceSession, NullRecorder
@@ -302,3 +302,75 @@ async def test_audit_chain_stays_intact_across_a_collection(
 
         result = await audit.verify_chain()
         assert result.valid and result.total >= 3
+
+
+class _FakeHttpTransport:
+    """Minimal HTTP transport for exercising DeviceSession.request without a network.
+
+    Duck-typed rather than a Transport subclass — DeviceSession.request only calls
+    ``request`` on it — so the test stays about recording, not transport plumbing.
+    """
+
+    def __init__(self) -> None:
+        self.seen: list[str] = []
+
+    async def request(self, method, path, *, body=None, timeout=None):
+        self.seen.append(f"{method} {path}")
+        return 200, "<response status='success'><result><key>ISSUED-KEY</key></result></response>"
+
+
+class TestHttpRecordingRedactsCredentials:
+    """A credential in a request URL must never reach the audit log in cleartext.
+
+    PAN-OS keygen-by-password builds `/api/?type=keygen&user=...&password=<cleartext>`,
+    and the recorded command is `METHOD path`, which lands verbatim in the tamper-evident
+    audit log's command_text and is served by GET /api/v1/audit-log. The 2026-09-30 audit
+    found it stored raw; the path is masked before it is ever recorded.
+    """
+
+    async def test_keygen_password_is_redacted_in_the_recorded_command(self) -> None:
+        recorder = NullRecorder()
+        transport = _FakeHttpTransport()
+        session = DeviceSession(
+            transport, ReadOnlyGuard(PANOS), recorder=recorder, device_id=uuid.uuid4()
+        )
+
+        await session.request(
+            "GET", "/api/?type=keygen&user=admin&password=SuperSecret123"
+        )
+
+        assert recorder.commands, "the request was not recorded"
+        recorded = recorder.commands[-1]
+        assert "SuperSecret123" not in recorded
+        assert "password=***REDACTED***" in recorded
+        # The real secret still went on the wire to the device.
+        assert "password=SuperSecret123" in transport.seen[-1]
+
+    async def test_api_key_is_redacted_in_the_recorded_command(self) -> None:
+        recorder = NullRecorder()
+        session = DeviceSession(
+            _FakeHttpTransport(), ReadOnlyGuard(PANOS), recorder=recorder, device_id=uuid.uuid4()
+        )
+
+        await session.request(
+            "GET", "/api/?type=op&cmd=<show><system><info></info></system></show>&key=LEAKED-APIKEY"
+        )
+
+        recorded = recorder.commands[-1]
+        assert "LEAKED-APIKEY" not in recorded
+        assert "key=***REDACTED***" in recorded
+
+    async def test_a_refused_write_over_get_does_not_log_its_query_secret(self) -> None:
+        # Even the violation record (a refused request) must not carry a URL secret.
+        recorder = NullRecorder()
+        session = DeviceSession(
+            _FakeHttpTransport(), ReadOnlyGuard(PANOS), recorder=recorder, device_id=uuid.uuid4()
+        )
+
+        with pytest.raises(ReadOnlyViolationError):
+            await session.request(
+                "GET", "/api/?type=config&action=set&key=LEAKED-APIKEY&element=<x/>"
+            )
+
+        assert recorder.violations
+        assert "LEAKED-APIKEY" not in recorder.violations[-1]

backend/tests/test_junos_parser.py

@@ -347,6 +347,43 @@ def test_a_policy_with_no_action_defaults_to_deny(self) -> None:
         ncm = parse(text)
         assert ncm.firewall.security_rules[0].action == "deny"
 
+    def test_a_global_policy_is_parsed_and_applies_to_any_zone(self) -> None:
+        # A `global` policy applies regardless of zone. The audit found it silently
+        # dropped, leaving a permit invisible to the connectivity walk.
+        text = (
+            "set security policies global policy allow-any match source-address any\n"
+            "set security policies global policy allow-any match destination-address any\n"
+            "set security policies global policy allow-any then permit\n"
+        )
+        rule = next(r for r in parse(text).firewall.security_rules if r.name == "allow-any")
+        assert rule.rulebase == "global"
+        assert rule.action == "allow"
+        # Empty zone sets mean "any zone" to the matcher (firewall/analysis.py).
+        assert list(rule.src_zones) == [] and list(rule.dst_zones) == []
+
+    def test_a_permit_all_default_policy_becomes_an_allow_rule(self) -> None:
+        # `default-policy permit-all` is the device-wide action when nothing else matches.
+        # Dropping it made a permissive default read as an implicit deny, so a query over
+        # otherwise-unmatched traffic returned a false `blocked` (audit CRITICAL).
+        rules = parse("set security policies default-policy permit-all\n").firewall.security_rules
+        assert rules and rules[-1].name == "default-policy"
+        assert rules[-1].action == "allow"
+        assert list(rules[-1].src) == ["any"] and list(rules[-1].dst) == ["any"]
+
+    def test_a_deny_all_default_policy_becomes_a_deny_rule(self) -> None:
+        rules = parse("set security policies default-policy deny-all\n").firewall.security_rules
+        assert rules[-1].name == "default-policy" and rules[-1].action == "deny"
+
+    def test_evaluation_order_is_zone_then_global_then_default(self) -> None:
+        text = (
+            "set security policies from-zone a to-zone b policy zp then permit\n"
+            "set security policies global policy gp then permit\n"
+            "set security policies default-policy permit-all\n"
+        )
+        rules = parse(text).firewall.security_rules
+        assert [r.name for r in rules] == ["zp", "gp", "default-policy"]
+        assert [r.order for r in rules] == [1, 2, 3]
+
     def test_a_deactivated_policy_is_disabled_not_missing(self, braces: NormalisedConfig) -> None:
         # It is still in the configuration and will come back the moment somebody
         # activates it. Dropped, nobody reviews it; reported as enabled, it is a rule