diff --git a/limacharlie/commands/cloudsec.py b/limacharlie/commands/cloudsec.py index 28951951..47d786b5 100644 --- a/limacharlie/commands/cloudsec.py +++ b/limacharlie/commands/cloudsec.py @@ -1027,6 +1027,14 @@ def _finding_filter_options(f): help="Include findings with no owner (the untriaged bucket); " "combines with --owner.", )(f) + f = click.option( + "--sla", "sla_states", multiple=True, + help="Filter by remediation-SLA state; repeatable (OR): breached, " + "due_soon, on_track, exempt (the finding is not open, so the " + "clock does not report on it), none (no SLA clause covers it). " + "There is no built-in default SLA, so every finding reads " + "'none' until the org writes an 'sla' cloudsec_policy record.", + )(f) f = click.option( "--owner", "owners", multiple=True, help="Filter by assigned owner; repeatable (OR). Use --unassigned " @@ -1236,7 +1244,9 @@ def _sort_options(f): )(f) f = click.option( "--sort", default=None, - help="Sort key: lc_risk (default), severity, or first_seen.", + help="Sort key: lc_risk (default), severity, first_seen, or due_at. " + "due_at is the one key that defaults to ASCENDING (soonest due " + "first) and puts findings with no due date last, not out.", )(f) return f @@ -1428,8 +1438,8 @@ def finding_group() -> None: @_paging_options @pass_context def finding_list(ctx, severities, finding_classes, statuses, accounts, - owners, unassigned, reachable, kev, q, sort, order, - cursor, limit) -> None: + owners, unassigned, sla_states, reachable, kev, q, sort, + order, cursor, limit) -> None: """List the merged, risk-ranked cloud-security findings. \b @@ -1438,6 +1448,7 @@ def finding_list(ctx, severities, finding_classes, statuses, accounts, limacharlie cloudsec finding list --class public_exposure --kev limacharlie cloudsec finding list --owner alice@corp.com limacharlie cloudsec finding list --unassigned + limacharlie cloudsec finding list --sla breached --sort due_at """ cs = _get_cloudsec(ctx) _output(ctx, cs.list_findings( @@ -1446,6 +1457,7 @@ def finding_list(ctx, severities, finding_classes, statuses, accounts, status=list(statuses) or None, account=list(accounts) or None, owner=_selector_with_empty(owners, unassigned), + sla=list(sla_states) or None, reachable=reachable, kev=kev, q=q, @@ -1467,7 +1479,8 @@ def finding_list(ctx, severities, finding_classes, statuses, accounts, "pin can still be dropped.") @pass_context def finding_facets(ctx, severities, finding_classes, statuses, accounts, - owners, unassigned, reachable, kev, q, owner_pins) -> None: + owners, unassigned, sla_states, reachable, kev, q, + owner_pins) -> None: """Cross-filtered facet counts for the findings worklist. \b @@ -1483,6 +1496,7 @@ def finding_facets(ctx, severities, finding_classes, statuses, accounts, account=list(accounts) or None, owner=_selector_with_empty(owners, unassigned), owner_pin=list(owner_pins) or None, + sla=list(sla_states) or None, reachable=reachable, kev=kev, q=q, @@ -1500,7 +1514,7 @@ def finding_facets(ctx, severities, finding_classes, statuses, accounts, "rollup is not paginated; 'distinct' reports the tail.") @pass_context def finding_causes(ctx, severities, finding_classes, statuses, accounts, - owners, unassigned, reachable, kev, q, cause, + owners, unassigned, sla_states, reachable, kev, q, cause, limit) -> None: """Findings grouped by CAUSE: one edit that closes N findings. @@ -1517,6 +1531,7 @@ def finding_causes(ctx, severities, finding_classes, statuses, accounts, status=list(statuses) or None, account=list(accounts) or None, owner=_selector_with_empty(owners, unassigned), + sla=list(sla_states) or None, reachable=reachable, kev=kev, q=q, @@ -2485,8 +2500,8 @@ def export_group() -> None: @_export_output_option @pass_context def export_findings(ctx, severities, finding_classes, statuses, accounts, - owners, unassigned, reachable, kev, q, sort, order, - output_path) -> None: + owners, unassigned, sla_states, reachable, kev, q, sort, + order, output_path) -> None: """Export the (filtered) findings worklist as CSV. \b @@ -2502,6 +2517,7 @@ def export_findings(ctx, severities, finding_classes, statuses, accounts, status=list(statuses) or None, account=list(accounts) or None, owner=_selector_with_empty(owners, unassigned), + sla=list(sla_states) or None, reachable=reachable, kev=kev, q=q, diff --git a/limacharlie/sdk/cloudsec.py b/limacharlie/sdk/cloudsec.py index 35547577..8b8cb8aa 100644 --- a/limacharlie/sdk/cloudsec.py +++ b/limacharlie/sdk/cloudsec.py @@ -92,6 +92,7 @@ def _finding_query_pairs( account: list[str] | None = None, owner: list[str] | None = None, owner_pin: list[str] | None = None, + sla: list[str] | None = None, reachable: bool | None = None, kev: bool | None = None, q: str | None = None, @@ -109,7 +110,7 @@ def _finding_query_pairs( """ return _query_pairs( severity=severity, finding_class=finding_class, status=status, - account=account, owner=owner, owner_pin=owner_pin, + account=account, owner=owner, owner_pin=owner_pin, sla=sla, reachable=reachable, kev=kev, q=q, sort=sort, order=order, cursor=cursor, limit=limit, ) @@ -264,6 +265,7 @@ def list_findings( status: list[str] | None = None, account: list[str] | None = None, owner: list[str] | None = None, + sla: list[str] | None = None, reachable: bool | None = None, kev: bool | None = None, q: str | None = None, @@ -296,12 +298,25 @@ def list_findings( Pass ``None`` for no owner constraint. Owners are set with :meth:`set_finding_owner` and counted by the ``owner`` facet of :meth:`get_finding_facets`. + sla: Remediation-SLA state filter values, OR'd: + ``breached`` (past due), ``due_soon`` (inside the last + quarter of its own window, clamped to 1-7 days), + ``on_track``, ``exempt`` (the finding is not open, so the + clock does not report on it), ``none`` (no SLA clause + covers it). The state is DERIVED at read time from the + finding's ``due_at`` and status — there is no built-in + default SLA, so on an org that has not written an ``sla`` + policy every finding is ``none``. Counted by the ``sla`` + facet of :meth:`get_finding_facets`. reachable: Only findings on (non-)reachable resources. kev: Only findings with (without) a KEV vulnerability. q: Substring search. sort: Server-side sort key: ``lc_risk`` (the default), - ``severity``, or ``first_seen``. - order: ``desc`` (the default) or ``asc``. + ``severity``, ``first_seen``, or ``due_at``. ``due_at`` + is the one key that defaults to ASCENDING (soonest due + first) and it places findings with no due date LAST rather + than excluding them. + order: ``desc`` (the default, except for ``due_at``) or ``asc``. cursor: Keyset-pagination token from a previous page. limit: Page size (server clamps to 1000). @@ -310,7 +325,8 @@ def list_findings( """ return self._get("findings", _finding_query_pairs( severity=severity, finding_class=finding_class, status=status, - account=account, owner=owner, reachable=reachable, kev=kev, q=q, + account=account, owner=owner, sla=sla, + reachable=reachable, kev=kev, q=q, sort=sort, order=order, cursor=cursor, limit=limit, )) @@ -323,6 +339,7 @@ def get_finding_facets( account: list[str] | None = None, owner: list[str] | None = None, owner_pin: list[str] | None = None, + sla: list[str] | None = None, reachable: bool | None = None, kev: bool | None = None, q: str | None = None, @@ -350,13 +367,20 @@ def get_finding_facets( tail truncation. Render a pinned-but-absent owner as zero rather than assuming the map is complete. + sla: Remediation-SLA state filter values — see + :meth:`list_findings`. The ``sla`` facet always carries + EVERY state key, zeroes included, so a caller never has to + invent a missing count (a ``breached`` chip that vanishes + reads as "the feature is off", not "you are on top of it"). + Returns: ``{"facets": {..., "owner": {"": 12, "alice@corp.com": 3}, - "owner_truncated": false}}``. + "owner_truncated": false, "sla": {"breached": 4, "due_soon": 1, + "on_track": 20, "exempt": 0, "none": 900}}}``. """ return self._get("findings/facets", _finding_query_pairs( severity=severity, finding_class=finding_class, status=status, - account=account, owner=owner, owner_pin=owner_pin, + account=account, owner=owner, owner_pin=owner_pin, sla=sla, reachable=reachable, kev=kev, q=q, )) @@ -369,6 +393,7 @@ def list_finding_causes( status: list[str] | None = None, account: list[str] | None = None, owner: list[str] | None = None, + sla: list[str] | None = None, reachable: bool | None = None, kev: bool | None = None, q: str | None = None, @@ -425,7 +450,8 @@ def list_finding_causes( """ pairs = _finding_query_pairs( severity=severity, finding_class=finding_class, status=status, - account=account, owner=owner, reachable=reachable, kev=kev, q=q, + account=account, owner=owner, sla=sla, + reachable=reachable, kev=kev, q=q, limit=limit, ) _add_scalar(pairs, "cause", cause) @@ -1479,6 +1505,7 @@ def export_findings_csv( status: list[str] | None = None, account: list[str] | None = None, owner: list[str] | None = None, + sla: list[str] | None = None, reachable: bool | None = None, kev: bool | None = None, q: str | None = None, @@ -1487,6 +1514,9 @@ def export_findings_csv( ) -> str: """Export the (filtered) findings worklist as CSV text. + The exported rows carry ``due_at``, ``sla_state`` and + ``sla_source`` alongside the worklist fields. + Takes the same filter selectors as :meth:`list_findings`; the server walks the full filtered set (no pagination), capped at 100k rows. @@ -1496,7 +1526,8 @@ def export_findings_csv( """ pairs = _finding_query_pairs( severity=severity, finding_class=finding_class, status=status, - account=account, owner=owner, reachable=reachable, kev=kev, q=q, + account=account, owner=owner, sla=sla, + reachable=reachable, kev=kev, q=q, sort=sort, order=order, ) pairs.append(("format", "csv")) diff --git a/tests/unit/test_cli_cloudsec.py b/tests/unit/test_cli_cloudsec.py index 06ef25a7..71094778 100644 --- a/tests/unit/test_cli_cloudsec.py +++ b/tests/unit/test_cli_cloudsec.py @@ -269,6 +269,7 @@ def test_list_repeatable_filters(self): status=None, account=None, owner=None, + sla=None, reachable=True, kev=True, q="prod", @@ -827,8 +828,8 @@ def test_export_findings_stdout(self): assert result.output == "col_a,col_b\n1,2\n" inst.export_findings_csv.assert_called_once_with( severity=["CRITICAL"], finding_class=None, status=["open"], - account=None, owner=None, reachable=None, kev=None, q=None, - sort=None, order=None, + account=None, owner=None, sla=None, reachable=None, kev=None, + q=None, sort=None, order=None, ) def test_export_findings_to_file(self, tmp_path): @@ -1110,6 +1111,63 @@ def test_no_owner_flags_send_no_constraint(self): assert result.exit_code == 0, result.output assert inst.list_findings.call_args[1]["owner"] is None + def test_sla_filter_reaches_every_findings_surface(self): + # --sla rides the shared filter decorator, so it must reach ALL FOUR + # commands that decorator feeds. A surface that silently drops it + # returns the unfiltered estate while the caller believes they asked + # for the overdue slice. + for argv, method, extra in ( + (["cloudsec", "finding", "list", "--sla", "breached"], + "list_findings", {"findings": []}), + (["cloudsec", "finding", "facets", "--sla", "breached"], + "get_finding_facets", {"facets": {}}), + (["cloudsec", "finding", "causes", "--sla", "breached"], + "list_finding_causes", {"causes": [], "distinct": 0}), + (["cloudsec", "export", "findings", "--sla", "breached"], + "export_findings_csv", None), + ): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + if extra is None: + result, inst = _invoke(argv, cls) + else: + result, inst = _invoke(argv, cls, return_value=extra) + assert result.exit_code == 0, result.output + got = getattr(inst, method).call_args[1]["sla"] + assert got == ["breached"], f"{method} got sla={got!r}" + + def test_sla_is_repeatable_and_absent_by_default(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "finding", "list", + "--sla", "breached", "--sla", "due_soon"], + cls, return_value={"findings": []}, + ) + assert result.exit_code == 0, result.output + assert inst.list_findings.call_args[1]["sla"] == ["breached", "due_soon"] + + # Unset must be None, not [] — an empty list would be forwarded as a + # present-but-empty selection. + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "finding", "list"], cls, + return_value={"findings": []}, + ) + assert result.exit_code == 0, result.output + assert inst.list_findings.call_args[1]["sla"] is None + + def test_sort_due_at_forwards(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "finding", "list", "--sort", "due_at"], cls, + return_value={"findings": []}, + ) + assert result.exit_code == 0, result.output + assert inst.list_findings.call_args[1]["sort"] == "due_at" + def test_facets_owner_pin_is_separate_from_the_filter(self): p1, p2, p3 = _patches() with p1, p2, p3 as cls: @@ -1155,6 +1213,7 @@ def test_causes_rollup_with_filters(self): status=None, account=None, owner=None, + sla=None, reachable=None, kev=None, q=None,