diff --git a/backend/netsecops/core/config.py b/backend/netsecops/core/config.py index 4f05eab..5bb3f64 100644 --- a/backend/netsecops/core/config.py +++ b/backend/netsecops/core/config.py @@ -400,6 +400,14 @@ def _guard_oidc(self) -> Self: def _guard_production(self) -> Self: """Fail closed on unsafe production configuration.""" if self.env is Environment.PROD: + if "secret_key" not in self.model_fields_set: + # secret_key has a default_factory that mints a random key per process. + # Convenient in dev; in production it fails open — with more than one + # worker a token signed by one worker fails to verify on another, and + # every restart silently invalidates every session. An unset key is a + # misconfiguration, and must fail loudly rather than fall back to an + # ephemeral one that only looks like it works from a single worker. + raise ValueError("SECRET_KEY must be set explicitly in production") if self.debug: raise ValueError("debug must be False in production") if not self.cookie_secure: diff --git a/backend/netsecops/core/security.py b/backend/netsecops/core/security.py index 515ed42..ec5434d 100644 --- a/backend/netsecops/core/security.py +++ b/backend/netsecops/core/security.py @@ -309,6 +309,34 @@ def verify_totp(secret: str, code: str, settings: Settings | None = None) -> boo return build_totp(secret, settings).verify(code, valid_window=settings.mfa_totp_valid_window) +def matched_totp_step(secret: str, code: str, settings: Settings | None = None) -> int | None: + """The absolute time-step a valid code belongs to, or ``None`` if it is not valid. + + ``verify_totp`` answers only yes/no, which is not enough to reject a replay. A code is + accepted anywhere in ``[S - window, S + window]``, so it stays valid across several + steps; a replay guard that records the *current* step rather than the step the code + belongs to leaves the code replayable for the rest of its own window. This returns the + step the code actually matched — the value the guard must store and compare against — + so a used code cannot be presented again while it is still inside its window. + """ + settings = settings or get_settings() + code = code.strip().replace(" ", "") + if not code.isdigit() or len(code) != settings.mfa_totp_digits: + return None + + totp = build_totp(secret, settings) + period = settings.mfa_totp_period_seconds + current_step = int(datetime.now(UTC).timestamp()) // period + window = settings.mfa_totp_valid_window + + matched: int | None = None + for offset in range(-window, window + 1): + step = current_step + offset + if totp.verify(code, for_time=step * period, valid_window=0): + matched = step # ascending, so the highest matching step wins + return matched + + def generate_recovery_codes(count: int = 10) -> list[str]: """Single-use recovery codes, issued alongside TOTP enrolment.""" return [f"{secrets.token_hex(4)}-{secrets.token_hex(4)}" for _ in range(count)] diff --git a/backend/netsecops/db/migrations/versions/0017_oidc_login_states.py b/backend/netsecops/db/migrations/versions/0017_oidc_login_states.py index 56c5960..091027d 100644 --- a/backend/netsecops/db/migrations/versions/0017_oidc_login_states.py +++ b/backend/netsecops/db/migrations/versions/0017_oidc_login_states.py @@ -24,6 +24,12 @@ The index on `expires_at` is for the sweep that removes abandoned sign-ins: a browser that never comes back leaves a row, and without the index that cleanup is a sequential scan on a table every login writes to. + +`org_id` carries `index=True` through `OrgMixin` (DATA-04), like every other tenant-scoped +table, so the model declares `ix_oidc_login_states_org_id`. It is created here rather than +in a later corrective migration (as 0011 had to do for `reports`): this is the head, so the +index can go in where it belongs instead of the model and the database disagreeing and +`alembic check` failing the build (C-5). """ from __future__ import annotations @@ -70,9 +76,13 @@ def upgrade() -> None: ) op.create_index("ix_oidc_login_states_state", "oidc_login_states", ["state"], unique=True) op.create_index("ix_oidc_login_states_expires_at", "oidc_login_states", ["expires_at"]) + op.create_index( + op.f("ix_oidc_login_states_org_id"), "oidc_login_states", ["org_id"], unique=False + ) def downgrade() -> None: + op.drop_index(op.f("ix_oidc_login_states_org_id"), table_name="oidc_login_states") op.drop_index("ix_oidc_login_states_expires_at", table_name="oidc_login_states") op.drop_index("ix_oidc_login_states_state", table_name="oidc_login_states") op.drop_table("oidc_login_states") diff --git a/backend/netsecops/discovery/executor.py b/backend/netsecops/discovery/executor.py index dc48a72..136d702 100644 --- a/backend/netsecops/discovery/executor.py +++ b/backend/netsecops/discovery/executor.py @@ -360,6 +360,15 @@ async def _record_batch( summary.addresses_probed += 1 summary.probes_sent += result.probes_sent + # Run-level caveats (ICMP going away mid-run) ride on whichever host was being + # probed when they occurred, and that host is frequently one that does not + # respond. Collected before the liveness short-circuit below, or the very + # degradation that makes a run partial is the note most likely to be dropped — + # and the run would then report "found N hosts" with no sign echo had stopped. + for note in result.notes: + if note not in summary.notes: + summary.notes.append(note) + if not result.responded: # Silence is the normal answer and is not recorded. A row per dead # address would bury the queue under the network's empty space. @@ -377,10 +386,6 @@ async def _record_batch( hostname=result.hostname, ) - for note in result.notes: - if note not in summary.notes: - summary.notes.append(note) - if scope_row.auto_onboard: device = await self.reviews.auto_onboard(host, actor=actor) if device is not None: diff --git a/backend/netsecops/discovery/transport.py b/backend/netsecops/discovery/transport.py index dcea620..b81bb13 100644 --- a/backend/netsecops/discovery/transport.py +++ b/backend/netsecops/discovery/transport.py @@ -110,6 +110,12 @@ class ProbeOutcome: probe: Probe responded: bool + #: The host proved it is on the network even if this probe did not succeed. A TCP RST + #: (connection refused) is the case: the port is closed, so `responded` is False and + #: the port is not opened or followed up, but the host is provably up. Distinct from + #: `responded` precisely so a refused port does not read as an open one. A successful + #: probe implies liveness and sets both. + host_alive: bool = False #: Whatever the host volunteered: a banner line, a certificate subject, a header #: block. Kept verbatim — the review queue shows a person the raw string, because #: "Cisco, 70%" is not something anybody can check. @@ -276,9 +282,21 @@ async def send_tcp_connect(probe: Probe, *, timeout: float = DEFAULT_PROBE_TIMEO reader, writer = await asyncio.open_connection(probe.host, probe.port) await _close(writer) del reader - return ProbeOutcome(probe=probe, responded=True) + return ProbeOutcome(probe=probe, responded=True, host_alive=True) except TimeoutError: return ProbeOutcome(probe=probe, responded=False, detail="Connect timed out.") + except ConnectionRefusedError: + # A refused connection is an RST from the host itself: the port is closed, but + # the host is provably on the network. Timeouts and unreachable errors are + # indistinguishable from dead space; a refusal is not. Recorded as not-open yet + # alive, so a hardened device that drops ICMP and closes every scanned port is + # still found rather than filed as absent. + return ProbeOutcome( + probe=probe, + responded=False, + host_alive=True, + detail="Connection refused (host up, port closed).", + ) except OSError as exc: return ProbeOutcome(probe=probe, responded=False, detail=str(exc)) @@ -516,6 +534,9 @@ async def probe(self, address: str) -> HostResult: timeout=self.timeout, ) result.probes_sent += 1 + # A refusal proves the host is up without opening the port: it counts towards + # liveness but is not an open port and gets no follow-up read. + result.responded |= connect.host_alive if not connect.responded: continue diff --git a/backend/netsecops/parsers/arista/eos.py b/backend/netsecops/parsers/arista/eos.py index e0e0e5d..d28701f 100644 --- a/backend/netsecops/parsers/arista/eos.py +++ b/backend/netsecops/parsers/arista/eos.py @@ -53,9 +53,14 @@ #: `ip address 10.10.10.2/30` — CIDR, unlike IOS. _CIDR_ADDRESS = re.compile(r"^\s*ip address (\d{1,3}(?:\.\d{1,3}){3}/\d{1,2})") -#: `ip route 0.0.0.0/0 10.10.10.1 [name X] [tag N]` +#: `ip route 0.0.0.0/0 10.10.10.1 [name X] [tag N]`, and the interface+gateway form +#: `ip route 0.0.0.0/0 Ethernet1 10.1.1.1 [tag N]` where the egress interface and the +#: next-hop gateway are both stated. The second group only matches a trailing address, +#: so an administrative distance (a bare number) or `name`/`tag` keywords do not capture. _ROUTE = re.compile( - r"^ip route (?:vrf (?P\S+) )?(?P\d{1,3}(?:\.\d{1,3}){3}/\d{1,2})\s+(?P\S+)" + r"^ip route (?:vrf (?P\S+) )?(?P\d{1,3}(?:\.\d{1,3}){3}/\d{1,2})" + r"\s+(?P\S+)" + r"(?:\s+(?P\d{1,3}(?:\.\d{1,3}){3}))?" ) #: `username admin privilege 15 role network-admin secret sha512 $6$…` @@ -321,15 +326,20 @@ def _routing(self, parse: CiscoConfParse, result: ParseResult) -> None: found = _ROUTE.match(obj.text.strip()) if not found: continue - next_hop = found.group("next_hop") - # EOS accepts an interface where IOS would want an address. Stored as a next - # hop it becomes an edge to a device that does not exist. - is_address = re.match(r"^\d{1,3}(?:\.\d{1,3}){3}$", next_hop) is not None + first = found.group("first") + gateway = found.group("gateway") + # EOS accepts an interface where IOS would want an address, and it accepts + # both together (`ip route `). Read the first token as an + # interface unless it is itself an address; the real next hop is then the + # trailing gateway. Storing an interface name as a next hop would make an edge + # to a device that does not exist; dropping the trailing gateway would make + # the route read as directly-attached and lose the forwarding hop entirely. + first_is_address = re.match(r"^\d{1,3}(?:\.\d{1,3}){3}$", first) is not None ncm.routing.routes.append( Route( destination=found.group("destination"), - next_hop=next_hop if is_address else None, - interface=None if is_address else next_hop, + next_hop=first if first_is_address else gateway, + interface=None if first_is_address else first, protocol="static", vrf=found.group("vrf"), ) diff --git a/backend/netsecops/parsers/cloud/aws.py b/backend/netsecops/parsers/cloud/aws.py index 2571765..9e43288 100644 --- a/backend/netsecops/parsers/cloud/aws.py +++ b/backend/netsecops/parsers/cloud/aws.py @@ -217,14 +217,28 @@ def _groups(self, groups: list[dict[str, Any]], result: ParseResult) -> None: # Prefix lists are AWS-managed address sets — `pl-...` for S3, DynamoDB and the # rest — and their contents are not in this export either. Same treatment as a # security group, for the same reason. + # + # A rule may also reference a group by id (`sg-...`) that is not in this export at + # all: one in a peered VPC or another account. Its membership is externally + # resolved, exactly like a group whose instances live in describe-instances — not + # a collection gap the operator can close by re-collecting this account. Left + # untyped it would fall to the MISSING bucket and advise a re-collection that can + # never produce it. known = {o.name for o in ncm.firewall.address_groups} for rule in ncm.firewall.security_rules: for member in (*rule.src, *rule.dst): - if member.startswith("pl-") and member not in known: + if member in known: + continue + if member.startswith("pl-"): known.add(member) ncm.firewall.address_groups.append( NetworkObject(name=member, type="prefix-list", members=[]) ) + elif member.startswith("sg-"): + known.add(member) + ncm.firewall.address_groups.append( + NetworkObject(name=member, type="security-group", members=[]) + ) ncm.firewall.zones = sorted(vpcs) # There is no single hostname for a VPC. The account's groups are the unit, and diff --git a/backend/netsecops/parsers/cloud/azure.py b/backend/netsecops/parsers/cloud/azure.py index 26a8508..3cec3a6 100644 --- a/backend/netsecops/parsers/cloud/azure.py +++ b/backend/netsecops/parsers/cloud/azure.py @@ -36,6 +36,7 @@ from __future__ import annotations +import ipaddress import json from typing import Any @@ -51,6 +52,19 @@ _ANY = frozenset({"*"}) +def _is_address_literal(member: str) -> bool: + """Whether a rule member is a CIDR or bare address rather than a service tag. + + A dot is not proof: regional service tags (Storage.EastUS, Sql.WestEurope) all carry + one. Only a value that actually parses as an IP address or network is a literal. + """ + try: + ipaddress.ip_network(member, strict=False) + except ValueError: + return False + return True + + def _payload(bundle: Any) -> list[dict[str, Any]]: """The NSGs in an export, whatever it was wrapped in. @@ -226,9 +240,13 @@ def _service_tags(result: ParseResult) -> None: for member in (*rule.src, *rule.dst): if member == "any" or member in known: continue - # Anything that is not an address literal is a tag. CIDRs and bare - # addresses contain a dot or a colon; tags never do. - if any(ch in member for ch in ".:/"): + # Anything that is not an address literal is a tag. A dot or colon is not + # proof of an address: regional service tags always carry a dot + # (Storage.EastUS, Sql.WestEurope, AzureCloud.westus2). Only something + # that actually parses as an address or network is a literal; everything + # else stands for a set Microsoft defines elsewhere and is typed a tag, + # rather than falling to the unresolved/MISSING bucket as a fake gap. + if _is_address_literal(member): continue known.add(member) ncm.firewall.address_objects.append( diff --git a/backend/netsecops/parsers/vmware/nsx.py b/backend/netsecops/parsers/vmware/nsx.py index 6a4999f..1b3dd8a 100644 --- a/backend/netsecops/parsers/vmware/nsx.py +++ b/backend/netsecops/parsers/vmware/nsx.py @@ -155,7 +155,13 @@ def _groups(self, bundle: dict[str, Any], result: ParseResult) -> None: if "/groups" not in endpoint: continue for item in _results(payload): - name = item.get("display_name") or item.get("id") + # Keyed on `id`, because that is the last path segment a rule's + # source_groups/destination_groups reduce to via _leaf. Preferring + # display_name (which differs from id for any API/Terraform-created group, + # e.g. id='grp-4821', display_name='Web Servers') catalogues the group + # under a name no rule references, so every rule using it loses its + # members and drops out of the overlap and shadowing analysis. + name = item.get("id") or item.get("display_name") if not name: continue @@ -186,7 +192,10 @@ def _services(self, bundle: dict[str, Any], result: ParseResult) -> None: if "/services" not in endpoint: continue for item in _results(payload): - name = item.get("display_name") or item.get("id") + # Keyed on `id` for the same reason as _groups: a rule's `services` + # reduce to the path's last segment (the id) via _leaf, so a service + # catalogued under a differing display_name would never resolve. + name = item.get("id") or item.get("display_name") if not name: continue ports: list[str] = [] diff --git a/backend/netsecops/services/aaa_correlation.py b/backend/netsecops/services/aaa_correlation.py index 34a18c3..ad05ea0 100644 --- a/backend/netsecops/services/aaa_correlation.py +++ b/backend/netsecops/services/aaa_correlation.py @@ -163,6 +163,11 @@ class AaaCorrelationReport: #: How many AAA servers contributed a client list. Zero means the orphan and #: registration conclusions could not be drawn at all. servers_examined: int = 0 + #: Known AAA servers whose client list came back empty or unparsed. Their clients are + #: unknown, not absent — a device could be registered on one without appearing here — + #: so their presence withholds the unregistered-device verdict rather than risk a + #: false "unregistered/HIGH" for a device that only authenticates against them. + servers_with_unreadable_clients: list[str] = field(default_factory=list) #: Client entries whose source masks the shared secret, so reuse is unknowable for #: them. FR-AAA-05 requires this be reported rather than counted as "not reused". secrets_not_exposable: int = 0 @@ -173,6 +178,16 @@ class AaaCorrelationReport: def registration_analysed(self) -> bool: return self.servers_examined > 0 + @property + def registration_reliable(self) -> bool: + """Whether an unregistered-device verdict can be trusted. + + A server whose client list could not be read leaves a hole exactly the shape of a + false positive: a device registered only there looks unregistered everywhere. The + verdict is only sound when every collected server's client list was readable. + """ + return self.registration_analysed and not self.servers_with_unreadable_clients + @property def counts(self) -> dict[str, int]: return { @@ -181,6 +196,7 @@ def counts(self) -> dict[str, int]: "unknown_servers": len(self.unknown_servers), "reused_secrets": len(self.reused_secrets), "servers_examined": self.servers_examined, + "servers_with_unreadable_clients": len(self.servers_with_unreadable_clients), "secrets_not_exposable": self.secrets_not_exposable, } @@ -210,7 +226,18 @@ async def correlate(self, *, org_id: int = 1) -> AaaCorrelationReport: raw = (snapshot.ncm or {}).get("aaa_server") or {} config = AaaServerConfig.model_validate(raw) - if config.product not in _SERVER_PRODUCTS or not config.clients: + if config.product not in _SERVER_PRODUCTS: + # Not an AAA server at all — there is nothing to read here. + continue + if not config.clients: + # A known AAA server product whose client list came back empty or + # unparsed. Its clients are unknown, not absent: dropping it silently and + # then judging registration on the other servers alone flags every device + # that authenticates only against this one as unregistered. Recorded so the + # verdict can be withheld instead. + report.servers_with_unreadable_clients.append( + device.hostname or str(device.mgmt_ip) + ) continue record = ServerRecord( @@ -302,7 +329,11 @@ async def correlate(self, *, org_id: int = 1) -> AaaCorrelationReport: if device.id not in existing.used_by_ids: existing.used_by_ids.append(device.id) - if not report.registration_analysed: + if not report.registration_reliable: + # No server examined, or at least one server's client list was unreadable. + # Either way an unregistered verdict would be a guess; the limitation says + # so. The unknown-server detection above still runs — that reads the + # device side and does not depend on the servers' client lists. continue if device.device_class == DeviceClass.MANAGER.value: # A manager authenticates its administrators, not itself, so it is not @@ -353,6 +384,15 @@ def _limitations(self, report: AaaCorrelationReport) -> list[str]: "devices are registered as clients. This is not a finding that every " "device is unregistered — it is the absence of the data needed to ask." ) + elif report.servers_with_unreadable_clients: + servers = ", ".join(sorted(report.servers_with_unreadable_clients)) + notes.append( + f"{len(report.servers_with_unreadable_clients)} collected AAA server(s) " + f"({servers}) returned no readable client list, so a device could be " + "registered on one of them without appearing here. Unregistered-device " + "findings are withheld rather than risk flagging a device that is in fact a " + "client of one of these servers. Re-collect those servers to complete it." + ) if report.secrets_not_exposable: notes.append( @@ -473,6 +513,7 @@ def summarise(report: AaaCorrelationReport) -> dict[str, Any]: "devices_with_central_auth": report.coverage.devices_with_central_auth, "devices_not_evaluated": report.coverage.devices_not_evaluated, "registration_analysed": report.registration_analysed, + "registration_reliable": report.registration_reliable, "limitations": report.limitations, } diff --git a/backend/netsecops/services/auth.py b/backend/netsecops/services/auth.py index 8a9a13b..c6bbdfd 100644 --- a/backend/netsecops/services/auth.py +++ b/backend/netsecops/services/auth.py @@ -38,6 +38,7 @@ generate_recovery_codes, hash_password, hash_token, + matched_totp_step, mfa_provisioning_uri, mfa_qr_svg, needs_rehash, @@ -264,7 +265,8 @@ async def complete_mfa( secret = self.vault.open(secret_row.encrypted_secret, aad=str(user.id)).decode("utf-8") - if not verify_totp(secret, code, self.settings): + matched_step = matched_totp_step(secret, code, self.settings) + if matched_step is None: if not await self._consume_recovery_code(secret_row, user, code): await self._register_failure(user, ip_address, user_agent, reason="bad_totp") await self.audit.record( @@ -277,11 +279,16 @@ async def complete_mfa( ) raise AuthenticationError("Invalid verification code.") else: - # Reject replay of a code that is still inside its validity window. - step = int(datetime.now(UTC).timestamp()) // self.settings.mfa_totp_period_seconds - if secret_row.last_used_step is not None and step <= secret_row.last_used_step: + # Reject replay by the step the code actually belongs to, not the current one. + # A code for step S is accepted through step S + window, so recording the + # current step would let it be replayed for the rest of its own window: a + # single intercepted code would yield a second session one step later. + if ( + secret_row.last_used_step is not None + and matched_step <= secret_row.last_used_step + ): raise AuthenticationError("This verification code has already been used.") - secret_row.last_used_step = step + secret_row.last_used_step = matched_step return await self._complete_login(user, ip_address=ip_address, user_agent=user_agent) diff --git a/backend/netsecops/services/reporting.py b/backend/netsecops/services/reporting.py index b36c014..748f52d 100644 --- a/backend/netsecops/services/reporting.py +++ b/backend/netsecops/services/reporting.py @@ -245,6 +245,9 @@ async def generate( # Stamped into the content, not only the row: a report exported to a file # and mailed onward keeps its own provenance. "scope": self._describe_scope(scope_device_id, scope_group_id), + # The effective population, so a later trend report can refuse to subtract two + # reports that were narrowed to different estates (see `_trend`). + "scope_fingerprint": self._scope_fingerprint(actor.scope), } report.content = content @@ -261,6 +264,23 @@ async def generate( ) return report + @staticmethod + def _scope_fingerprint(scope: Scope) -> str: + """A stable identifier for the population an RBAC scope resolves to. + + `meta.scope` records only the device/group *parameters*, which are identical for + an executive summary whether the caller is unrestricted or group-scoped — yet + `_visible` silently narrows the summary to the caller's groups, so the two cover + different populations. A trend report reads an earlier report's frozen numbers and + subtracts them from the current ones; comparing across two different populations + fabricates deltas. This fingerprint is stamped at generation so the trend can + refuse a comparison whose two sides do not describe the same estate. + """ + if scope.unrestricted: + return "unrestricted" + ids = ",".join(sorted(str(gid) for gid in scope.device_group_ids)) + return f"groups:{ids}" + @staticmethod def _describe_scope(device_id: uuid.UUID | None, group_id: uuid.UUID | None) -> str: if device_id: @@ -772,6 +792,22 @@ async def _trend(self, scope: Scope, compare_to_id: uuid.UUID | None) -> dict[st "compare against." ) + # The two sides must describe the same estate. `get()` enforces only org_id, so a + # group-scoped caller can fetch a report generated over the whole estate (or a + # different group), and subtracting its frozen counts from this scoped summary + # would invent a trend — "960 findings closed" — out of a population mismatch. + # Refuse rather than fabricate. A report generated before the fingerprint existed + # cannot be shown comparable, so it is refused too. + current_fingerprint = self._scope_fingerprint(scope) + earlier_fingerprint = ((earlier.content or {}).get("meta") or {}).get("scope_fingerprint") + if earlier_fingerprint != current_fingerprint: + raise ValidationProblem( + f"Report {compare_to_id} was generated over a different scope " + f"({earlier_fingerprint or 'unknown'}) than this one ({current_fingerprint}), " + "so their totals describe different populations and cannot be compared. " + "Generate the earlier report under the same scope, or compare within it." + ) + current = await self._executive_summary(scope) previous = earlier.content or {} diff --git a/backend/netsecops/services/vuln_view.py b/backend/netsecops/services/vuln_view.py index 9e872f2..402f669 100644 --- a/backend/netsecops/services/vuln_view.py +++ b/backend/netsecops/services/vuln_view.py @@ -29,7 +29,7 @@ from netsecops.core.rbac import Scope from netsecops.db.models.collection import Finding, FindingKind, FindingStatus -from netsecops.db.models.inventory import Device +from netsecops.db.models.inventory import Device, DeviceStatus from netsecops.db.models.vulnerability import VulnAdvisory, VulnCve, VulnMatch from netsecops.schemas.vulnerability import ( AffectedDeviceRead, @@ -71,6 +71,33 @@ def _cvss(payload: dict[str, Any] | None) -> CvssRead | None: ) +def _score(cve: VulnCve) -> float: + """The best available base score for ranking, across every scored CVSS version. + + A CVE scored only under CVSS 4.0 — increasingly common for post-2024 CVEs, and both + the NVD and CSAF importers ingest it — has a NULL cvss31. Ranking on v3.1 alone + reads that as 0.0, sorting a genuinely critical finding to the bottom of the page. + """ + return max( + (_cvss(cve.cvss31) or CvssRead()).base_score or 0.0, + (_cvss(cve.cvss40) or CvssRead()).base_score or 0.0, + ) + + +def _headline_cvss(cve: VulnCve | None) -> CvssRead | None: + """The score to display for a row, preferring whichever version scores highest. + + Falls back to whichever version is present, so a v4-only CVE still shows its score + rather than a blank cell for a real high/critical. + """ + if cve is None: + return None + candidates = [c for c in (_cvss(cve.cvss31), _cvss(cve.cvss40)) if c is not None] + if not candidates: + return None + return max(candidates, key=lambda c: c.base_score or 0.0) + + class VulnViewService: """Read-only assembly of the vulnerability views.""" @@ -234,13 +261,10 @@ async def _enrich(self, rows: Sequence[tuple[Finding, Device]]) -> list[Vulnerab advisory = advisories.get((evidence.get("source"), evidence.get("advisory_id"))) # The worst-scored CVE on the advisory is the one that decides how the row - # reads, since a single advisory routinely carries several. + # reads, since a single advisory routinely carries several. Ranked across + # every CVSS version so a v4-only CVE is not sorted to the bottom at 0.0. scored = [cves[c] for c in finding_cves if c in cves] - headline = max( - scored, - key=lambda c: (_cvss(c.cvss31) or CvssRead()).base_score or 0.0, - default=None, - ) + headline = max(scored, key=_score, default=None) results.append( VulnerabilityRead( @@ -254,7 +278,7 @@ async def _enrich(self, rows: Sequence[tuple[Finding, Device]]) -> list[Vulnerab advisory_source=evidence.get("source"), cve_ids=finding_cves, cwe_ids=list(advisory.cwe_ids) if advisory else [], - cvss=_cvss(headline.cvss31 if headline else None), + cvss=_headline_cvss(headline), epss=headline.epss if headline else None, kev=_kev_flag([cve.kev for cve in scored]), kev_due_date=next( @@ -440,15 +464,38 @@ async def summary(self, *, scope: Scope) -> VulnerabilitySummary: if row.confidence: by_confidence[row.confidence] = by_confidence.get(row.confidence, 0) + 1 + # Numerator and denominator must describe the same population, or the + # "unassessed" count is fiction. `total` and `by_severity` above are narrowed to + # this org and this principal's visible groups; the coverage count has to match. + # `assess_all` also skips archived devices, so a device count that includes them + # would report devices as unassessed that the assessor deliberately never touches. + from netsecops.services.inventory import InventoryService + + inventory = InventoryService(self.session) + + assessed_stmt = ( + select(VulnMatch.device_id) + .join(Device, Device.id == VulnMatch.device_id) + .where( + VulnMatch.org_id == self.org_id, + Device.status != DeviceStatus.ARCHIVED.value, + ) + .distinct() + ) + assessed_stmt = await inventory.scoped(assessed_stmt, scope) assessed = { - device_id - for (device_id,) in ( - await self.session.execute( - select(VulnMatch.device_id).where(VulnMatch.org_id == self.org_id).distinct() - ) - ).all() + device_id for (device_id,) in (await self.session.execute(assessed_stmt)).all() } - device_stmt = select(func.count()).select_from(Device) + + device_stmt = ( + select(func.count()) + .select_from(Device) + .where( + Device.org_id == self.org_id, + Device.status != DeviceStatus.ARCHIVED.value, + ) + ) + device_stmt = await inventory.scoped(device_stmt, scope) total_devices = int((await self.session.execute(device_stmt)).scalar_one()) return VulnerabilitySummary( diff --git a/backend/netsecops/topology/path.py b/backend/netsecops/topology/path.py index d34f903..738dc3a 100644 --- a/backend/netsecops/topology/path.py +++ b/backend/netsecops/topology/path.py @@ -771,6 +771,31 @@ def walk( if translation.port is not None: port = translation.port + # A range that one of this device's prefixes cuts across does not take one path, + # so a single trace cannot describe it — not even when a connected route covers + # the representative address and the walk would otherwise declare arrival below. + # Checked before the arrival and routing branches, and per hop, because a range can + # be whole on the first device and subdivided several hops later. This is the guard + # the old post-route-lookup placement skipped whenever `serves()` fired first, + # which reported a range that is part directly-connected and part routed-away as + # fully arrived. + if dst_endpoint.is_range: + subdividing = current.routes_subdividing( + int(dst_endpoint.addresses.intervals[0][0]), + int(dst_endpoint.addresses.intervals[-1][1]), + ) + if subdividing: + result.hops.append(hop) + result.routing = RoutingConfidence.UNKNOWN + result.stopped_at_device = current.hostname + result.notes.append( + f"{current.hostname} routes {destination} through more than one " + f"prefix ({', '.join(subdividing)}), so different parts of that range " + "take different paths. Narrow the query to one of those prefixes to " + "get a single answer." + ) + return _finalise(result) + # Arrived: the destination is on a subnet this device is directly attached to. if current.serves(dst): # The interface as well as the zone: an outbound access list is bound by @@ -831,26 +856,6 @@ def walk( ) return _finalise(result) - # A range that one of this device's prefixes cuts across does not take one path, - # so a single trace cannot describe it. Checked per hop rather than once, because - # a range can be whole on the first device and subdivided three hops later. - if dst_endpoint.is_range: - subdividing = current.routes_subdividing( - int(dst_endpoint.addresses.intervals[0][0]), - int(dst_endpoint.addresses.intervals[-1][1]), - ) - if subdividing: - result.hops.append(hop) - result.routing = RoutingConfidence.UNKNOWN - result.stopped_at_device = current.hostname - result.notes.append( - f"{current.hostname} routes {destination} through more than one " - f"prefix ({', '.join(subdividing)}), so different parts of that range " - "take different paths. Narrow the query to one of those prefixes to " - "get a single answer." - ) - return _finalise(result) - # More than one route ties for best. A router chooses per flow by hashing the # header, and nothing in a configuration says which way this flow goes — so the # trace continues down one of them and the result has to say the others exist. diff --git a/backend/netsecops/vuln/matcher.py b/backend/netsecops/vuln/matcher.py index 22162b5..f68259f 100644 --- a/backend/netsecops/vuln/matcher.py +++ b/backend/netsecops/vuln/matcher.py @@ -460,11 +460,25 @@ def _nothing_applies( def _fixed_versions(advisory: Advisory) -> list[str]: - """The releases the vendor named as containing the fix (FR-VUL-10).""" + """The releases the vendor named as containing the fix (FR-VUL-10). + + Two feeds express the fix two ways. CSAF names the exact patched release in a + dedicated ``fixed`` list. NVD never populates that list at all — it states the fix + only as the exclusive upper bound of an affected range (``versionEndExcluding``, the + first release no longer affected), which lands on ``constraint.fixed``. Reading only + the CSAF form left the upgrade-path view blind to every NVD-sourced vulnerability. + + ``last_affected`` (``versionEndIncluding``) is deliberately not read here: it names a + release where no fix shipped, and treating it as fixed would tell an operator to + "upgrade" to a version that is still vulnerable. + """ versions: list[str] = [] for entry in advisory.fixed: if entry.constraint.kind is ConstraintKind.EXACT and entry.constraint.version: versions.append(entry.constraint.version) + for entry in advisory.affected: + if entry.constraint.kind is ConstraintKind.RANGE and entry.constraint.fixed: + versions.append(entry.constraint.fixed) return sorted(set(versions)) diff --git a/backend/tests/test_aaa_correlation.py b/backend/tests/test_aaa_correlation.py index f261daa..7f5e7e9 100644 --- a/backend/tests/test_aaa_correlation.py +++ b/backend/tests/test_aaa_correlation.py @@ -224,6 +224,43 @@ async def test_nothing_is_claimed_when_no_aaa_server_was_collected( assert report.unregistered_devices == [] assert any("cannot tell which devices are registered" in n for n in report.limitations) + async def test_a_server_with_an_unreadable_client_list_withholds_the_verdict( + self, session: AsyncSession, actor: Principal + ) -> None: + """A device registered only on a server whose client list failed to parse must not + be flagged unregistered on the strength of the other servers alone. + + Server B (ISE) is collected but its client list comes back empty/unparsed. A device + that authenticates only against it appears on no readable list, so an analysis that + silently drops B and judges on server A alone reports that device — a real client — + as unregistered/HIGH. The verdict must be withheld and the gap stated instead. + """ + radius = await add_device( + session, actor, ip="10.100.0.60", hostname="radius-01", platform="cisco_ios" + ) + ise = await add_device( + session, actor, ip="10.100.0.61", hostname="ise-07", platform="cisco_ise" + ) + only_on_b = await add_device( + session, actor, ip="198.51.100.42", hostname="only-on-ise" + ) + + await snapshot( + session, radius, server_ncm("freeradius", [client("some-sw", "198.51.100.99")]) + ) + # Collected, but the client list is empty/unparsed — unknown, not absent. + await snapshot(session, ise, server_ncm("ise", [])) + await snapshot(session, only_on_b, client_ncm([])) + + report = await AaaCorrelationService(session).correlate() + + assert report.servers_with_unreadable_clients == ["ise-07"] + assert report.registration_reliable is False + assert report.registration_analysed is True, "server A was still examined" + assert "only-on-ise" not in [d.hostname for d in report.unregistered_devices] + assert report.unregistered_devices == [] + assert any("no readable client list" in n for n in report.limitations) + async def test_the_aaa_server_itself_is_not_reported_as_unregistered( self, session: AsyncSession, actor: Principal ) -> None: diff --git a/backend/tests/test_arista_eos_parser.py b/backend/tests/test_arista_eos_parser.py index 103cfc6..b7b32d9 100644 --- a/backend/tests/test_arista_eos_parser.py +++ b/backend/tests/test_arista_eos_parser.py @@ -204,6 +204,36 @@ def test_a_null_route_has_no_next_hop(self, ncm: NormalisedConfig) -> None: assert discard.next_hop is None assert discard.interface == "Null0" + def test_a_route_with_both_interface_and_gateway_keeps_the_gateway(self) -> None: + # `ip route ` — EOS accepts both an egress interface and a + # next-hop gateway. Capturing only the first token drops the gateway and the path + # walk then treats the route as directly-attached instead of forwarding to the + # real next hop, truncating or misrouting the walk. + ncm = parse( + "hostname sw1\n" + "!\n" + "ip route 0.0.0.0/0 Ethernet1 10.1.1.1\n" + "ip route 10.20.0.0/24 Vlan10 192.168.1.254 tag 5\n" + ) + by_destination = {r.destination: r for r in ncm.routing.routes} + + default = by_destination["0.0.0.0/0"] + assert default.interface == "Ethernet1" + assert default.next_hop == "10.1.1.1" + + tagged = by_destination["10.20.0.0/24"] + assert tagged.interface == "Vlan10" + assert tagged.next_hop == "192.168.1.254" + + def test_a_gateway_only_route_still_has_no_interface(self) -> None: + # The added optional gateway group must not steal a trailing administrative + # distance or `name`/`tag` keyword and turn a plain next-hop route into an + # interface route. + ncm = parse("hostname sw1\n!\nip route 10.40.0.0/16 10.10.10.9 200\n") + route = next(r for r in ncm.routing.routes if r.destination == "10.40.0.0/16") + assert route.next_hop == "10.10.10.9" + assert route.interface is None + class TestVersionAndRobustness: def test_version_model_and_serial_from_one_command(self) -> None: diff --git a/backend/tests/test_cloud_and_sdn_parsers.py b/backend/tests/test_cloud_and_sdn_parsers.py index d7e0c19..aecbdda 100644 --- a/backend/tests/test_cloud_and_sdn_parsers.py +++ b/backend/tests/test_cloud_and_sdn_parsers.py @@ -238,6 +238,52 @@ 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"] + def test_a_group_is_catalogued_by_id_so_rules_referencing_it_resolve(self) -> None: + # A rule refers to a group by its path id (grp-4821). An API/Terraform group whose + # display_name differs ("Web Servers") must still be catalogued under the id the + # rule uses, or every rule referencing it loses its members and drops out of the + # overlap and shadowing analysis. + body = { + "/policy/api/v1/infra/domains/default/groups": { + "results": [ + { + "id": "grp-4821", + "display_name": "Web Servers", + "expression": [ + { + "resource_type": "IPAddressExpression", + "ip_addresses": ["10.0.0.5"], + } + ], + } + ] + }, + "/policy/api/v1/infra/domains/default/security-policies": { + "results": [ + { + "id": "app", + "display_name": "app", + "rules": [ + { + "id": "r1", + "display_name": "web-in", + "action": "ALLOW", + "source_groups": ["/infra/domains/default/groups/grp-4821"], + "destination_groups": ["ANY"], + } + ], + } + ] + }, + } + ncm = parse_text("vmware_nsx", json.dumps(body)) + catalogue = {g.name for g in ncm.firewall.address_groups} + + assert "grp-4821" in catalogue, "the group must be keyed by the id a rule uses" + assert "Web Servers" not in catalogue + rule = next(r for r in ncm.firewall.security_rules if r.name == "web-in") + assert rule.src == ["grp-4821"], "the rule member must match the catalogue key" + def test_the_manager_version(self, nsx: NormalisedConfig) -> None: assert nsx.device.version == "4.1.2.3.0" @@ -361,6 +407,32 @@ def test_the_group_itself_is_an_object(self, aws: NormalisedConfig) -> None: # So a rule naming another group as its source resolves to something. assert any(g.name == "sg-0aaa1111" for g in aws.firewall.address_groups) + def test_a_cross_account_group_reference_is_externally_resolved(self) -> None: + # A rule may permit from a group in a peered VPC or another account whose export + # we do not hold. Left untyped it lands in the MISSING bucket and advises a + # re-collection that can never produce it; its membership is externally resolved, + # exactly like a prefix list. + body = [ + { + "GroupId": "sg-0aaa1111", + "GroupName": "web", + "VpcId": "vpc-1", + "IpPermissions": [ + { + "IpProtocol": "tcp", + "FromPort": 443, + "ToPort": 443, + "UserIdGroupPairs": [{"GroupId": "sg-99999999"}], + } + ], + "IpPermissionsEgress": [], + } + ] + ncm = parse_text("aws_vpc", json.dumps(body)) + peer = next((g for g in ncm.firewall.address_groups if g.name == "sg-99999999"), None) + assert peer is not None, "a cross-account group reference must become an object" + assert peer.type == "security-group" + def test_vpcs_become_zones(self, aws: NormalisedConfig) -> None: assert set(aws.firewall.zones) == {"vpc-01234567", "vpc-89abcdef"} @@ -429,3 +501,34 @@ def test_the_priority_is_kept(self, azure: NormalisedConfig) -> None: def test_an_arm_wrapped_export_is_accepted(self) -> None: body = {"value": [{"name": "nsg-1", "properties": {"securityRules": []}}]} assert parse_text("azure_nsg", json.dumps(body)).parse_failed is False + + def test_a_regional_service_tag_is_typed_not_read_as_a_literal(self) -> None: + # Regional tags (Storage.EastUS, Sql.WestEurope) carry a dot, but a dot is not an + # address. Skipped as a "literal" the tag resolves to nothing and an inbound allow + # reads as matching no traffic — the opposite of the truth. A genuine CIDR in the + # same rule must still be left alone. + body = [ + { + "name": "nsg-1", + "properties": { + "securityRules": [ + { + "name": "allow-storage", + "properties": { + "priority": 100, + "direction": "Inbound", + "access": "Allow", + "protocol": "Tcp", + "sourceAddressPrefix": "Storage.EastUS", + "destinationAddressPrefix": "10.0.0.0/24", + "destinationPortRange": "443", + }, + } + ] + }, + } + ] + ncm = parse_text("azure_nsg", json.dumps(body)) + tags = {o.name for o in ncm.firewall.address_objects if o.type == "service-tag"} + assert "Storage.EastUS" in tags + assert "10.0.0.0/24" not in tags, "a real CIDR must not be typed a service tag" diff --git a/backend/tests/test_config.py b/backend/tests/test_config.py index 969c5bf..03dc556 100644 --- a/backend/tests/test_config.py +++ b/backend/tests/test_config.py @@ -165,6 +165,38 @@ def test_wildcard_cors_is_rejected(self, monkeypatch: pytest.MonkeyPatch) -> Non with pytest.raises(ValidationError, match="wildcard CORS"): build(monkeypatch, NETSECOPS_ENV="prod", NETSECOPS_CORS_ORIGINS="*") + def test_unset_secret_key_is_rejected(self, monkeypatch: pytest.MonkeyPatch) -> None: + """An unset key would fall to the random per-process default_factory and run + silently — tokens signed by one worker failing to verify on another, and every + restart invalidating every session. It must fail at boot instead.""" + for key in ( + "SECRET_KEY", + "NETSECOPS_ENV", + "NETSECOPS_DEBUG", + "NETSECOPS_COOKIE_SECURE", + "NETSECOPS_CORS_ORIGINS", + ): + monkeypatch.delenv(key, raising=False) + monkeypatch.setenv("DATABASE_URL", BASE_ENV["DATABASE_URL"]) + monkeypatch.setenv("NETSECOPS_ENV", "prod") + monkeypatch.setenv("NETSECOPS_COOKIE_SECURE", "true") + monkeypatch.setenv("NETSECOPS_CORS_ORIGINS", "https://netsecops.example") + + with pytest.raises(ValidationError, match="SECRET_KEY must be set"): + Settings(_env_file=None) # type: ignore[call-arg] + + def test_an_explicit_secret_key_satisfies_the_production_guard( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + # build() sets SECRET_KEY from the environment, so the key is explicitly provided. + settings = build( + monkeypatch, + NETSECOPS_ENV="prod", + NETSECOPS_COOKIE_SECURE="true", + NETSECOPS_CORS_ORIGINS="https://netsecops.example", + ) + assert settings.is_production + def test_valid_production_config_is_accepted(self, monkeypatch: pytest.MonkeyPatch) -> None: settings = build( monkeypatch, diff --git a/backend/tests/test_discovery_executor.py b/backend/tests/test_discovery_executor.py index 43b463b..7fb525f 100644 --- a/backend/tests/test_discovery_executor.py +++ b/backend/tests/test_discovery_executor.py @@ -257,6 +257,30 @@ async def test_a_scope_not_asking_for_snmp_gets_no_such_note(self, session, acto assert SNMP_UNAVAILABLE_NOTE not in summary.notes + async def test_a_caveat_on_a_non_responding_host_still_reaches_the_run( + self, session, actor + ) -> None: + """ICMP going away mid-run rides on the host being probed when it happens, and that + host usually does not respond. Dropped with the silent result, the very degradation + that makes a run partial vanishes and 'found N hosts' reads as complete coverage. + """ + note = "ICMP became unavailable during the run." + + def degraded(address: str) -> HostResult: + return HostResult(address=address, responded=False, probes_sent=3, notes=(note,)) + + scope = await make_scope(session, targets=["198.51.100.1"]) + summary = await DiscoveryExecutor(session, probe_host=Recorder(degraded)).run( + scope, actor=actor + ) + + assert note in summary.notes, "a caveat on a silent host must not be dropped" + + run = ( + await session.execute(select(DiscoveryRun).where(DiscoveryRun.id == summary.run_id)) + ).scalar_one() + assert note in run.notes + # ════════════════════════════ stopping ═══════════════════════════════════════ diff --git a/backend/tests/test_discovery_transport.py b/backend/tests/test_discovery_transport.py index 78f303f..6f3cb3b 100644 --- a/backend/tests/test_discovery_transport.py +++ b/backend/tests/test_discovery_transport.py @@ -228,6 +228,26 @@ async def test_a_closed_port_is_a_non_response_rather_than_an_error(self) -> Non assert outcome.responded is False assert outcome.detail + async def test_a_refused_port_proves_the_host_is_alive(self, monkeypatch) -> None: + """A RST is not silence. The port is closed, but the host answered the network. + + Recorded identically to a timeout, a hardened host that drops ICMP and closes every + scanned port would be reported as absent — a device that definitively answered + filed with dead address space. A refusal is injected rather than provoked, because + whether a just-closed loopback port RSTs or times out is an OS/timing detail. + """ + + async def refuse(*_args, **_kwargs): + raise ConnectionRefusedError("connection refused") + + monkeypatch.setattr(asyncio, "open_connection", refuse) + + probe = authorise(ProbeKind.TCP_CONNECT, LOOPBACK, port=9, allowed_ports=(9,)) + outcome = await send_tcp_connect(probe, timeout=QUICK) + + assert outcome.responded is False, "a closed port is not an open one" + assert outcome.host_alive is True, "but a refusal proves the host is up" + async def test_connecting_sends_no_payload(self) -> None: """FR-DISC-02 permits a *connect*. Anything written would be interaction.""" async with RecordingServer() as server: @@ -378,6 +398,23 @@ async def test_an_open_port_makes_a_host_responsive(self) -> None: assert result.responded is True assert result.open_ports == (server.port,) + async def test_a_refused_port_makes_a_host_responsive_without_opening_it( + self, monkeypatch + ) -> None: + """A host that refuses every scanned port is still found — it answered — but the + refused port is not recorded as open and gets no follow-up read.""" + + async def refuse(*_args, **_kwargs): + raise ConnectionRefusedError("connection refused") + + monkeypatch.setattr(asyncio, "open_connection", refuse) + + result = await self._prober(9, ProbeKind.SSH_BANNER).probe(LOOPBACK) + + assert result.responded is True, "a refusal proves the host is on the network" + assert result.open_ports == (), "but a refused port is not an open one" + assert result.evidence == [], "and a closed port yields no banner" + async def test_a_banner_port_yields_weighted_ssh_evidence(self) -> None: async with RecordingServer(greeting=b"SSH-2.0-FortiSSH_1.0\r\n") as server: result = await self._prober(server.port, ProbeKind.SSH_BANNER).probe(LOOPBACK) diff --git a/backend/tests/test_mfa_recovery.py b/backend/tests/test_mfa_recovery.py index 301372e..b816b1a 100644 --- a/backend/tests/test_mfa_recovery.py +++ b/backend/tests/test_mfa_recovery.py @@ -95,6 +95,28 @@ async def test_reset_is_idempotent(self, session: AsyncSession, vault) -> None: assert user.mfa_enabled is False +class TestTotpReplay: + async def test_a_totp_code_cannot_be_used_twice(self, session: AsyncSession, vault) -> None: + """A single intercepted code must not yield a second session while still valid. + + The guard records the step the code belongs to, so the same six digits presented + again — a fresh challenge relayed a moment later — are refused. Recording the + current step instead would accept the replay one step on, inside the code's window. + """ + auth = AuthService(session, vault=vault) + user = await make_user(session, username="totp_replay") + secret = await enrol(auth, user) + code = pyotp.TOTP(secret).now() + + first = await auth.authenticate(user.username, TEST_PASSWORD) + signed_in = await auth.complete_mfa(first.mfa_token, code) # type: ignore[union-attr] + assert hasattr(signed_in, "access_token") + + second = await auth.authenticate(user.username, TEST_PASSWORD) + with pytest.raises(AuthenticationError, match="already been used"): + await auth.complete_mfa(second.mfa_token, code) # type: ignore[union-attr] + + class TestReEnrolmentAfterReset: async def test_can_enrol_again_with_a_fresh_secret(self, session: AsyncSession, vault) -> None: auth = AuthService(session, vault=vault) diff --git a/backend/tests/test_reports.py b/backend/tests/test_reports.py index 4780a12..797b135 100644 --- a/backend/tests/test_reports.py +++ b/backend/tests/test_reports.py @@ -346,6 +346,45 @@ async def test_a_trend_without_a_comparison_is_refused( assert report.status == ReportStatus.FAILED.value assert "compare_to_id" in (report.error_message or "") + async def test_a_trend_across_mismatched_scopes_is_refused( + self, session: AsyncSession, principal: Principal, estate + ) -> None: + """The earlier report covered the whole estate; a group-scoped caller comparing + against it would subtract two different populations and invent a delta.""" + import uuid as _uuid + + service = ReportingService(session) + whole_estate = await service.generate(ReportTemplate.EXECUTIVE_SUMMARY, actor=principal) + await session.commit() + + scoped = Principal( + id=principal.id, + username=principal.username, + roles=principal.roles, + scope=Scope(device_group_ids=frozenset({_uuid.uuid4()})), + ) + report = await ReportingService(session).generate( + ReportTemplate.TREND, actor=scoped, compare_to_id=whole_estate.id + ) + + assert report.status == ReportStatus.FAILED.value + assert "different scope" in (report.error_message or "") + + async def test_a_trend_within_the_same_scope_still_compares( + self, session: AsyncSession, principal: Principal, estate + ) -> None: + """The guard must not block the ordinary case: two reports over the same estate.""" + service = ReportingService(session) + first = await service.generate(ReportTemplate.EXECUTIVE_SUMMARY, actor=principal) + await session.commit() + + trend = await ReportingService(session).generate( + ReportTemplate.TREND, actor=principal, compare_to_id=first.id + ) + + assert trend.status == ReportStatus.READY.value + assert "findings_delta" in trend.content + class TestStatesAreNotConfused: async def test_a_failed_report_says_why( diff --git a/backend/tests/test_security.py b/backend/tests/test_security.py index 17974f9..f7ecef3 100644 --- a/backend/tests/test_security.py +++ b/backend/tests/test_security.py @@ -24,6 +24,7 @@ generate_recovery_codes, hash_password, hash_token, + matched_totp_step, mfa_provisioning_uri, validate_password_policy, verify_password, @@ -172,6 +173,38 @@ def test_code_outside_the_window_rejected(self, settings: Settings) -> None: stale = pyotp.TOTP(secret).at(int(time.time()) - 300) assert not verify_totp(secret, stale, settings) + def test_matched_step_names_the_codes_own_step_not_the_current_one( + self, settings: Settings + ) -> None: + """The property the replay guard depends on. + + A code minted for the previous step is still valid now (window=1). Its matched + step must be the step it belongs to — not the current step — so a guard that + stores it can reject the code being presented again later in its window. Recording + the current step instead is exactly the replay hole this fixes. + """ + secret = generate_mfa_secret() + period = settings.mfa_totp_period_seconds + prev_time = int(time.time()) - period + prev_step = prev_time // period + code = pyotp.TOTP(secret, interval=period).at(prev_time) + + assert matched_totp_step(secret, code, settings) == prev_step + + def test_matched_step_is_none_for_an_invalid_code(self, settings: Settings) -> None: + assert matched_totp_step(generate_mfa_secret(), "000000", settings) is None + + def test_matched_step_advances_with_a_fresh_code(self, settings: Settings) -> None: + """A guard comparing `matched <= last_used` must still admit the next code.""" + secret = generate_mfa_secret() + period = settings.mfa_totp_period_seconds + now = int(time.time()) + prev = matched_totp_step(secret, pyotp.TOTP(secret, interval=period).at(now - period)) + curr = matched_totp_step(secret, pyotp.TOTP(secret, interval=period).at(now)) + + assert prev is not None and curr is not None + assert curr > prev + def test_provisioning_uri_carries_issuer(self, settings: Settings) -> None: uri = mfa_provisioning_uri(generate_mfa_secret(), "user@example.test", settings) assert uri.startswith("otpauth://totp/") diff --git a/backend/tests/test_topology.py b/backend/tests/test_topology.py index 7e3a3d8..ee8773d 100644 --- a/backend/tests/test_topology.py +++ b/backend/tests/test_topology.py @@ -800,6 +800,31 @@ def test_a_range_split_across_two_routes_is_not_traced_as_one(self) -> None: assert "more than one prefix" in notes assert "10.20.0.0/25" in notes + def test_a_range_part_connected_and_part_routed_away_is_not_arrived(self) -> None: + """The low half of the range is directly attached; the high half is routed onward. + + The representative address (the range's low host) sits on the connected prefix, so + `serves()` would short-circuit and declare the whole range arrived — reporting a + range that only half-arrives as fully routed. The subdividing guard must fire first. + """ + edge = node( + "edge-rtr", + addresses={"lan": "10.10.0.1/24", "low": "10.20.0.1/25", "up": "10.0.1.1/30"}, + routes=[ + connected("10.10.0.0/24", "lan"), + connected("10.20.0.0/25", "low"), # low half directly attached + static("10.20.0.128/25", "10.0.1.2", "up"), # high half routed away + ], + ) + result = walk( + build_graph([edge]), source="10.10.0.0/24", destination="10.20.0.0/24", port=443 + ) + + assert result.routing is RoutingConfidence.UNKNOWN + notes = " ".join(result.notes) + assert "more than one prefix" in notes + assert "10.20.0.0/25" in notes + def test_an_ipv6_range_is_refused_rather_than_guessed(self) -> None: """Forwarding tables are parsed for IPv4 only, so v6 has nothing to walk.""" with pytest.raises(ValidationProblem, match="IPv6"): diff --git a/backend/tests/test_vuln_matcher.py b/backend/tests/test_vuln_matcher.py index 220c524..d25f63b 100644 --- a/backend/tests/test_vuln_matcher.py +++ b/backend/tests/test_vuln_matcher.py @@ -696,6 +696,34 @@ def test_the_fixed_release_is_carried_for_the_upgrade_view(self) -> None: assert result.fixed_versions == ["15.2(7)E6"] + def test_the_range_fix_bound_feeds_the_upgrade_view_for_nvd(self) -> None: + """NVD never populates the dedicated `fixed` list. + + It states the fix only as the exclusive upper bound of the affected range + (`versionEndExcluding`, on `constraint.fixed`). Reading the CSAF `fixed` list + alone left the upgrade-path view blind to every NVD-sourced vulnerability, even + though the release to upgrade to is right there on the range. + """ + result = match( + device(), + advisory(affected("<15.2(7)E6", fixed="15.2(7)E6")), # no dedicated `fixed` list + ) + + assert result.fixed_versions == ["15.2(7)E6"] + + def test_an_inclusive_last_affected_bound_is_never_offered_as_a_fix(self) -> None: + """`versionEndIncluding` names a release that is *itself* still vulnerable. + + Offering it as an upgrade target would tell an operator to move to a version that + does not contain the fix — worse than offering nothing. + """ + result = match( + device(version="15.2(7)E6"), + advisory(affected("<=15.2(7)E6", last_affected="15.2(7)E6", train="E")), + ) + + assert result.fixed_versions == [] + @pytest.mark.parametrize( ("confidence", "expected"), [ diff --git a/backend/tests/test_vulnerabilities_api.py b/backend/tests/test_vulnerabilities_api.py index 3b93452..3d05753 100644 --- a/backend/tests/test_vulnerabilities_api.py +++ b/backend/tests/test_vulnerabilities_api.py @@ -29,7 +29,7 @@ from netsecops.core.rbac import Principal, Role, Scope from netsecops.db.models import Device, User from netsecops.db.models.collection import Finding, FindingKind, FindingStatus -from netsecops.db.models.inventory import DeviceClass, Vendor +from netsecops.db.models.inventory import DeviceClass, DeviceStatus, Vendor from netsecops.db.models.vulnerability import VulnAdvisory, VulnCve, VulnMatch from netsecops.services.inventory import InventoryService from tests.conftest import make_user @@ -276,6 +276,56 @@ async def test_the_matchers_reasoning_survives_to_the_api( assert likely["reasoning"] == ["version matched", "feature condition unknown"] + async def test_a_cvss_v4_only_cve_still_shows_its_score( + self, + client: AsyncClient, + session: AsyncSession, + principal, + analyst_user, + authenticate, + estate, + ) -> None: + """CVSS 4.0 is increasingly the only score a post-2024 CVE carries. + + Ranking and displaying the row on v3.1 alone blanks the score cell and sorts a + genuinely critical finding to the bottom of the page — it reads as unscored when + it is one of the worst. + """ + device = await add_device( + session, principal, ip="10.0.0.40", hostname="sw-v4only", version="15.2(7)E3" + ) + advisory = await add_advisory( + session, source="nvd", advisory_id="CVE-2025-40000", cve_ids=["CVE-2025-40000"] + ) + session.add( + VulnCve( + org_id=1, + cve_id="CVE-2025-40000", + description="scored only under CVSS 4.0", + cvss31=None, + cvss40={ + "version": "4.0", + "base_score": 9.3, + "base_severity": "CRITICAL", + "vector": "CVSS:4.0/AV:N/AC:L/AT:N/PR:N/UI:N/VC:H/VI:H/VA:H", + }, + kev=False, + published=datetime(2025, 1, 1, tzinfo=UTC), + ) + ) + await add_finding( + session, device, advisory, confidence="confirmed", severity="critical" + ) + await session.commit() + authenticate(analyst_user) + + rows = (await client.get(LIST, params={"device_id": str(device.id)})).json()["data"] + row = next(r for r in rows if "CVE-2025-40000" in r["cve_ids"]) + + assert row["cvss"] is not None, "a v4-only CVE must not present as unscored" + assert row["cvss"]["base_score"] == 9.3 + assert row["cvss"]["version"] == "4.0" + async def test_filtering_by_confidence(self, client: AsyncClient, estate) -> None: rows = (await client.get(LIST, params={"confidence": "confirmed"})).json()["data"] @@ -508,6 +558,35 @@ async def test_a_device_with_no_match_rows_counts_as_unassessed( assert body["devices_unassessed"] >= 1 + async def test_archived_devices_are_not_counted_as_unassessed( + self, + client: AsyncClient, + session: AsyncSession, + principal, + analyst_user, + authenticate, + estate, + ) -> None: + """`assess_all` deliberately skips archived devices, so the coverage denominator + must skip them too. + + Counting an archived device as "unassessed" reports a coverage gap the assessor + will never close — a number that can only ever go up, describing devices no longer + in service. + """ + before = (await client.get(SUMMARY)).json()["devices_unassessed"] + + archived = await add_device( + session, principal, ip="10.0.0.50", hostname="sw-retired", version="15.2(7)E3" + ) + archived.status = DeviceStatus.ARCHIVED.value + await session.commit() + authenticate(analyst_user) + + after = (await client.get(SUMMARY)).json()["devices_unassessed"] + + assert after == before, "an archived device must not inflate the unassessed count" + class TestFeeds: async def test_feed_history_is_empty_before_any_import(