From c3b3f837693080e5ebff94f2f4849e2376813a04 Mon Sep 17 00:00:00 2001 From: Maxime Lamothe-Brassard Date: Mon, 21 Sep 2026 16:03:06 -0700 Subject: [PATCH 1/3] CS-02 expose bounded IaC provenance SDK and CLI selectors --- doc/cli/cloud-security.md | 28 ++++++++++++++++ limacharlie/commands/cloudsec.py | 39 ++++++++++++++++------ limacharlie/sdk/cloudsec.py | 55 +++++++++++++++++++++++++++++--- tests/unit/test_cli_cloudsec.py | 45 ++++++++++++++++++++++++++ tests/unit/test_sdk_cloudsec.py | 40 +++++++++++++++++++++++ 5 files changed, 192 insertions(+), 15 deletions(-) diff --git a/doc/cli/cloud-security.md b/doc/cli/cloud-security.md index d8f6bb0e..fde286c4 100644 --- a/doc/cli/cloud-security.md +++ b/doc/cli/cloud-security.md @@ -342,3 +342,31 @@ A rule set document holds one entry per rule file, where each `rules` value is a Before the scan starts, the CLI refuses a document the scanner would not accept: unknown fields, a version other than 1, a record without a unique non-empty `key` or a `rules` object, more than 32 MiB, or no rules at all. The scanner checks each rule, then reports and skips any rule it cannot load. **Scanner version.** The default image is pinned to scanner v0.16.0. A `--image` or `--binary` running `sast` must be v0.16.0 or newer, because older scanners reject the rule-set flags. That failure is a usage error (exit 2), and the CLI's error message names the version you need. A scan without `sast` passes no rule-set flag, so it still runs on older scanners. + +### IaC provenance selectors (CS-02) + +When server-side provenance queries are enabled, finding list, facets, causes and +CSV export accept repeatable `--iac-attribution` values (`attributed`, `ambiguous`, +`none`, `unknown`). Findings and inventory list/facets/export accept +`--has-iac-origin` or `--no-has-iac-origin`. Omit both for no origin constraint. +The negative selector means no **recorded** origin evidence, not proof that a +resource has no IaC. Unknown, partial and stale evidence never establish safety. + +```bash +limacharlie cloudsec finding list --iac-attribution unknown --no-has-iac-origin +limacharlie cloudsec inventory list --has-iac-origin +limacharlie cloudsec export findings --iac-attribution ambiguous +``` + +SDK equivalents use keyword arguments `iac_attribution=["unknown"]` and +`has_iac_origin=False`; explicit false is preserved on the wire. Malformed values +are rejected before HTTP. Selectors remain scoped to the bound organization. +Facets exclude their own selector. Resource detail may include bounded +`iac_origin` and `iac_origin_partial` evidence even for a clean resource. Source +permalinks are optional and commit-bound; missing revision metadata is unknown, +not a link to HEAD. These read-only clients do not fetch source URLs. + +Rollout requires the compatible gateway and graph reader before use; a disabled +provenance server rejects the new selectors. Revert clients independently or omit +the selectors; no schema rollback or feature enablement is performed by this CLI. +Program: maximelb/claude-config#137, epic maximelb/claude-config#134. diff --git a/limacharlie/commands/cloudsec.py b/limacharlie/commands/cloudsec.py index 3a455a34..cad480aa 100644 --- a/limacharlie/commands/cloudsec.py +++ b/limacharlie/commands/cloudsec.py @@ -1216,12 +1216,21 @@ def _selector_with_empty(values, include_empty: bool) -> list[str] | None: return out or None +def _iac_origin_option(f): + return click.option("--has-iac-origin/--no-has-iac-origin", default=None, + help="Filter recorded IaC origin evidence. No recorded evidence does not prove no IaC exists; omit for no constraint.")(f) + + def _finding_filter_options(f): """The findings worklist filter selectors (shared by list/facets). Click stacks decorators bottom-up, so the option that should show first in --help is applied last. """ + f = _iac_origin_option(f) + f = click.option("--iac-attribution", "iac_attributions", multiple=True, + type=click.Choice(["attributed", "ambiguous", "none", "unknown"]), + help="IaC attribution verdict; repeat up to four times. Missing evidence is unknown, never safe.")(f) f = click.option( "-q", "--search", "q", default=None, help="Substring search over the findings.", @@ -1347,6 +1356,7 @@ def _finding_filter_options(f): def _inventory_filter_options(f): """The inventory filter selectors (shared by list/export).""" + f = _iac_origin_option(f) f = click.option( "--account-empty", "account_empty", is_flag=True, default=False, help="Select only resources that have no cloud account.", @@ -3082,7 +3092,7 @@ def finding_group() -> None: @_sort_options @_paging_options @pass_context -def finding_list(ctx, severities, finding_classes, statuses, accounts, repos, +def finding_list(ctx, has_iac_origin, iac_attributions, severities, finding_classes, statuses, accounts, repos, image_urns, fix_states, exploit_bands, grains, source, owners, unassigned, sla_states, reachable, kev, q, cause, sort, order, cursor, limit) -> None: @@ -3099,6 +3109,8 @@ def finding_list(ctx, severities, finding_classes, statuses, accounts, repos, """ cs = _get_cloudsec(ctx) _output(ctx, cs.list_findings( + has_iac_origin=has_iac_origin, + iac_attribution=list(iac_attributions) or None, severity=list(severities) or None, finding_class=list(finding_classes) or None, status=list(statuses) or None, @@ -3136,7 +3148,7 @@ def finding_list(ctx, severities, finding_classes, statuses, accounts, repos, "slots with any --owner values, so past ~50 combined a " "pin can still be dropped.") @pass_context -def finding_facets(ctx, severities, finding_classes, statuses, accounts, repos, +def finding_facets(ctx, has_iac_origin, iac_attributions, severities, finding_classes, statuses, accounts, repos, image_urns, fix_states, exploit_bands, grains, source, owners, unassigned, sla_states, reachable, kev, q, cause, owner_pins) -> None: @@ -3149,6 +3161,8 @@ def finding_facets(ctx, severities, finding_classes, statuses, accounts, repos, """ cs = _get_cloudsec(ctx) _output(ctx, cs.get_finding_facets( + has_iac_origin=has_iac_origin, + iac_attribution=list(iac_attributions) or None, severity=list(severities) or None, finding_class=list(finding_classes) or None, status=list(statuses) or None, @@ -3179,7 +3193,7 @@ def finding_facets(ctx, severities, finding_classes, statuses, accounts, repos, help="Rollup size (default 20, cap 200). Not a page size — the " "rollup is not paginated; 'distinct' reports the tail.") @pass_context -def finding_causes(ctx, severities, finding_classes, statuses, accounts, repos, +def finding_causes(ctx, has_iac_origin, iac_attributions, severities, finding_classes, statuses, accounts, repos, image_urns, fix_states, exploit_bands, grains, source, owners, unassigned, sla_states, reachable, kev, q, cause, limit) -> None: @@ -3193,6 +3207,8 @@ def finding_causes(ctx, severities, finding_classes, statuses, accounts, repos, cs = _get_cloudsec(ctx) _output(ctx, cs.list_finding_causes( cause=cause, + has_iac_origin=has_iac_origin, + iac_attribution=list(iac_attributions) or None, severity=list(severities) or None, finding_class=list(finding_classes) or None, status=list(statuses) or None, @@ -3474,7 +3490,7 @@ def inventory_group() -> None: @_inventory_filter_options @_paging_options @pass_context -def inventory_list(ctx, resource_type, provider, account, region, q, +def inventory_list(ctx, has_iac_origin, resource_type, provider, account, region, q, all_accounts, account_empty, cursor, limit) -> None: """List the cloud resource inventory. @@ -3489,7 +3505,7 @@ def inventory_list(ctx, resource_type, provider, account, region, q, cs = _get_cloudsec(ctx) _output(ctx, cs.list_inventory( resource_type=resource_type, provider=provider, account=account, - region=region, q=q, + region=region, q=q, has_iac_origin=has_iac_origin, account_empty=_inventory_account_empty( account, all_accounts, account_empty), cursor=cursor, limit=limit, @@ -3497,8 +3513,9 @@ def inventory_list(ctx, resource_type, provider, account, region, q, @inventory_group.command("facets") +@_iac_origin_option @pass_context -def inventory_facets(ctx) -> None: +def inventory_facets(ctx, has_iac_origin) -> None: """Inventory facet counts by type/account/region. \b @@ -3506,7 +3523,7 @@ def inventory_facets(ctx) -> None: limacharlie cloudsec inventory facets """ cs = _get_cloudsec(ctx) - _output(ctx, cs.get_inventory_facets()) + _output(ctx, cs.get_inventory_facets(has_iac_origin=has_iac_origin)) # --------------------------------------------------------------------------- @@ -4527,7 +4544,7 @@ def export_group() -> None: @_sort_options @_export_output_option @pass_context -def export_findings(ctx, severities, finding_classes, statuses, accounts, repos, +def export_findings(ctx, has_iac_origin, iac_attributions, severities, finding_classes, statuses, accounts, repos, image_urns, fix_states, exploit_bands, grains, source, owners, unassigned, sla_states, reachable, kev, q, cause, sort, order, output_path) -> None: @@ -4541,6 +4558,8 @@ def export_findings(ctx, severities, finding_classes, statuses, accounts, repos, """ cs = _get_cloudsec(ctx) _emit_csv(ctx, cs.export_findings_csv( + has_iac_origin=has_iac_origin, + iac_attribution=list(iac_attributions) or None, severity=list(severities) or None, finding_class=list(finding_classes) or None, status=list(statuses) or None, @@ -4566,7 +4585,7 @@ def export_findings(ctx, severities, finding_classes, statuses, accounts, repos, @_inventory_filter_options @_export_output_option @pass_context -def export_inventory(ctx, resource_type, provider, account, region, q, +def export_inventory(ctx, has_iac_origin, resource_type, provider, account, region, q, all_accounts, account_empty, output_path) -> None: """Export the (filtered) cloud resource inventory as CSV. @@ -4578,7 +4597,7 @@ def export_inventory(ctx, resource_type, provider, account, region, q, cs = _get_cloudsec(ctx) _emit_csv(ctx, cs.export_inventory_csv( resource_type=resource_type, provider=provider, account=account, - region=region, q=q, + region=region, q=q, has_iac_origin=has_iac_origin, account_empty=_inventory_account_empty( account, all_accounts, account_empty), ), output_path) diff --git a/limacharlie/sdk/cloudsec.py b/limacharlie/sdk/cloudsec.py index 1ecdc69f..0e665c11 100644 --- a/limacharlie/sdk/cloudsec.py +++ b/limacharlie/sdk/cloudsec.py @@ -143,8 +143,20 @@ def _inventory_account_selector( return {"account_unscoped": account_unscoped or None} +def _validate_iac_selectors(attribution, origin): + if origin is not None and type(origin) is not bool: + raise ValueError("has_iac_origin must be a boolean or None") + if attribution is not None: + if not isinstance(attribution, (list, tuple)) or not 1 <= len(attribution) <= 4: + raise ValueError("iac_attribution must contain one to four verdicts") + if any(v not in ("attributed", "ambiguous", "none", "unknown") for v in attribution): + raise ValueError("invalid iac_attribution verdict") + + def _finding_query_pairs( *, + has_iac_origin: bool | None = None, + iac_attribution: list[str] | None = None, severity: list[str] | None = None, finding_class: list[str] | None = None, status: list[str] | None = None, @@ -193,7 +205,9 @@ def _finding_query_pairs( with no error and no signal in the response. A script fanning out over more than 100 repositories, owners or image urns must batch them. """ + _validate_iac_selectors(iac_attribution, has_iac_origin) return _query_pairs( + iac_attribution=iac_attribution, has_iac_origin=has_iac_origin, severity=severity, finding_class=finding_class, status=status, account=account, owner=owner, owner_pin=owner_pin, sla=sla, repo=repo, image_urn=image_urn, fix_state=fix_state, @@ -347,6 +361,8 @@ def _post( def list_findings( self, *, + has_iac_origin: bool | None = None, + iac_attribution: list[str] | None = None, severity: list[str] | None = None, finding_class: list[str] | None = None, status: list[str] | None = None, @@ -371,6 +387,8 @@ def list_findings( """List the merged, risk-ranked cloud-security findings. Args: + iac_attribution: One to four attributed, ambiguous, none or unknown verdicts. + has_iac_origin: Recorded origin evidence exists; False is not proof of no IaC. severity: Filter values (CRITICAL/HIGH/MEDIUM/LOW/INFO), OR'd. finding_class: Filter values (toxic_combination, public_exposure, ciem_risk, privilege_escalation, vulnerability, misconfig, @@ -481,6 +499,7 @@ def list_findings( ``{"findings": [...], "next_cursor": str}``. """ return self._get("findings", _finding_query_pairs( + iac_attribution=iac_attribution, has_iac_origin=has_iac_origin, severity=severity, finding_class=finding_class, status=status, account=account, owner=owner, sla=sla, repo=repo, image_urn=image_urn, fix_state=fix_state, @@ -492,6 +511,8 @@ def list_findings( def get_finding_facets( self, *, + has_iac_origin: bool | None = None, + iac_attribution: list[str] | None = None, severity: list[str] | None = None, finding_class: list[str] | None = None, status: list[str] | None = None, @@ -516,6 +537,8 @@ def get_finding_facets( facet dimension is counted against the other active filters. Args: + iac_attribution: One to four attributed, ambiguous, none or unknown verdicts. + has_iac_origin: Recorded origin evidence exists; False is not proof of no IaC. owner_pin: Owners to keep in the ``owner`` facet even when they would not rank into it. NOT a filter — it selects no rows and changes no count. The ``owner`` facet is capped at the @@ -581,6 +604,7 @@ def get_finding_facets( "on_track": 20, "exempt": 0, "none": 900}}}``. """ return self._get("findings/facets", _finding_query_pairs( + iac_attribution=iac_attribution, has_iac_origin=has_iac_origin, severity=severity, finding_class=finding_class, status=status, account=account, owner=owner, owner_pin=owner_pin, sla=sla, repo=repo, image_urn=image_urn, fix_state=fix_state, @@ -591,6 +615,8 @@ def get_finding_facets( def list_finding_causes( self, *, + has_iac_origin: bool | None = None, + iac_attribution: list[str] | None = None, cause: str | None = None, severity: list[str] | None = None, finding_class: list[str] | None = None, @@ -630,6 +656,8 @@ def list_finding_causes( class", not "no findings". Args: + iac_attribution: One to four attributed, ambiguous, none or unknown verdicts. + has_iac_origin: Recorded origin evidence exists; False is not proof of no IaC. cause: One cause key. Set it for the count of that cause alone, returned as a single-entry ``causes`` — or an EMPTY ``causes`` when no finding under the filter carries it, so @@ -659,6 +687,7 @@ def list_finding_causes( authority on its status. """ return self._get("findings/causes", _finding_query_pairs( + iac_attribution=iac_attribution, has_iac_origin=has_iac_origin, severity=severity, finding_class=finding_class, status=status, account=account, owner=owner, sla=sla, repo=repo, image_urn=image_urn, fix_state=fix_state, @@ -946,6 +975,7 @@ def get_identity(self, urn: str) -> dict[str, Any]: def list_inventory( self, *, + has_iac_origin: bool | None = None, resource_type: str | None = None, provider: str | None = None, account: str | None = None, @@ -959,6 +989,7 @@ def list_inventory( """List the cloud resource inventory. Args: + has_iac_origin: Recorded origin evidence exists; False is not proof of no IaC. resource_type: Filter by resource type (the ``type`` selector). provider: Filter by the producing provider sweep (e.g. ``gcp``, ``aws``, ``okta``, ``google_workspace``). @@ -973,17 +1004,26 @@ def list_inventory( Returns: ``{"resources": [...], "next_cursor": str}``. """ + _validate_iac_selectors(None, has_iac_origin) selector = _inventory_account_selector(account_empty, account_unscoped) return self._get("inventory", _query_pairs( type=resource_type, provider=provider, account=account, - region=region, q=q, + region=region, q=q, has_iac_origin=has_iac_origin, **selector, cursor=cursor, limit=limit, )) - def get_inventory_facets(self) -> dict[str, Any]: - """Inventory facet counts (by type/account/region).""" - return self._get("inventory/facets") + def get_inventory_facets(self, *, has_iac_origin: bool | None = None) -> dict[str, Any]: + """Get inventory facet counts under an optional origin selector. + + Args: + has_iac_origin: Recorded origin evidence exists; False is not proof of no IaC. + + Returns: + dict: Cross-filtered type, account, region and available origin buckets. + """ + _validate_iac_selectors(None, has_iac_origin) + return self._get("inventory/facets", _query_pairs(has_iac_origin=has_iac_origin)) def get_topology(self) -> dict[str, Any]: """Pre-aggregated estate topology (exact at any estate size). @@ -2979,6 +3019,8 @@ def get_fleet_overview( def export_findings_csv( self, *, + has_iac_origin: bool | None = None, + iac_attribution: list[str] | None = None, severity: list[str] | None = None, finding_class: list[str] | None = None, status: list[str] | None = None, @@ -3011,6 +3053,7 @@ def export_findings_csv( The CSV document as a string. """ pairs = _finding_query_pairs( + iac_attribution=iac_attribution, has_iac_origin=has_iac_origin, severity=severity, finding_class=finding_class, status=status, account=account, owner=owner, sla=sla, repo=repo, image_urn=image_urn, fix_state=fix_state, @@ -3024,6 +3067,7 @@ def export_findings_csv( def export_inventory_csv( self, *, + has_iac_origin: bool | None = None, resource_type: str | None = None, provider: str | None = None, account: str | None = None, @@ -3041,10 +3085,11 @@ def export_inventory_csv( Returns: The CSV document as a string. """ + _validate_iac_selectors(None, has_iac_origin) selector = _inventory_account_selector(account_empty, account_unscoped) pairs = _query_pairs( type=resource_type, provider=provider, account=account, - region=region, q=q, + region=region, q=q, has_iac_origin=has_iac_origin, **selector, ) pairs.append(("format", "csv")) diff --git a/tests/unit/test_cli_cloudsec.py b/tests/unit/test_cli_cloudsec.py index 565c148b..b386063c 100644 --- a/tests/unit/test_cli_cloudsec.py +++ b/tests/unit/test_cli_cloudsec.py @@ -326,6 +326,7 @@ def test_list_repeatable_filters(self): ) assert result.exit_code == 0, result.output inst.list_findings.assert_called_once_with( + has_iac_origin=None, iac_attribution=None, severity=["CRITICAL", "HIGH"], finding_class=["toxic_combination"], status=None, @@ -927,6 +928,7 @@ def test_list_forwards_provider(self): ) assert result.exit_code == 0, result.output inst.list_inventory.assert_called_once_with( + has_iac_origin=None, resource_type=None, provider="okta", account=None, region=None, q=None, account_empty=None, cursor=None, limit=None, @@ -950,6 +952,7 @@ def test_export_findings_stdout(self): # Raw CSV on stdout, NOT the JSON renderer. assert result.output == "col_a,col_b\n1,2\n" inst.export_findings_csv.assert_called_once_with( + has_iac_origin=None, iac_attribution=None, severity=["CRITICAL"], finding_class=None, status=["open"], account=None, repo=None, image_urn=None, fix_state=None, exploit_band=None, grain=None, cause=None, @@ -976,6 +979,7 @@ def test_export_inventory(self): ) assert result.exit_code == 0, result.output inst.export_inventory_csv.assert_called_once_with( + has_iac_origin=None, resource_type="Bucket", provider="gcp", account=None, region=None, q=None, account_empty=None, ) @@ -1062,6 +1066,7 @@ def test_list_all_accounts_flag(self): ) assert result.exit_code == 0, result.output inst.list_inventory.assert_called_once_with( + has_iac_origin=None, resource_type=None, provider=None, account=None, region=None, q=None, account_empty=None, cursor=None, limit=None, @@ -1076,6 +1081,7 @@ def test_list_account_empty_flag(self): ) assert result.exit_code == 0, result.output inst.list_inventory.assert_called_once_with( + has_iac_origin=None, resource_type=None, provider=None, account=None, region=None, q=None, account_empty=True, cursor=None, limit=None, @@ -1101,6 +1107,7 @@ def test_export_all_accounts_flag(self): ) assert result.exit_code == 0, result.output inst.export_inventory_csv.assert_called_once_with( + has_iac_origin=None, resource_type=None, provider=None, account=None, region=None, q=None, account_empty=None, ) @@ -1113,6 +1120,7 @@ def test_export_account_empty_flag(self): ) assert result.exit_code == 0, result.output inst.export_inventory_csv.assert_called_once_with( + has_iac_origin=None, resource_type=None, provider=None, account=None, region=None, q=None, account_empty=True, ) @@ -1371,6 +1379,7 @@ def test_causes_rollup_with_filters(self): ) assert result.exit_code == 0, result.output inst.list_finding_causes.assert_called_once_with( + has_iac_origin=None, iac_attribution=None, cause=None, severity=["CRITICAL"], finding_class=None, @@ -2918,3 +2927,39 @@ def test_scope_hierarchy(self): ) assert result.exit_code == 0, result.output inst.get_azure_scope_hierarchy.assert_called_once_with() + + +class TestIaCSelectors: + @pytest.mark.parametrize('command,method', [ + (['finding', 'list'], 'list_findings'), + (['finding', 'facets'], 'get_finding_facets'), + (['finding', 'causes'], 'list_finding_causes'), + (['export', 'findings'], 'export_findings_csv'), + ]) + def test_finding_flags(self, command, method): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke(['cloudsec', *command, '--no-has-iac-origin', '--iac-attribution', 'unknown'], cls) + assert result.exit_code == 0, result.output + kwargs = getattr(inst, method).call_args.kwargs + assert kwargs['has_iac_origin'] is False + assert kwargs['iac_attribution'] == ['unknown'] + + @pytest.mark.parametrize('command,method', [ + (['inventory', 'list'], 'list_inventory'), + (['inventory', 'facets'], 'get_inventory_facets'), + (['export', 'inventory'], 'export_inventory_csv'), + ]) + def test_inventory_flags(self, command, method): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke(['cloudsec', *command, '--no-has-iac-origin'], cls) + assert result.exit_code == 0, result.output + assert getattr(inst, method).call_args.kwargs['has_iac_origin'] is False + + def test_unknown_verdict_refused(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke(['cloudsec', 'finding', 'list', '--iac-attribution', 'safe'], cls) + assert result.exit_code != 0 + inst.list_findings.assert_not_called() diff --git a/tests/unit/test_sdk_cloudsec.py b/tests/unit/test_sdk_cloudsec.py index 5139cb95..e1a6ba29 100644 --- a/tests/unit/test_sdk_cloudsec.py +++ b/tests/unit/test_sdk_cloudsec.py @@ -1409,3 +1409,43 @@ def test_azure_scope_hierarchy(self, cs, mock_org): url, qp = _get_call(mock_org) assert url == f"cloudsec/{OID}/azure/scope-hierarchy" assert qp is None + + +class TestIaCQuerySelectors: + @pytest.mark.parametrize('method', ['list_findings', 'get_finding_facets', 'list_finding_causes', 'export_findings_csv']) + def test_same_selectors_and_tenant_on_every_finding_surface(self, cs, mock_org, method): + mock_org.client.request.return_value = {} + getattr(cs, method)(has_iac_origin=False, iac_attribution=['unknown', 'ambiguous']) + url, pairs = _get_call(mock_org) + assert url.startswith(f'cloudsec/{OID}/') + assert ('has_iac_origin', 'false') in pairs + assert [v for k, v in pairs if k == 'iac_attribution'] == ['unknown', 'ambiguous'] + assert all(k != 'oid' for k, _ in pairs) + + @pytest.mark.parametrize('method', ['list_inventory', 'get_inventory_facets', 'export_inventory_csv']) + def test_inventory_false_is_not_omitted(self, cs, mock_org, method): + getattr(cs, method)(has_iac_origin=False) + url, pairs = _get_call(mock_org) + assert url.startswith(f'cloudsec/{OID}/') + assert ('has_iac_origin', 'false') in pairs + + @pytest.mark.parametrize('value', ['', 'false', 0, 1, [], {}]) + @pytest.mark.parametrize('method', ['list_findings', 'get_finding_facets', 'list_finding_causes', 'export_findings_csv', 'list_inventory', 'get_inventory_facets', 'export_inventory_csv']) + def test_malformed_boolean_never_sends_request(self, cs, mock_org, value, method): + with pytest.raises(ValueError): + getattr(cs, method)(has_iac_origin=value) + mock_org.client.request.assert_not_called() + + @pytest.mark.parametrize('value', [[], 'unknown', ['safe'], ['unknown'] * 5, [None], [{}], ['UNKNOWN']]) + def test_malformed_verdict_never_sends_request(self, cs, mock_org, value): + with pytest.raises(ValueError): + cs.list_findings(iac_attribution=value) + mock_org.client.request.assert_not_called() + + def test_other_tenant_cannot_change_bound_client(self, cs, mock_org): + other = MagicMock() + other.oid = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee' + CloudSec(other).list_inventory(has_iac_origin=True) + cs.list_inventory(has_iac_origin=True) + assert _get_call(mock_org)[0] == f'cloudsec/{OID}/inventory' + assert _get_call(other)[0] == f'cloudsec/{other.oid}/inventory' From a59b6a3db3b484040c2b3a2628c4bec43245720d Mon Sep 17 00:00:00 2001 From: Maxime Lamothe-Brassard Date: Mon, 21 Sep 2026 16:04:26 -0700 Subject: [PATCH 2/3] CS-02 clarify strict selector bounds in SDK documentation --- limacharlie/sdk/cloudsec.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/limacharlie/sdk/cloudsec.py b/limacharlie/sdk/cloudsec.py index 0e665c11..0e53fab6 100644 --- a/limacharlie/sdk/cloudsec.py +++ b/limacharlie/sdk/cloudsec.py @@ -201,9 +201,10 @@ def _finding_query_pairs( the vocabularies live at the backend, and a value outside one returns an empty page rather than silently widening the read. - Every repeatable selector is TRUNCATED AT 100 VALUES by the gateway, + Legacy repeatable selectors are TRUNCATED AT 100 VALUES by the gateway, with no error and no signal in the response. A script fanning out over more than 100 repositories, owners or image urns must batch them. + IaC attribution instead rejects more than four or unknown verdicts locally. """ _validate_iac_selectors(iac_attribution, has_iac_origin) return _query_pairs( From d5ca7bbbd5084910a3b5786b9e35255ac40f0835 Mon Sep 17 00:00:00 2001 From: Maxime Lamothe-Brassard Date: Mon, 21 Sep 2026 16:26:37 -0700 Subject: [PATCH 3/3] CS-02 require applied-selector receipts for JSON and CSV reads --- doc/cli/cloud-security.md | 6 ++++++ limacharlie/sdk/cloudsec.py | 29 ++++++++++++++++++++++++++++- tests/unit/test_sdk_cloudsec.py | 29 ++++++++++++++++++++++++++++- 3 files changed, 62 insertions(+), 2 deletions(-) diff --git a/doc/cli/cloud-security.md b/doc/cli/cloud-security.md index fde286c4..52c72418 100644 --- a/doc/cli/cloud-security.md +++ b/doc/cli/cloud-security.md @@ -370,3 +370,9 @@ Rollout requires the compatible gateway and graph reader before use; a disabled provenance server rejects the new selectors. Revert clients independently or omit the selectors; no schema rollback or feature enablement is performed by this CLI. Program: maximelb/claude-config#137, epic maximelb/claude-config#134. + +New-selector requests require an exact `applied_iac_filters` receipt in JSON. +CSV responses carry a bounded first comment line with a base64url JSON receipt; +the SDK validates and removes it before returning CSV. Older or partly upgraded +servers that do not acknowledge the requested selectors raise an error instead +of presenting an unfiltered result. Selector-free calls remain unchanged. diff --git a/limacharlie/sdk/cloudsec.py b/limacharlie/sdk/cloudsec.py index 0e53fab6..012a56a4 100644 --- a/limacharlie/sdk/cloudsec.py +++ b/limacharlie/sdk/cloudsec.py @@ -153,6 +153,32 @@ def _validate_iac_selectors(attribution, origin): raise ValueError("invalid iac_attribution verdict") +def _require_iac_receipt(response, pairs, raw_response=False): + expected = {} + for key, value in pairs or []: + if key == "has_iac_origin": + expected[key] = value == "true" + elif key == "iac_attribution": + expected.setdefault(key, []).append(value) + if not expected: + return response + try: + if raw_response: + line, separator, body = response.partition("\n") + prefix = "# lc_iac_filters_v1=" + if not separator or len(line) > 1024 or not line.startswith(prefix): + raise ValueError("missing receipt") + encoded = line[len(prefix):].rstrip("\r") + applied = json.loads(base64.b64decode(encoded + "=" * (-len(encoded) % 4), altchars=b"-_", validate=True)) + else: + applied = response.get("applied_iac_filters") + if json.dumps(applied, sort_keys=True, separators=(",", ":")) != json.dumps(expected, sort_keys=True, separators=(",", ":")): + raise ValueError("receipt mismatch") + except (ValueError, TypeError, AttributeError): + raise RuntimeError("IaC query selectors were not acknowledged by this server version") from None + return body if raw_response else response + + def _finding_query_pairs( *, has_iac_origin: bool | None = None, @@ -331,12 +357,13 @@ def _get( *, raw_response: bool = False, ) -> Any: - return self._org.client.request( + response = self._org.client.request( "GET", f"cloudsec/{self.oid}/{path}", query_params=query_params or None, raw_response=raw_response, ) + return _require_iac_receipt(response, query_params, raw_response) def _post( self, diff --git a/tests/unit/test_sdk_cloudsec.py b/tests/unit/test_sdk_cloudsec.py index e1a6ba29..0ca7e7ac 100644 --- a/tests/unit/test_sdk_cloudsec.py +++ b/tests/unit/test_sdk_cloudsec.py @@ -1411,10 +1411,17 @@ def test_azure_scope_hierarchy(self, cs, mock_org): assert qp is None +def _iac_response(receipt, method): + if method.startswith('export_'): + token = base64.urlsafe_b64encode(json.dumps(receipt, sort_keys=True, separators=(',', ':')).encode()).decode().rstrip('=') + return '# lc_iac_filters_v1=' + token + '\nname\nfixture\n' + return {'applied_iac_filters': receipt} + + class TestIaCQuerySelectors: @pytest.mark.parametrize('method', ['list_findings', 'get_finding_facets', 'list_finding_causes', 'export_findings_csv']) def test_same_selectors_and_tenant_on_every_finding_surface(self, cs, mock_org, method): - mock_org.client.request.return_value = {} + mock_org.client.request.return_value = _iac_response({"has_iac_origin": False, "iac_attribution": ["unknown", "ambiguous"]}, method) getattr(cs, method)(has_iac_origin=False, iac_attribution=['unknown', 'ambiguous']) url, pairs = _get_call(mock_org) assert url.startswith(f'cloudsec/{OID}/') @@ -1424,6 +1431,7 @@ def test_same_selectors_and_tenant_on_every_finding_surface(self, cs, mock_org, @pytest.mark.parametrize('method', ['list_inventory', 'get_inventory_facets', 'export_inventory_csv']) def test_inventory_false_is_not_omitted(self, cs, mock_org, method): + mock_org.client.request.return_value = _iac_response({"has_iac_origin": False}, method) getattr(cs, method)(has_iac_origin=False) url, pairs = _get_call(mock_org) assert url.startswith(f'cloudsec/{OID}/') @@ -1445,7 +1453,26 @@ def test_malformed_verdict_never_sends_request(self, cs, mock_org, value): def test_other_tenant_cannot_change_bound_client(self, cs, mock_org): other = MagicMock() other.oid = 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee' + other.client.request.return_value = _iac_response({'has_iac_origin': True}, 'list_inventory') + mock_org.client.request.return_value = _iac_response({'has_iac_origin': True}, 'list_inventory') CloudSec(other).list_inventory(has_iac_origin=True) cs.list_inventory(has_iac_origin=True) assert _get_call(mock_org)[0] == f'cloudsec/{OID}/inventory' assert _get_call(other)[0] == f'cloudsec/{other.oid}/inventory' + + +@pytest.mark.parametrize('method', ['list_findings', 'get_finding_facets', 'list_finding_causes', 'export_findings_csv', 'list_inventory', 'get_inventory_facets', 'export_inventory_csv']) +def test_iac_query_refuses_legacy_or_mismatched_server(cs, mock_org, method): + for receipt in [None, {}, {'has_iac_origin': True}, {'has_iac_origin': 0}]: + mock_org.client.request.return_value = _iac_response(receipt, method) + with pytest.raises(RuntimeError, match='not acknowledged'): + getattr(cs, method)(has_iac_origin=False) + + +def test_iac_csv_receipt_is_removed_after_validation(cs, mock_org): + mock_org.client.request.return_value = _iac_response({'has_iac_origin': False}, 'export_inventory_csv') + assert cs.export_inventory_csv(has_iac_origin=False) == 'name\nfixture\n' + for invalid in ['name\nfixture\n', '# lc_iac_filters_v1=%%%\nname\n', '# lc_iac_filters_v1=' + 'a' * 1024 + '\n']: + mock_org.client.request.return_value = invalid + with pytest.raises(RuntimeError, match='not acknowledged'): + cs.export_inventory_csv(has_iac_origin=False)