cloudsec: bind the SLA state selector and the due_at sort

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,

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"))

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,