diff --git a/backend/netsecops/core/redaction.py b/backend/netsecops/core/redaction.py index f1b7da9..3238813 100644 --- a/backend/netsecops/core/redaction.py +++ b/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 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 ` (which fortios_secret handles). + _rule("junos_secret", r'(\bsecret\s+)("[^"]*")'), # ── VLAN trunking ─────────────────────────────────────────────────── _rule("vtp_password", r"^(\s*vtp\s+password\s+)(\S+)"), # ── Local credentials ─────────────────────────────────────────────── diff --git a/backend/netsecops/discovery/fingerprint.py b/backend/netsecops/discovery/fingerprint.py index 98d9dbb..1803581 100644 --- a/backend/netsecops/discovery/fingerprint.py +++ b/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: diff --git a/backend/netsecops/integrations/triggers.py b/backend/netsecops/integrations/triggers.py index f4f7b21..65eed83 100644 --- a/backend/netsecops/integrations/triggers.py +++ b/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 diff --git a/backend/netsecops/parsers/checkpoint/mgmt.py b/backend/netsecops/parsers/checkpoint/mgmt.py index 30cfc43..5431497 100644 --- a/backend/netsecops/parsers/checkpoint/mgmt.py +++ b/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" ] diff --git a/backend/netsecops/parsers/juniper/junos.py b/backend/netsecops/parsers/juniper/junos.py index 96ffdc9..b085d2a 100644 --- a/backend/netsecops/parsers/juniper/junos.py +++ b/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 ` line alongside its `set ` 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 {"}", "};"}: diff --git a/backend/netsecops/parsers/vmware/nsx.py b/backend/netsecops/parsers/vmware/nsx.py index 6a4999f..f715ae0 100644 --- a/backend/netsecops/parsers/vmware/nsx.py +++ b/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, ) ) diff --git a/backend/netsecops/services/aaa_correlation.py b/backend/netsecops/services/aaa_correlation.py index 34a18c3..f18c0e3 100644 --- a/backend/netsecops/services/aaa_correlation.py +++ b/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. diff --git a/backend/netsecops/services/jobs.py b/backend/netsecops/services/jobs.py index 403a2c1..9e7c4c1 100644 --- a/backend/netsecops/services/jobs.py +++ b/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, *, diff --git a/backend/netsecops/services/schedules.py b/backend/netsecops/services/schedules.py index 388e3e7..807253f 100644 --- a/backend/netsecops/services/schedules.py +++ b/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 {}), diff --git a/backend/netsecops/vuln/eol.py b/backend/netsecops/vuln/eol.py index 217ba97..c15040b 100644 --- a/backend/netsecops/vuln/eol.py +++ b/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 diff --git a/backend/netsecops/workers/scheduler.py b/backend/netsecops/workers/scheduler.py index 862468f..f8e9ee8 100644 --- a/backend/netsecops/workers/scheduler.py +++ b/backend/netsecops/workers/scheduler.py @@ -75,6 +75,15 @@ async def run(*, interval: float = TICK_SECONDS, stop: asyncio.Event | None = No 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: diff --git a/backend/tests/test_aaa_correlation.py b/backend/tests/test_aaa_correlation.py index f261daa..d0b6bd4 100644 --- a/backend/tests/test_aaa_correlation.py +++ b/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: diff --git a/backend/tests/test_checkpoint_parsers.py b/backend/tests/test_checkpoint_parsers.py index 787fe76..22747e0 100644 --- a/backend/tests/test_checkpoint_parsers.py +++ b/backend/tests/test_checkpoint_parsers.py @@ -334,6 +334,41 @@ def test_nat_direction_distinguishes_publishing_from_hiding(self, mgmt: dict[str assert nat["Publish web"]["translated"] == "Web_01" assert nat["Internal egress hide"]["direction"] == "source" + def test_a_hide_rule_with_the_original_object_as_destination_is_source(self) -> None: + """A hide/source NAT carries Check Point's built-in "Original" object as its + translated destination. Counting that non-empty value as a destination + translation misread every source NAT as publishing an internal host — a + fabricated exposure (2026-09-30 audit, FR-FW-04).""" + bundle = { + "show-nat-rulebase": { + "rulebase": [ + { + "type": "nat-rule", + "name": "Internal hide", + "original-source": {"name": "Internal_Net"}, + "translated-source": {"name": "HideBehindGW"}, + "translated-destination": {"name": "Original"}, + "original-service": {"name": "Any"}, + }, + { + "type": "nat-rule", + "name": "Publish web", + "original-destination": {"name": "Public_IP"}, + "translated-destination": {"name": "Web_01"}, + }, + ] + } + } + ncm = ( + get_parser("checkpoint_mgmt") + .parse(ParseContext(text=json.dumps(bundle))) + .to_storage() + ) + nat = {r["name"]: r for r in ncm["firewall"]["nat_rules"]} + assert nat["Internal hide"]["direction"] == "source" + assert nat["Internal hide"]["translated"] == "HideBehindGW" + assert nat["Publish web"]["direction"] == "destination" + class TestMgmtParserHealth: def test_no_section_fails_silently( diff --git a/backend/tests/test_cloud_and_sdn_parsers.py b/backend/tests/test_cloud_and_sdn_parsers.py index d7e0c19..92d3424 100644 --- a/backend/tests/test_cloud_and_sdn_parsers.py +++ b/backend/tests/test_cloud_and_sdn_parsers.py @@ -234,6 +234,36 @@ def test_a_dynamic_group_is_marked_rather_than_emptied(self, nsx: NormalisedConf assert group.type == "dynamic-group" assert group.members == [] + def test_a_mixed_static_and_dynamic_group_is_still_dynamic(self) -> None: + """A group with static members AND a tag condition is still dynamic: its true + membership also includes whatever carries the tag, which is not in the export. + Typing it a plain 'group' made the static members look like the complete set, so a + rule using it silently excluded everything the tag matches (2026-09-30 audit, + invariant 2). It is typed dynamic-group (externally resolved); the static members + are still recorded as a partial view.""" + bundle = { + "policy/api/v1/infra/domains/default/groups": { + "results": [ + { + "display_name": "web-mixed", + "id": "web-mixed", + "expression": [ + { + "resource_type": "IPAddressExpression", + "ip_addresses": ["10.0.0.5"], + }, + {"resource_type": "ConjunctionOperator", "conjunction_operator": "OR"}, + {"resource_type": "Condition", "key": "Tag", "value": "web"}, + ], + } + ] + } + } + ncm = parse_text("vmware_nsx", json.dumps(bundle)) + group = next(g for g in ncm.firewall.address_groups if g.name == "web-mixed") + assert group.type == "dynamic-group" + assert "10.0.0.5" in group.members + def test_static_group_members(self, nsx: NormalisedConfig) -> None: group = next(g for g in nsx.firewall.address_groups if g.name == "web-servers") assert group.members == ["10.20.0.0/24"] diff --git a/backend/tests/test_discovery_fingerprint.py b/backend/tests/test_discovery_fingerprint.py index 9a0dabc..0a6e690 100644 --- a/backend/tests/test_discovery_fingerprint.py +++ b/backend/tests/test_discovery_fingerprint.py @@ -112,6 +112,15 @@ def test_the_more_specific_pattern_wins(self) -> None: assert evidence.platform == "cisco_nxos" + def test_a_string_naming_two_vendors_is_ambiguous_not_a_confident_pick(self) -> None: + """A header claiming both Cisco and Fortinet named neither reliably. The old logic + let the later platform-bearing match overwrite the first vendor and flagged no + conflict; the signal must instead decline to name a vendor (2026-09-30 audit).""" + evidence = read_text(Signal.HTTP_HEADER, "Server: Cisco Systems; X-Powered-By: FortiGate") + + assert evidence.vendor is None + assert evidence.platform is None + @pytest.mark.parametrize( "text", [ diff --git a/backend/tests/test_junos_parser.py b/backend/tests/test_junos_parser.py index dd5140d..bfe9ba6 100644 --- a/backend/tests/test_junos_parser.py +++ b/backend/tests/test_junos_parser.py @@ -354,6 +354,30 @@ def test_a_deactivated_policy_is_disabled_not_missing(self, braces: NormalisedCo legacy = next(r for r in braces.firewall.security_rules if r.name == "legacy-inbound") assert legacy.enabled is False + def test_a_display_set_deactivate_line_disables_the_statement(self) -> None: + # `| display set` (the profile's primary form) renders a deactivation as its own + # `deactivate ` line, not an `inactive:` prefix. Ignoring it reported the + # policy as live (2026-09-30 audit). The deactivation of a parent path disables + # the statement under it. + text = ( + "set security policies from-zone a to-zone b policy p then permit\n" + "set security policies from-zone a to-zone b policy live then permit\n" + "deactivate security policies from-zone a to-zone b policy p\n" + ) + rules = {r.name: r for r in parse(text).firewall.security_rules} + assert rules["p"].enabled is False + assert rules["live"].enabled is True + + def test_a_radius_secret_is_redacted_not_leaked(self) -> None: + # The Junos AAA secret syntax (`secret "$9$..."`) matched no redaction rule, so it + # reached the AAA-server provenance excerpt and flowed into findings (2026-09-30). + from netsecops.core.redaction import contains_secret, redact_line + + redacted, rule = redact_line('set system radius-server 10.0.0.1 secret "$9$AbCdEfGh"') + assert "$9$AbCdEfGh" not in redacted + assert rule == "junos_secret" + assert not contains_secret(redacted) + def test_logging_on_a_rule(self, braces: NormalisedConfig) -> None: allow = next(r for r in braces.firewall.security_rules if r.name == "allow-web") assert allow.log_end is True diff --git a/backend/tests/test_schedules.py b/backend/tests/test_schedules.py index 3f91d34..0a34f04 100644 --- a/backend/tests/test_schedules.py +++ b/backend/tests/test_schedules.py @@ -290,6 +290,55 @@ async def test_a_due_schedule_creates_a_job_linked_back_to_it( assert schedule.last_job_id == job.id assert result.job_id == job.id + async def test_a_due_retention_schedule_fires_a_device_less_retention_job( + self, session: AsyncSession, schedules, actor, device + ) -> None: + """RETENTION is device-less, so the generic create() (which requires a device + scope) would reject it. Without its own branch a retention schedule could not fire + at all — which is why retention never ran (2026-09-30 audit).""" + from sqlalchemy import select + + schedule = await make(schedules, actor, device, job_type=JobType.RETENTION) + schedule.next_run_at = WEDNESDAY - timedelta(minutes=1) + await session.flush() + + [result] = await schedules.tick(now=WEDNESDAY) + + assert result.job_id is not None + job = ( + await session.execute(select(Job).where(Job.job_type == JobType.RETENTION.value)) + ).scalars().one() + assert job.schedule_id == schedule.id + assert job.scope == {} + + async def test_ensure_system_schedules_seeds_retention_once( + self, session: AsyncSession, schedules + ) -> None: + """A fresh deployment must sweep artefacts and abandoned SSO states without an + operator creating a schedule; the seed is idempotent so it does not multiply.""" + from sqlalchemy import func, select + + def count_retention(): + return session.execute( + select(func.count()) + .select_from(Schedule) + .where(Schedule.job_type == JobType.RETENTION.value) + ) + + await schedules.ensure_system_schedules() + assert (await count_retention()).scalar_one() == 1 + + await schedules.ensure_system_schedules() + assert (await count_retention()).scalar_one() == 1 + + seeded = ( + await session.execute( + select(Schedule).where(Schedule.job_type == JobType.RETENTION.value) + ) + ).scalars().one() + assert seeded.enabled is True + assert seeded.next_run_at is not None + async def test_a_schedule_not_yet_due_does_not_fire( self, session: AsyncSession, schedules, actor, device ) -> None: diff --git a/backend/tests/test_triggers.py b/backend/tests/test_triggers.py new file mode 100644 index 0000000..392383c --- /dev/null +++ b/backend/tests/test_triggers.py @@ -0,0 +1,88 @@ +"""Event derivation (integrations.triggers.scan), and the KEV promotion in particular. + +The 2026-09-30 audit found scan() filtered findings by severity BEFORE promoting KEV +matches, so a medium/low finding for a CISA-KEV CVE — "being exploited right now", the +alert the whole KEV path exists to raise — was never emitted. scan() had no direct test, +which is how it shipped. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +from sqlalchemy.ext.asyncio import AsyncSession + +from netsecops.core.rbac import Principal, Role, Scope +from netsecops.db.models.collection import Finding, FindingKind, FindingStatus +from netsecops.db.models.inventory import DeviceClass, Vendor +from netsecops.db.models.vulnerability import VulnCve +from netsecops.integrations.events import EventKind, EventSeverity +from netsecops.integrations.triggers import scan +from netsecops.services.inventory import InventoryService +from tests.conftest import make_user + + +async def _device(session: AsyncSession, ip: str = "10.7.0.1"): + user = await make_user(session, username=f"trig_{ip}", roles={Role.SECURITY_ANALYST}) + actor = Principal(id=user.id, username=user.username, roles=user.role_set, scope=Scope.all()) + return await InventoryService(session).create_device( + mgmt_ip=ip, + actor=actor, + hostname="trig-dev", + vendor=Vendor.CISCO, + platform="cisco_ios", + device_class=DeviceClass.SWITCH, + ) + + +async def _vuln_finding(session: AsyncSession, device, *, cve_id: str, severity: str) -> None: + now = datetime.now(UTC) + session.add( + Finding( + org_id=1, + device_id=device.id, + kind=FindingKind.VULN.value, + fingerprint=f"vuln:{cve_id}:{device.id}", + title=f"{cve_id} affects this device", + severity=severity, + status=FindingStatus.OPEN.value, + cve_id=cve_id, + first_seen_at=now - timedelta(hours=1), + last_seen_at=now, + ) + ) + await session.flush() + + +# Findings are fetched with created_at <= moment - COMMIT_LAG (30s); created_at is set at +# insert, so the scan moment is pushed a minute ahead. +def _moment() -> datetime: + return datetime.now(UTC) + timedelta(minutes=1) + + +class TestKevPromotionIgnoresSeverityFloor: + async def test_a_medium_kev_finding_is_promoted_to_a_critical_kev_event( + self, session: AsyncSession + ) -> None: + device = await _device(session, "10.7.0.1") + session.add(VulnCve(org_id=1, cve_id="CVE-2026-0001", kev=True)) + await _vuln_finding(session, device, cve_id="CVE-2026-0001", severity="medium") + await session.flush() + + events = await scan(session, org_id=1, now=_moment()) + + kev = [e for e in events if e.kind == EventKind.KEV_MATCHED] + assert len(kev) == 1 + assert kev[0].severity is EventSeverity.CRITICAL + assert kev[0].attributes.get("cve") == "CVE-2026-0001" + + async def test_a_medium_non_kev_vuln_finding_is_not_notified( + self, session: AsyncSession + ) -> None: + device = await _device(session, "10.7.0.2") + await _vuln_finding(session, device, cve_id="CVE-2026-0002", severity="medium") + await session.flush() + + events = await scan(session, org_id=1, now=_moment()) + + assert not any(e.attributes.get("cve") == "CVE-2026-0002" for e in events) diff --git a/backend/tests/test_vuln_eol.py b/backend/tests/test_vuln_eol.py index 1bbccd5..9337e67 100644 --- a/backend/tests/test_vuln_eol.py +++ b/backend/tests/test_vuln_eol.py @@ -142,6 +142,31 @@ def test_a_cycle_approaching_end_of_support_warns_early(self) -> None: assert result.status is LifecycleStatus.APPROACHING_END_OF_SUPPORT assert "planning now" in result.reasoning + def test_another_products_cycle_does_not_shadow_this_ones_lifecycle(self) -> None: + """FortiOS and FortiAnalyzer are versioned in lockstep. _eol_records filters only + by vendor, so both products' 7.0 cycles reach the matcher; matching purely on the + numeric cycle let FortiAnalyzer's supported 7.0 clear FortiOS's end of support + (2026-09-30 audit). The device's platform must select its own product's cycle.""" + fortios = parse_endoflife_date( + [{"cycle": "7.0", "support": "2024-03-01", "eol": "2027-03-01"}], + vendor="fortinet", + product="fortios", + ) + analyzer = parse_endoflife_date( + [{"cycle": "7.0", "support": "2030-01-01", "eol": "2032-01-01"}], + vendor="fortinet", + product="fortianalyzer", + ) + records = fortios + analyzer + + fos = assess("7.0.5", records, platform="fortios", today=NOW) + assert fos.status is LifecycleStatus.END_OF_SUPPORT + assert fos.record is not None and fos.record.product == "fortios" + + faz = assess("7.0.5", records, platform="fortianalyzer", today=NOW) + assert faz.status is LifecycleStatus.SUPPORTED + assert faz.record is not None and faz.record.product == "fortianalyzer" + def test_a_cycle_with_no_dates_is_unknown_not_supported(self, asa_records) -> None: """An entry exists; its lifecycle does not. Saying "supported" invents it.""" result = assess("9.8(4)", asa_records, platform="cisco_asa", today=NOW)