Audit fixes: the remaining 9 HIGH findings (parsers, analysis, wiring)

backend/netsecops/core/redaction.py

@@ -91,6 +91,13 @@ def _rule(name: str, pattern: str, *, whole_line: bool = False) -> RedactionRule
     _rule("server_key", r"^(\s*server-private\s+\S+\s+key\s+(?:\d\s+)?)(\S+)"),
     _rule("ntp_auth_key", r"(\s*ntp\s+authentication-key\s+\d+\s+\w+\s+)(\S+)"),
     _rule("key_string", r"^(\s*key-string\s+(?:\d\s+)?)(\S+)"),
+    # Junos RADIUS/TACACS+ (and LDAP) shared secrets: `... secret "$9$AbCd"` in set form,
+    # `secret "$9$AbCd"; ## SECRET-DATA` in brace form. None of the Cisco (`key`/`=`) or
+    # FortiOS (`set <key> ENC`) rules matched it, so the secret reached the AAA-server
+    # provenance excerpt and flowed into findings/reports (2026-09-30 audit). Anchored on
+    # the quoted value Junos always writes, so it does not half-fire on FortiOS's
+    # `set secret ENC <unquoted>` (which fortios_secret handles).
+    _rule("junos_secret", r'(\bsecret\s+)("[^"]*")'),
     # ── VLAN trunking ───────────────────────────────────────────────────
     _rule("vtp_password", r"^(\s*vtp\s+password\s+)(\S+)"),
     # ── Local credentials ───────────────────────────────────────────────

backend/netsecops/discovery/fingerprint.py

@@ -196,17 +196,29 @@ def read_text(signal: Signal, text: str) -> Evidence:
     if not value:
         return evidence
 
-    best: tuple[str, str | None] | None = None
-    for pattern, vendor, platform in TEXT_PATTERNS:
-        if not pattern.search(value):
-            continue
-        # A pattern naming a platform is more specific than one naming only a vendor.
-        if best is None or (platform is not None and best[1] is None):
-            best = (vendor, platform)
-
-    if best is None:
+    matches = [
+        (vendor, platform)
+        for pattern, vendor, platform in TEXT_PATTERNS
+        if pattern.search(value)
+    ]
+    if not matches:
         return evidence
-    return Evidence(signal=signal, raw=value, vendor=best[0], platform=best[1])
+
+    if len({vendor for vendor, _ in matches}) > 1:
+        # The text names more than one vendor — a header carrying both "Cisco Systems" and
+        # "FortiGate", a subject naming one vendor with another as issuer. The old logic
+        # let a later platform-bearing pattern overwrite an earlier vendor-only one, so it
+        # returned one vendor, discarded the other, and flagged NO conflict — and
+        # fingerprint() then built a confident verdict from a signal that could not decide
+        # (2026-09-30 audit). An ambiguous signal contributes no vendor rather than a
+        # wrong one; disagreement across signals is still caught by fingerprint().
+        return evidence
+
+    vendor = matches[0][0]
+    # One vendor: take the most specific match — a pattern naming a platform beats one
+    # naming only the vendor (Cisco + NX-OS is a Nexus).
+    platform = next((p for _, p in matches if p is not None), None)
+    return Evidence(signal=signal, raw=value, vendor=vendor, platform=platform)
 
 
 def fingerprint(evidence: list[Evidence]) -> Fingerprint:

backend/netsecops/integrations/triggers.py

@@ -24,7 +24,7 @@
 from datetime import UTC, datetime, timedelta
 from typing import Final
 
