From 02026d2b5c8453fac97cdeb60ec866aa86691bba Mon Sep 17 00:00:00 2001 From: Maxime Lamothe-Brassard Date: Sat, 1 Aug 2026 18:55:25 -0700 Subject: [PATCH] cloudsec: bind the SLA state selector and the due_at sort Adds `sla` (breached | due_soon | on_track | exempt | none) and the `due_at` sort key to the cloudsec findings surface, keeping the SDK at full parity with the gateway's findings selectors. It rides the shared `_finding_query_pairs` / `_finding_filter_options` helpers, so it reaches all four surfaces those feed - `finding list`, `finding facets`, `finding causes`, and `export findings` - and a test asserts that explicitly: a surface that silently dropped the selector would return the unfiltered estate while the caller believed they had asked for the overdue slice. Docs record the two things about this dimension that are not guessable: the state is DERIVED at read time (there is no built-in default SLA, so every finding reads `none` until the org writes an `sla` policy), and `--sort due_at` is the one sort key that defaults to ASCENDING and places findings with no due date last rather than excluding them. The three existing exhaustive-kwargs assertions gained `sla=None` so they keep asserting the complete call rather than a subset. Tests: 3862 passed, 5 skipped (the CI invocation: pytest tests/unit/ tests/microbenchmarks/ --benchmark-disable). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01Y3AnxeGmcGA9scrsRZFtg9 --- limacharlie/commands/cloudsec.py | 30 +++++++++++---- limacharlie/sdk/cloudsec.py | 47 ++++++++++++++++++++---- tests/unit/test_cli_cloudsec.py | 63 +++++++++++++++++++++++++++++++- 3 files changed, 123 insertions(+), 17 deletions(-) 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,