diff --git a/doc/cli/cloud-security.md b/doc/cli/cloud-security.md index 35c37894..a8ea1a88 100644 --- a/doc/cli/cloud-security.md +++ b/doc/cli/cloud-security.md @@ -343,6 +343,41 @@ Before the scan starts, the CLI refuses a document the scanner would not accept: **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. + +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. + + ## Build provenance `limacharlie cloudsec code provenance push -f provenance.json` sends an LC diff --git a/limacharlie/commands/cloudsec.py b/limacharlie/commands/cloudsec.py index d1f412a6..d1acaf9e 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.", @@ -3111,7 +3121,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: @@ -3128,6 +3138,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, @@ -3165,7 +3177,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: @@ -3178,6 +3190,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, @@ -3208,7 +3222,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: @@ -3222,6 +3236,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, @@ -3503,7 +3519,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. @@ -3518,7 +3534,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, @@ -3526,8 +3542,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 @@ -3535,7 +3552,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)) # --------------------------------------------------------------------------- @@ -4556,7 +4573,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: @@ -4570,6 +4587,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, @@ -4595,7 +4614,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. @@ -4607,7 +4626,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 23446ccd..5b2f420e 100644 --- a/limacharlie/sdk/cloudsec.py +++ b/limacharlie/sdk/cloudsec.py @@ -143,8 +143,46 @@ 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 _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, + iac_attribution: list[str] | None = None, severity: list[str] | None = None, finding_class: list[str] | None = None, status: list[str] | None = None, @@ -189,11 +227,14 @@ 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( + 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, @@ -316,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, @@ -347,6 +389,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 +415,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 +527,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 +539,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 +565,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 +632,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 +643,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 +684,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 +715,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 +1003,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 +1017,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 +1032,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). @@ -3026,6 +3094,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, @@ -3058,6 +3128,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, @@ -3071,6 +3142,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, @@ -3088,10 +3160,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 4f483e87..b16bfad5 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, @@ -2920,6 +2929,42 @@ def test_scope_hierarchy(self): 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() + + class TestProvenanceCommands: def test_push_preserves_bytes(self, tmp_path): document = tmp_path / "provenance.json" diff --git a/tests/unit/test_sdk_cloudsec.py b/tests/unit/test_sdk_cloudsec.py index 65a5cb56..3285eaa2 100644 --- a/tests/unit/test_sdk_cloudsec.py +++ b/tests/unit/test_sdk_cloudsec.py @@ -1410,6 +1410,74 @@ def test_azure_scope_hierarchy(self, cs, mock_org): assert url == f"cloudsec/{OID}/azure/scope-hierarchy" 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 = _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}/') + 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): + 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}/') + 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' + 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) + + class TestCodeProvenance: def test_raw_bundle_bytes_and_tenant_path(self, cs, mock_org): raw = b'{ "mediaType" : "bundle" }'