-from sqlalchemy import select
+from sqlalchemy import and_, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from netsecops.db.models.audit import Setting
@@ -178,7 +178,15 @@ async def scan(
         .where(
             Finding.org_id == org_id,
             Finding.created_at <= moment - COMMIT_LAG,
-            Finding.severity.in_(NOTIFIED_SEVERITIES),
+            or_(
+                Finding.severity.in_(NOTIFIED_SEVERITIES),
+                # A KEV-listed CVE is frequently scored medium/low (CVSS < 7), so the
+                # severity filter alone excluded exactly the "being exploited right now"
+                # alerts KEV exists to surface (2026-09-30 audit). Fetch every VULN finding
+                # with a CVE so the KEV promotion below can raise it; non-KEV low-severity
+                # ones are dropped in the loop, not notified.
+                and_(Finding.kind == FindingKind.VULN.value, Finding.cve_id.isnot(None)),
+            ),
         )
         .order_by(Finding.created_at)
         .limit(limit)
@@ -194,9 +202,15 @@ async def scan(
     )
 
     for finding, device in rows:
+        is_kev = finding.kind == FindingKind.VULN.value and finding.cve_id in kev
+        if not is_kev and finding.severity not in NOTIFIED_SEVERITIES:
+            # Fetched only so its KEV status could be checked: a VULN finding that is not
+            # in the catalogue and not severe enough to notify on its own.
+            continue
+
         kind = BY_KIND.get(finding.kind, EventKind.FINDING_OPENED)
         severity = severity_of(finding.severity)
-        if finding.kind == FindingKind.VULN.value and finding.cve_id in kev:
+        if is_kev:
             # "Being exploited right now" is a different page at 3am from "severe", so it
             # gets its own kind and is raised to critical whatever the CVSS said.
             kind = EventKind.KEV_MATCHED

backend/netsecops/parsers/checkpoint/mgmt.py

@@ -510,7 +510,12 @@ def _parse_nat(self, bundle: dict[str, Any], result: ParseResult) -> None:
         firewall = result.ncm.firewall
         for entry in self._flatten(_as_list(response.get("rulebase"))):
             original = _names(entry.get("original-source")) or ["any"]
-            translated_dst = _names(entry.get("translated-destination"))
+            # "Original" is Check Point's built-in "unchanged" object. A hide/source-NAT
+            # rule carries it as the translated destination, so counting a non-empty
+            # translated-destination as a destination translation misread every source
+            # NAT as publishing an internal host — a fabricated exposure (FR-FW-04).
+            # Only a translation to something other than "Original" is a real one.
+            translated_dst = [n for n in _names(entry.get("translated-destination")) if n != "Original"]
             rule = NatRule(
                 order=len(firewall.nat_rules) + 1,
                 name=str(entry.get("name") or f"NAT {len(firewall.nat_rules) + 1}"),
@@ -538,7 +543,7 @@ def _parse_nat(self, bundle: dict[str, Any], result: ParseResult) -> None:
             rule.original_destination = [
                 n for n in _names(entry.get("original-destination")) if n != "Any"
             ]
-            rule.translated_destination = [n for n in translated_dst if n != "Original"]
+            rule.translated_destination = list(translated_dst)  # already excludes "Original"
             rule.translated_source = [
                 n for n in _names(entry.get("translated-source")) if n != "Original"
             ]

backend/netsecops/parsers/juniper/junos.py

@@ -100,9 +100,25 @@ def to_set_statements(lines: list[str]) -> list[tuple[int, list[str], bool]]:
     the one that *terminated* it — the line a finding should cite.
 
     A capture already in `set` form is passed through: every line starting with `set` is
-    taken as-is, and `delete`/`deactivate` lines are ignored because a `| display set`
-    dump contains none and anything else is not a configuration.
+    taken as-is. In that flat form `| display set` renders an inactive statement as a
+    separate `deactivate <path>` line alongside its `set <path>` line — NOT as an
+    `inactive:` prefix — so those lines are collected first and any `set` statement whose
+    path lies under a deactivated one is marked inactive. `delete` lines are ignored.
     """
+    # `display set` puts the deactivation on its own line, and not necessarily adjacent to
+    # the statement it disables, so collect them up front. A `set` under any of these
+    # paths (the path itself or a descendant — deactivating a block deactivates its
+    # subtree) is inactive. Ignoring them reported deactivated services, policies and
+    # interfaces as live on the profile's primary collection path (2026-09-30 audit).
+    deactivated_paths: list[list[str]] = [
+        _tokens(stripped[len("deactivate ") :])
+        for line in lines
+        if (stripped := line.strip()).startswith("deactivate ")
+    ]
+
+    def _under_deactivated(tokens: list[str]) -> bool:
+        return any(prefix and tokens[: len(prefix)] == prefix for prefix in deactivated_paths)
+
     statements: list[tuple[int, list[str], bool]] = []
     stack: list[str] = []
     #: One entry per open brace: how many tokens it pushed, and whether it was
@@ -134,9 +150,15 @@ def to_set_statements(lines: list[str]) -> list[tuple[int, list[str], bool]]:
             inactive = True
             text = text[len(_INACTIVE) :].strip()
 
+        if text.startswith(("deactivate ", "delete ")):
+            # `deactivate` was consumed up front; `delete` is not configuration. Neither
+            # is a statement to emit.
+            continue
+
         if text.startswith("set "):
             # Already flat. `display set` never nests, so the stack is irrelevant here.
-            statements.append((number, _tokens(text[4:]), inactive))
+            tokens = _tokens(text[4:])
+            statements.append((number, tokens, inactive or _under_deactivated(tokens)))
             continue
 
         if text in {"}", "};"}:

backend/netsecops/parsers/vmware/nsx.py

@@ -175,7 +175,14 @@ def _groups(self, bundle: dict[str, Any], result: ParseResult) -> None:
                 ncm.firewall.address_groups.append(
                     NetworkObject(
                         name=str(name),
-                        type="dynamic-group" if dynamic and not members else "group",
+                        # Any dynamic condition makes the group externally resolved — its
+                        # real membership is whatever currently carries the tag, which is
+                        # not in the export. A group that ALSO lists static members is
+                        # still dynamic: typing it a plain 'group' made the static members
+                        # look like the complete set, so a rule using it resolved to just
+                        # those and silently excluded everything the tag matches
+                        # (invariant 2). Only a purely static group is 'group'.
+                        type="dynamic-group" if dynamic else "group",
                         members=members,
                     )
                 )

backend/netsecops/services/aaa_correlation.py

@@ -304,6 +304,12 @@ async def correlate(self, *, org_id: int = 1) -> AaaCorrelationReport:
 
             if not report.registration_analysed:
                 continue
+            if snapshot is None or aaa is None:
+                # Never collected, or no AAA block was parsed — already counted as
+                # not-evaluated above. We cannot know whether it uses central auth, so it
+                # must not be reported as unregistered/local-only: that fabricates a HIGH
+                # finding from absent data (2026-09-30 audit).
+                continue
             if device.device_class == DeviceClass.MANAGER.value:
                 # A manager authenticates its administrators, not itself, so it is not
                 # expected on a RADIUS client list.

backend/netsecops/services/jobs.py

@@ -351,6 +351,46 @@ async def create_notify(
         await self.session.flush()
         return job
 
+    async def create_retention(
+        self,
+        *,
+        actor: Principal,
+        schedule_id: uuid.UUID | None = None,
+        idempotency_key: str | None = None,
+        org_id: int = 1,
+    ) -> Job:
+        """Queue a data-retention run: purge artefacts past the window (FR-ADM-01) and
+        sweep abandoned SSO login states (FR-AUTH-04).
+
+        Device-less, like the notification and feed-sync jobs. It had an executor
+        (`_run_retention`) but no creator, so nothing ever queued it — artefact retention
+        never applied and the unauthenticated SSO-state table grew without bound
+        (2026-09-30 audit). It is queued by the seeded system schedule.
+        """
+        if idempotency_key:
+            existing = (
+                await self.session.execute(
+                    select(Job).where(Job.idempotency_key == idempotency_key)
+                )
+            ).scalar_one_or_none()
+            if existing is not None:
+                return existing
+
+        job = Job(
+            org_id=org_id,
+            job_type=JobType.RETENTION.value,
+            status=JobStatus.QUEUED.value,
+            scope={},
+            requested_by_id=actor.id,
+            schedule_id=schedule_id,
+            idempotency_key=idempotency_key,
+            correlation_id=correlation_id.get(),
+            stats={},
+        )
+        self.session.add(job)
+        await self.session.flush()
+        return job
+
     async def create_report(
         self,
         *,

backend/netsecops/services/schedules.py

@@ -228,6 +228,53 @@ async def create(
         )
         return schedule
 
+    #: Name of the seeded data-retention schedule.
+    SYSTEM_RETENTION_NAME = "System: data retention"
+
+    async def ensure_system_schedules(self) -> None:
+        """Seed the recurring system schedules a deployment needs but no operator creates.
+
+        The data-retention sweep (FR-ADM-01, FR-AUTH-04) has an executor and a creator, but
+        nothing scheduled it — so artefact retention never applied and the unauthenticated
+        SSO-state table grew without bound (2026-09-30 audit). Idempotent: seeds it only
+        when the org has no retention schedule, so an operator may disable or retune it and
+        it will not reappear.
+        """
+        existing = (
+            await self.session.execute(
+                select(Schedule.id).where(
+                    Schedule.org_id == self.org_id,
+                    Schedule.job_type == JobType.RETENTION.value,
+                )
+            )
+        ).first()
+        if existing is not None:
+            return
+
+        schedule = Schedule(
+            org_id=self.org_id,
+            name=self.SYSTEM_RETENTION_NAME,
+            description=(
+                "Purges collection artefacts past the retention window and sweeps "
+                "abandoned SSO login states. Created automatically; disable it if "
+                "retention is managed elsewhere."
+            ),
+            job_type=JobType.RETENTION.value,
+            scope={},
+            cron="15 3 * * *",  # daily, 03:15
+            timezone="UTC",
+            enabled=True,
+            created_by_id=None,
+        )
+        schedule.next_run_at, _ = next_occurrence(schedule, after=datetime.now(UTC))
+        self.session.add(schedule)
+        await self.session.flush()
+        log.info(
+            "scheduler.system_schedule_seeded",
+            name=self.SYSTEM_RETENTION_NAME,
+            org_id=self.org_id,
+        )
+
     async def update(self, schedule: Schedule, *, actor: Principal, **changes: Any) -> Schedule:
         if (cron := changes.get("cron")) is not None:
             validate_cron(cron)
@@ -432,6 +479,16 @@ async def _create_job(self, schedule: Schedule) -> Job:
                 org_id=self.org_id,
             )
 
+        if job_type is JobType.RETENTION:
+            # Device-less, like notify and feed-sync — the generic create() below requires
+            # a device scope and would reject it. Without this branch a retention schedule
+            # could not fire at all.
+            return await jobs.create_retention(
+                actor=_scheduler_principal(schedule),
+                schedule_id=schedule.id,
+                org_id=self.org_id,
+            )
+
         return await jobs.create(
             job_type=job_type,
             scope=JobScope.from_json(schedule.scope or {}),

backend/netsecops/vuln/eol.py

@@ -221,6 +221,36 @@ def assess(
     return LifecycleAssessment(status, _explain(record, status, version), record)
 
 
+def _normalise_product(text: str) -> str:
+    """Lower-cased, alphanumerics only, for comparing a product to a platform id."""
+    return "".join(character for character in text.lower() if character.isalnum())
+
+
+def _records_for_product(records: list[EolRecord], platform: str | None) -> list[EolRecord]:
+    """Records whose product matches this device's platform, or all if none can be told.
+
+    `_eol_records` filters only by vendor, so a vendor with several products versioned in
+    lockstep — FortiOS 7.0 (end of support) and FortiAnalyzer 7.0 (supported) — returns
+    both, and `_best_cycle` matching purely on the numeric cycle let one product's
+    lifecycle clear a real end-of-support finding on another (2026-09-30 audit). Restrict
+    to records whose product aligns with the platform (`asa` within `cisco_asa`, `FortiOS`
+    equal to `fortios`), by containment either way since a product name is often a
+    component of the platform id. Falls back to all records when the platform is unknown or
+    no product aligns, so a dataset whose product names do not match the platform scheme
+    still works as before.
+    """
+    if not platform:
+        return records
+    plat = _normalise_product(platform)
+    matched = [
+        record
+        for record in records
+        if record.product
+        and ((product := _normalise_product(record.product)) in plat or plat in product)
+    ]
+    return matched or records
+
+
 def _best_cycle(
     version: str, records: list[EolRecord], *, platform: str | None
 ) -> EolRecord | None:
@@ -238,7 +268,7 @@ def _best_cycle(
     best: EolRecord | None = None
     best_depth = -1
 
-    for record in records:
+    for record in _records_for_product(records, platform):
         cycle = parse(record.cycle, platform=platform)
         if cycle is None:
             continue

backend/netsecops/workers/scheduler.py

@@ -75,6 +75,15 @@
     log.info("scheduler.started", interval_seconds=interval)
     halt = stop or asyncio.Event()
 
+    # Seed the recurring system schedules (data retention) once at startup, so a fresh
+    # deployment sweeps artefacts and abandoned SSO states without an operator creating a
+    # schedule by hand. Idempotent, and a failure here must not stop the scheduler.
+    try:
+        async with session_scope() as session:
+            await ScheduleService(session).ensure_system_schedules()
+    except Exception as exc:  # noqa: BLE001 - a seed failure must not stop the loop
+        log.warning("scheduler.seed_failed", error=str(exc))
+
     while not halt.is_set():
         started = datetime.now(UTC)
         try:

backend/tests/test_aaa_correlation.py

@@ -206,6 +206,25 @@ async def test_a_device_on_no_client_list_is_reported(
 
         assert [d.hostname for d in report.unregistered_devices] == ["forgotten-sw"]
 
+    async def test_a_never_collected_device_is_not_reported_as_unregistered(
+        self, session: AsyncSession, actor: Principal
+    ) -> None:
+        """Registration analysis runs (an AAA server was collected), but a device in
+        inventory that was never collected has no config to say whether it uses central
+        auth. Reporting it as unregistered/local-only fabricates a HIGH finding from absent
+        data (2026-09-30 audit); it is not-evaluated, not unregistered."""
+        ise = await add_device(
+            session, actor, ip="10.100.0.55", hostname="ise-07", platform="cisco_ise"
+        )
+        await snapshot(session, ise, server_ncm("ise", [client("known-sw", "198.51.100.41")]))
+        await add_device(session, actor, ip="198.51.100.42", hostname="never-collected-sw")
+
+        report = await AaaCorrelationService(session).correlate()
+
+        assert report.registration_analysed is True
+        assert "never-collected-sw" not in [d.hostname for d in report.unregistered_devices]
+        assert report.coverage.devices_not_evaluated >= 1
+
     async def test_nothing_is_claimed_when_no_aaa_server_was_collected(
         self, session: AsyncSession, actor: Principal
     ) -> None: