diff --git a/limacharlie/cli.py b/limacharlie/cli.py index 96d48d61..033cda5c 100644 --- a/limacharlie/cli.py +++ b/limacharlie/cli.py @@ -111,6 +111,7 @@ def _config_no_warnings() -> bool: "billing": ("billing", "group"), "case": ("case_cmd", "group"), "cloud-adapter": ("cloud_sensor", "group"), + "cloudsec": ("cloudsec", "group"), "completion": ("completion", "cmd"), "config": ("config_cmd", "group"), "detection": ("detection", "group"), diff --git a/limacharlie/client.py b/limacharlie/client.py index ec2bca4b..1305959d 100644 --- a/limacharlie/client.py +++ b/limacharlie/client.py @@ -552,7 +552,7 @@ def _call_jwt_endpoint(self, auth_data: dict[str, Any]) -> str: except Exception as e: raise AuthenticationError(f"Failed to get JWT: {e}") - def _rest_call(self, url: str, verb: str, params: dict[str, Any] | None = None, alt_root: str | None = None, query_params: dict[str, str] | None = None, + def _rest_call(self, url: str, verb: str, params: dict[str, Any] | None = None, alt_root: str | None = None, query_params: dict[str, Any] | list[tuple[str, str]] | None = None, raw_body: bytes | None = None, content_type: str | None = None, is_no_auth: bool = False, timeout: int | None = None, extra_headers: dict[str, str] | None = None) -> tuple[int, Any]: """Make a single HTTP request to the API. @@ -575,7 +575,9 @@ def _rest_call(self, url: str, verb: str, params: dict[str, Any] | None = None, full_url = f"{alt_root}/{url}" if url else alt_root if query_params: - full_url = f"{full_url}?{urlencode(query_params)}" + # doseq so sequence values expand to repeated keys + # (?severity=HIGH&severity=LOW) instead of a Python list repr. + full_url = f"{full_url}?{urlencode(query_params, doseq=True)}" # Build request body if raw_body is not None: @@ -645,7 +647,7 @@ def _rest_call(self, url: str, verb: str, params: dict[str, Any] | None = None, self._debug(f"SSL error: {e}") return (HTTP_GATEWAY_TIMEOUT, {"error": f"SSL error: {e}"}) - def request(self, verb: str, url: str, params: dict[str, Any] | None = None, alt_root: str | None = None, query_params: dict[str, str] | None = None, + def request(self, verb: str, url: str, params: dict[str, Any] | None = None, alt_root: str | None = None, query_params: dict[str, Any] | list[tuple[str, str]] | None = None, raw_body: bytes | None = None, content_type: str | None = None, is_no_auth: bool = False, max_retries: int = 3, timeout: int | None = None, extra_headers: dict[str, str] | None = None) -> dict[str, Any]: """Make an API request with retry logic and JWT management. @@ -752,7 +754,7 @@ def request(self, verb: str, url: str, params: dict[str, Any] | None = None, alt return data def raw_request(self, verb: str, url: str, params: dict[str, Any] | None = None, alt_root: str | None = None, - query_params: dict[str, str] | None = None, raw_body: bytes | None = None, + query_params: dict[str, Any] | list[tuple[str, str]] | None = None, raw_body: bytes | None = None, content_type: str | None = None, is_no_auth: bool = False, extra_headers: dict[str, str] | None = None) -> tuple[int, Any]: """Make a raw API request, returning (status_code, response_data). diff --git a/limacharlie/commands/_input_helpers.py b/limacharlie/commands/_input_helpers.py new file mode 100644 index 00000000..3e233065 --- /dev/null +++ b/limacharlie/commands/_input_helpers.py @@ -0,0 +1,53 @@ +"""Shared CLI input-loading helpers. + +The canonical YAML-or-JSON input loaders for command modules. Several +older modules (hive, ioc, output_cmd, replay_cmd, usp, dr) carry their +own private copies of this idiom; new code should import from here so +parser and error-message fixes land in one place. + +Both loaders raise ``click.BadParameter`` with a clean usage message on +unparseable input instead of letting raw tracebacks reach the user. +""" + +from __future__ import annotations + +import json +import sys +from typing import Any + +import click +import yaml + + +def load_file(path: str, param_hint: str) -> Any: + """Read + parse a JSON or YAML file, raising a clean usage error.""" + try: + with open(path, "r") as f: + content = f.read() + except OSError as exc: + raise click.BadParameter(f"cannot read file: {exc}", param_hint=param_hint) + return parse_yaml_or_json(content, param_hint) + + +def load_stdin() -> Any: + """Parse piped stdin as YAML-or-JSON; ``None`` when stdin is a TTY.""" + if sys.stdin.isatty(): + return None + return parse_yaml_or_json(sys.stdin.read(), "stdin") + + +def parse_yaml_or_json(content: str, param_hint: str) -> Any: + """Parse a string as YAML first (a JSON superset), then JSON. + + Raises ``click.BadParameter`` when the content is neither. + """ + try: + return yaml.safe_load(content) + except Exception: + pass + try: + return json.loads(content) + except json.JSONDecodeError as exc: + raise click.BadParameter( + f"input is neither valid YAML nor JSON: {exc}", param_hint=param_hint, + ) diff --git a/limacharlie/commands/cloudsec.py b/limacharlie/commands/cloudsec.py new file mode 100644 index 00000000..aebe7237 --- /dev/null +++ b/limacharlie/commands/cloudsec.py @@ -0,0 +1,1406 @@ +"""Cloud Security (CNAPP) commands for LimaCharlie CLI v2. + +Commands for the ``/cloudsec`` API surface: the merged, risk-ranked +findings worklist (CSPM misconfigurations + attack paths + CIEM), +the cloud resource inventory and security graph, compliance +assessment, the risk overview, CAASM (third-party asset attack +surface), sensor<->cloud-asset resolution, and finding triage. + +Reads require the ``cloudsec.get`` permission and writes require +``cloudsec.set``. Every command requires the org to be subscribed to +the ``ext-cloud-inventory`` extension: + + limacharlie extension subscribe --name ext-cloud-inventory + +Provider credentials and the cloudsec policies are hive records — +manage them with the hive commands (``limacharlie hive list +cloudsec_provider``, ``... cloudsec_policy``, ``... cloudsec_query``); +the one provider command here is the pre-save credential preflight +(``cloudsec provider test``). +""" + +from __future__ import annotations + +import json +from typing import Any + +import click + +from ..cli import pass_context +from ..client import Client +from ..sdk.organization import Organization +from ..sdk.cloudsec import CloudSec +from ..output import format_output, detect_output_format +from ..discovery import register_explain + + +# --------------------------------------------------------------------------- +# Explain texts +# --------------------------------------------------------------------------- + +_EXPLAIN_OVERVIEW = """\ +Composed risk overview for the org in one round-trip: the risk +score, severity distribution, top attack paths, account coverage, +the score trend, and recent finding changes. + +Examples: + limacharlie cloudsec overview + limacharlie cloudsec overview --trend-days 90 +""" + +_EXPLAIN_CHANGES = """\ +Recent cloud-finding lifecycle changes (created/closed), newest +first. + +Examples: + limacharlie cloudsec changes + limacharlie cloudsec changes --limit 100 +""" + +_EXPLAIN_RISK_TREND = """\ +The org's risk-score history over time (the Overview sparkline), +oldest first. + +Examples: + limacharlie cloudsec risk-trend --trend-days 90 +""" + +_EXPLAIN_SCAN_STATUS = """\ +Cloud-collection run status for one provider: whether a sweep is in +progress, when it last started/completed, the last diff stats, and +any error. + +Examples: + limacharlie cloudsec scan-status + limacharlie cloudsec scan-status --provider aws +""" + +_EXPLAIN_FINDING_LIST = """\ +List the merged, risk-ranked cloud-security findings (CSPM +misconfigurations + graph toxic-combination attack paths + CIEM +access), ordered by lc_risk. + +Repeatable filters are OR within a key and AND across keys: + + --severity CRITICAL --severity HIGH --class toxic_combination + +Finding classes: toxic_combination, public_exposure, ciem_risk, +privilege_escalation, vulnerability, misconfig, malware, secret, +scan_finding, coverage_gap. + +Pagination is keyset-based: pass the previous page's next_cursor +via --cursor. + +Examples: + limacharlie cloudsec finding list --severity CRITICAL + limacharlie cloudsec finding list --class public_exposure --kev + limacharlie cloudsec finding list --status open --limit 50 +""" + +_EXPLAIN_FINDING_FACETS = """\ +Cross-filtered facet counts for the findings worklist under the +same filter selectors as 'finding list'. Each facet dimension is +counted against the other active filters. + +Example: + limacharlie cloudsec finding facets --severity CRITICAL +""" + +_EXPLAIN_FINDING_GET = """\ +Get a single finding by its id (e.g. fnd_). + +Example: + limacharlie cloudsec finding get fnd_0123abcd... +""" + +_EXPLAIN_FINDING_RESOLVE = """\ +Disposition a finding: record an operator resolution, or reopen it. + +--kind is one of: + mitigated the risk was fixed + accepted the risk is accepted (optionally until --expires-at) + false_positive the finding was wrong + open clear the disposition and reopen (owner/ticket kept) + +--expires-at is unix seconds and only meaningful with 'accepted'. + +Examples: + limacharlie cloudsec finding resolve fnd_abc... --kind mitigated --reason "SG tightened" + limacharlie cloudsec finding resolve fnd_abc... --kind accepted --expires-at 1767225600 + limacharlie cloudsec finding resolve fnd_abc... --kind open +""" + +_EXPLAIN_FINDING_BULK_RESOLVE = """\ +Apply one resolution to many findings in a single call. + +--kind is mitigated, accepted, or false_positive. Reopening is a +single-finding operation only (the bulk API does not accept 'open'); +use 'finding resolve --kind open' per finding. + +Example: + limacharlie cloudsec finding bulk-resolve --finding-id fnd_a... --finding-id fnd_b... \\ + --kind false_positive --reason "scanner artifact" +""" + +_EXPLAIN_FINDING_SET_OWNER = """\ +Assign the owner of a finding, or clear it with --clear. + +Examples: + limacharlie cloudsec finding set-owner fnd_abc... --owner alice@corp.com + limacharlie cloudsec finding set-owner fnd_abc... --clear +""" + +_EXPLAIN_FINDING_SET_TICKET = """\ +Link a ticket id/url to a finding, or clear it with --clear. + +Examples: + limacharlie cloudsec finding set-ticket fnd_abc... --ticket JIRA-123 + limacharlie cloudsec finding set-ticket fnd_abc... --clear +""" + +_EXPLAIN_ATTACK_PATH_LIST = """\ +The headline toxic-combination attack paths for the org +(internet-exposed workload with a KEV vulnerability that can reach +a sensitive resource). Accepts the severity/account/status/q +selectors to narrow the list. + +Examples: + limacharlie cloudsec attack-path list + limacharlie cloudsec attack-path list --severity CRITICAL +""" + +_EXPLAIN_CIEM_PUBLIC_ACCESS = """\ +CIEM: which identities have public or external access to sensitive +resources. + +Example: + limacharlie cloudsec ciem public-access +""" + +_EXPLAIN_CIEM_FACETS = """\ +CIEM identity facet counts (identity kinds, external/public splits). + +Example: + limacharlie cloudsec ciem facets +""" + +_EXPLAIN_INVENTORY_LIST = """\ +List the cloud resource inventory (system-of-record rows). + +Examples: + limacharlie cloudsec inventory list + limacharlie cloudsec inventory list --type gcp_bucket --region us-central1 + limacharlie cloudsec inventory list -q prod --limit 50 +""" + +_EXPLAIN_INVENTORY_FACETS = """\ +Inventory facet counts by type/account/region. + +Example: + limacharlie cloudsec inventory facets +""" + +_EXPLAIN_DATA_SECURITY_FACETS = """\ +DSPM data-store facet counts: total/sensitive/public/public-sensitive +data stores plus store-kind, sensitivity, and exposure histograms. + +Example: + limacharlie cloudsec data-security facets +""" + +_EXPLAIN_RESOURCE_GET = """\ +Get the single canonical record for any urn the system-of-record or +security graph knows (including derived nodes like vulnerabilities +and identities that have no inventory row). Returns a null resource +when the urn is unknown. + +Example: + limacharlie cloudsec resource get "lcrn:gcp:...:bucket/prod-data" +""" + +_EXPLAIN_GRAPH_NEIGHBORS = """\ +Expand a resource's 1-hop neighborhood in the security graph: every +node directly connected to the urn (either direction) plus the +connecting edges. Server-bounded and ranked (sensitive -> public -> +data/identity); 'truncated' is true past the cap. + +Examples: + limacharlie cloudsec graph neighbors "lcrn:gcp:...:instance/web-1" + limacharlie cloudsec graph neighbors "lcrn:..." --limit 500 +""" + +_EXPLAIN_QUERY_LIST = """\ +List the named graph queries available in the query pack (name, +title, description, and the underlying DSL). + +Saved org-defined queries live in the cloudsec_query hive +(limacharlie hive list cloudsec_query). + +Example: + limacharlie cloudsec query list +""" + +_EXPLAIN_QUERY_RUN = """\ +Run a graph query against the org's security graph. Provide exactly +one of --named (a query-pack name), --text (a text query), or +--query-json (a raw DSL object). Returns alias->urn rows. + +Examples: + limacharlie cloudsec query run --named public-buckets + limacharlie cloudsec query run --text "public bucket with sensitive data" + limacharlie cloudsec query run --query-json '{"match": ...}' --project a,b +""" + +_EXPLAIN_COMPLIANCE_REPORT = """\ +Per-control pass/fail compliance assessment against the org's open +findings, with evidence and a summary score. Defaults to the +cis-gcp framework; pass --assignment to evaluate a named scoped +assignment instead (--framework is then ignored). + +Examples: + limacharlie cloudsec compliance report + limacharlie cloudsec compliance report --framework cis-aws + limacharlie cloudsec compliance report --assignment prod-scope +""" + +_EXPLAIN_COMPLIANCE_FRAMEWORKS = """\ +List the selectable compliance frameworks (id, name, version, +control count). + +Example: + limacharlie cloudsec compliance frameworks +""" + +_EXPLAIN_COMPLIANCE_ASSIGNMENTS = """\ +List the org's scoped compliance assignments (name, framework, +scope, and a full scoped summary score per assignment). Empty when +the org has defined none. + +Example: + limacharlie cloudsec compliance assignments +""" + +_EXPLAIN_CHOKEPOINT_LIST = """\ +Estate-wide chokepoints: the shared attack-path hops ranked by how +many distinct paths each one breaks, plus the total path count — so +"fix this one resource" can be framed as "closes N of M paths". + +Example: + limacharlie cloudsec chokepoint list +""" + +_EXPLAIN_CHOKEPOINT_DISMISS = """\ +Dismiss an estate-wide choke point (by its resource urn) so it no +longer surfaces on the risk overview. Optionally records a reason. + +Example: + limacharlie cloudsec chokepoint dismiss "lcrn:..." --reason "planned decom" +""" + +_EXPLAIN_CHOKEPOINT_RESTORE = """\ +Restore (un-dismiss) a previously dismissed choke point so it +surfaces on the risk overview again. + +Example: + limacharlie cloudsec chokepoint restore "lcrn:..." +""" + +_EXPLAIN_RESOLVE_SENSORS = """\ +Resolve LimaCharlie sensor ids to the cloud asset (URN, with +posture flags) each runs on. Pass any number of SIDs — requests are +chunked automatically to stay within URL limits; unresolved sensors +are returned in 'unresolved'. + +Example: + limacharlie cloudsec resolve sensors +""" + +_EXPLAIN_RESOLVE_ASSETS = """\ +Resolve cloud asset URNs to the LimaCharlie sensor ids running on +each. Pass any number of URNs — requests are chunked automatically +to stay within URL limits; unresolved URNs are returned in +'unresolved'. + +Example: + limacharlie cloudsec resolve assets "lcrn:...instance/web-1" +""" + +_EXPLAIN_CAASM_ASSETS = """\ +The merged third-party asset inventory: every device/identity the +org's connected tools (EDR / IdP / MDM / scanners) report, +entity-resolved to one row per real asset with per-source +provenance in props. + +Examples: + limacharlie cloudsec caasm assets + limacharlie cloudsec caasm assets -q laptop --limit 50 +""" + +_EXPLAIN_CAASM_COVERAGE = """\ +Coverage-gap findings: assets observed by at least one connected +tool but missing a tool the org's expected-coverage policy requires +(e.g. seen by the IdP, no EDR). Same shape as 'finding list' with +the coverage_gap class stamped server-side. + +Examples: + limacharlie cloudsec caasm coverage + limacharlie cloudsec caasm coverage --status open --severity HIGH +""" + +_EXPLAIN_CAASM_POLICY_GET = """\ +Get the stored expected-coverage policy. The response is the +standard resource-list shape: 'resources' holds zero rows (no +policy declared — coverage evaluation is then a no-op) or one row +whose 'props' object is the policy ({expect:[...]}). + +Example: + limacharlie cloudsec caasm policy get +""" + +_EXPLAIN_CAASM_POLICY_SET = """\ +Set (upsert) the expected-coverage policy: the declarative +expectations the coverage engine evaluates over the merged asset +inventory. Validated server-side; an invalid policy is rejected. + +The policy can come from inline JSON, a JSON/YAML file, or stdin. + +Examples: + limacharlie cloudsec caasm policy set \\ + --policy-json '{"expect":[{"label":"edr-on-devices","capability":"edr","kinds":["device"]}]}' + limacharlie cloudsec caasm policy set --input-file policy.yaml + cat policy.yaml | limacharlie cloudsec caasm policy set +""" + +_EXPLAIN_CAASM_INGEST = """\ +Ingest a batch of raw third-party asset records into the merged +asset inventory. --source names the CAASM source the records come +from (today: sentinelone, crowdstrike, defender, okta, entraid, +ms_graph, wiz — the registry grows and is validated server-side); +records are the raw vendor-shaped JSON objects that source ships. +Re-ingesting identical records is a no-op (idempotent). Chunk large +imports — the request body is capped at 1 MiB. + +Examples: + limacharlie cloudsec caasm ingest --source okta --records-file users.json + limacharlie cloudsec caasm ingest --source crowdstrike --record-json '{...}' +""" + +_EXPLAIN_PROVIDER_TEST = """\ +Preflight a cloud provider configuration before saving it: connect +to the provider with the given credentials (ephemeral — never +stored) and probe every permission surface collection needs. +report.ok is the overall verdict over the REQUIRED checks; a failed +optional check flags a gracefully-degraded surface, not a failure. + +The input is a cloudsec_provider hive record shape; 'credentials' +may be inline plaintext or a hive://secret/ reference. Saved +provider configs are managed via the hive: + + limacharlie hive set --hive-name cloudsec_provider --key my-gcp \\ + --input-file provider.json --enabled + +The record can come from inline JSON, a JSON/YAML file, or stdin. + +Examples: + limacharlie cloudsec provider test --input-file provider.yaml + limacharlie cloudsec provider test --provider-json '{"provider_type":"gcp",...}' +""" + +register_explain("cloudsec.overview", _EXPLAIN_OVERVIEW) +register_explain("cloudsec.changes", _EXPLAIN_CHANGES) +register_explain("cloudsec.risk-trend", _EXPLAIN_RISK_TREND) +register_explain("cloudsec.scan-status", _EXPLAIN_SCAN_STATUS) +register_explain("cloudsec.finding.list", _EXPLAIN_FINDING_LIST) +register_explain("cloudsec.finding.facets", _EXPLAIN_FINDING_FACETS) +register_explain("cloudsec.finding.get", _EXPLAIN_FINDING_GET) +register_explain("cloudsec.finding.resolve", _EXPLAIN_FINDING_RESOLVE) +register_explain("cloudsec.finding.bulk-resolve", _EXPLAIN_FINDING_BULK_RESOLVE) +register_explain("cloudsec.finding.set-owner", _EXPLAIN_FINDING_SET_OWNER) +register_explain("cloudsec.finding.set-ticket", _EXPLAIN_FINDING_SET_TICKET) +register_explain("cloudsec.attack-path.list", _EXPLAIN_ATTACK_PATH_LIST) +register_explain("cloudsec.ciem.public-access", _EXPLAIN_CIEM_PUBLIC_ACCESS) +register_explain("cloudsec.ciem.facets", _EXPLAIN_CIEM_FACETS) +register_explain("cloudsec.inventory.list", _EXPLAIN_INVENTORY_LIST) +register_explain("cloudsec.inventory.facets", _EXPLAIN_INVENTORY_FACETS) +register_explain("cloudsec.data-security.facets", _EXPLAIN_DATA_SECURITY_FACETS) +register_explain("cloudsec.resource.get", _EXPLAIN_RESOURCE_GET) +register_explain("cloudsec.graph.neighbors", _EXPLAIN_GRAPH_NEIGHBORS) +register_explain("cloudsec.query.list", _EXPLAIN_QUERY_LIST) +register_explain("cloudsec.query.run", _EXPLAIN_QUERY_RUN) +register_explain("cloudsec.compliance.report", _EXPLAIN_COMPLIANCE_REPORT) +register_explain("cloudsec.compliance.frameworks", _EXPLAIN_COMPLIANCE_FRAMEWORKS) +register_explain("cloudsec.compliance.assignments", _EXPLAIN_COMPLIANCE_ASSIGNMENTS) +register_explain("cloudsec.chokepoint.list", _EXPLAIN_CHOKEPOINT_LIST) +register_explain("cloudsec.chokepoint.dismiss", _EXPLAIN_CHOKEPOINT_DISMISS) +register_explain("cloudsec.chokepoint.restore", _EXPLAIN_CHOKEPOINT_RESTORE) +register_explain("cloudsec.resolve.sensors", _EXPLAIN_RESOLVE_SENSORS) +register_explain("cloudsec.resolve.assets", _EXPLAIN_RESOLVE_ASSETS) +register_explain("cloudsec.caasm.assets", _EXPLAIN_CAASM_ASSETS) +register_explain("cloudsec.caasm.coverage", _EXPLAIN_CAASM_COVERAGE) +register_explain("cloudsec.caasm.policy.get", _EXPLAIN_CAASM_POLICY_GET) +register_explain("cloudsec.caasm.policy.set", _EXPLAIN_CAASM_POLICY_SET) +register_explain("cloudsec.caasm.ingest", _EXPLAIN_CAASM_INGEST) +register_explain("cloudsec.provider.test", _EXPLAIN_PROVIDER_TEST) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _output(ctx: click.Context, data: Any) -> None: + fmt = ctx.obj.output_format or detect_output_format() + if not ctx.obj.quiet: + click.echo(format_output(data, fmt)) + + +def _get_cloudsec(ctx: click.Context) -> CloudSec: + client = Client( + oid=ctx.obj.oid, + environment=ctx.obj.environment, + print_debug_fn=ctx.obj.debug_fn, + debug_full_response=ctx.obj.debug_full, + debug_curl=ctx.obj.debug_curl, + debug_verbose=ctx.obj.debug_verbose, + ) + org = Organization(client) + return CloudSec(org) + + +def _parse_json_opt(value: str | None, param_hint: str) -> Any: + """Parse a JSON CLI option value, raising a clean usage error.""" + if value is None: + return None + try: + return json.loads(value) + except json.JSONDecodeError as exc: + raise click.BadParameter(f"invalid JSON: {exc}", param_hint=param_hint) + + +from ._input_helpers import load_file as _load_file, load_stdin as _load_stdin + + +def _one_of(param_hint: str, **provided: Any) -> None: + """Require exactly one of the given options to be set, and non-empty. + + ``None`` means absent; an explicit empty string is an error (either as + the single provided option, or alongside a real one — both are + ambiguous invocations worth rejecting, not guessing about). + """ + given = [k for k, v in provided.items() if v is not None] + if len(given) != 1: + names = ", ".join(f"--{k.replace('_', '-')}" for k in provided) + raise click.UsageError( + f"provide exactly one of {names} for {param_hint}", + ) + if not provided[given[0]]: + raise click.BadParameter( + "must not be empty", param_hint=f"--{given[0].replace('_', '-')}", + ) + + +def _load_json_object_arg( + *, + inline: str | None, + inline_hint: str, + input_file: str | None, + file_hint: str, + what: str, + shape_msg: str, +) -> dict[str, Any]: + """Load a required JSON-object argument from inline JSON, a file, or stdin. + + At most one of the inline/file options may be set; with neither, piped + stdin is read (YAML or JSON). The result must be an object — a `null` + or non-object input is rejected here rather than reaching the server. + """ + if inline is not None and input_file is not None: + raise click.UsageError( + f"provide only one of {inline_hint} or {file_hint} for {what}", + ) + if inline is not None: + if not inline.strip(): + raise click.BadParameter("must not be empty", param_hint=inline_hint) + value = _parse_json_opt(inline, inline_hint) + hint = inline_hint + elif input_file is not None: + value = _load_file(input_file, file_hint) + hint = file_hint + else: + value = _load_stdin() + hint = "stdin" + if value is None: + raise click.UsageError( + f"provide {inline_hint}, {file_hint}, or pipe {what} via stdin", + ) + if not isinstance(value, dict): + raise click.BadParameter(shape_msg, param_hint=hint) + return value + + +# Resolution kinds are a fixed server contract; 'open' (reopen) is only +# accepted by the single-finding endpoint, not the bulk one. +_KIND_CHOICES = click.Choice( + ["mitigated", "accepted", "false_positive", "open"], case_sensitive=False, +) +_BULK_KIND_CHOICES = click.Choice( + ["mitigated", "accepted", "false_positive"], case_sensitive=False, +) +_ORDER_CHOICES = click.Choice(["asc", "desc"], case_sensitive=False) +# Provider and CAASM-source values are deliberately NOT click.Choice lists: +# both are growing server-side registries (new providers/sources ship without +# a CLI release) and the server rejects unknown values with a clean error. + + +def _paging_options(f): + f = click.option( + "--limit", default=None, type=int, + help="Page size (server clamps to 1000).", + )(f) + f = click.option( + "--cursor", default=None, + help="Keyset-pagination token (next_cursor from the previous page).", + )(f) + return 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 = click.option( + "-q", "--search", "q", default=None, + help="Substring search over the findings.", + )(f) + f = click.option( + "--kev/--no-kev", "kev", default=None, + help="Only findings with (without) a KEV vulnerability.", + )(f) + f = click.option( + "--reachable/--no-reachable", "reachable", default=None, + help="Only findings on (non-)reachable resources.", + )(f) + f = click.option( + "--account", "accounts", multiple=True, + help="Filter by cloud account; repeatable (OR).", + )(f) + f = click.option( + "--status", "statuses", multiple=True, + help="Filter by status (open/resolved); repeatable (OR).", + )(f) + f = click.option( + "--class", "finding_classes", multiple=True, + help="Filter by finding class (e.g. toxic_combination, misconfig); repeatable (OR).", + )(f) + f = click.option( + "--severity", "severities", multiple=True, + help="Filter by severity (CRITICAL/HIGH/MEDIUM/LOW/INFO); repeatable (OR).", + )(f) + return f + + +def _sort_options(f): + f = click.option( + "--order", default=None, type=_ORDER_CHOICES, + help="Sort order: asc or desc.", + )(f) + f = click.option( + "--sort", default=None, + help="Field to sort by (server-side).", + )(f) + return f + + +# --------------------------------------------------------------------------- +# Group +# --------------------------------------------------------------------------- + +@click.group("cloudsec") +def group() -> None: + """Cloud Security (CNAPP): findings, inventory, graph, compliance. + + Requires the org to be subscribed to the ext-cloud-inventory + extension. Reads need the 'cloudsec.get' permission; triage and + other writes need 'cloudsec.set'. Provider configs and policies + are hive records (cloudsec_provider / cloudsec_policy hives). + + \b + Subgroups / commands: + overview Composed risk overview (score, top paths, trend) + changes Recent finding created/closed feed + risk-trend Risk-score history + scan-status Cloud-collection run status per provider + finding ... Findings worklist + triage (resolve, owner, ticket) + attack-path list Headline toxic-combination attack paths + ciem ... Identity access views (public-access, facets) + inventory ... Cloud resource inventory + data-security ... DSPM data-store facets + resource get Canonical record for any urn + graph neighbors 1-hop graph expansion around a urn + query ... Graph queries (list pack, run) + compliance ... Framework assessment, frameworks, assignments + chokepoint ... Estate-wide chokepoints (list, dismiss, restore) + resolve ... Sensor <-> cloud asset resolution + caasm ... Third-party asset inventory, coverage, ingest + provider test Preflight provider credentials before saving + """ + + +# --------------------------------------------------------------------------- +# Top-level reads +# --------------------------------------------------------------------------- + +@group.command("overview") +@click.option("--trend-days", default=None, type=int, help="Days of score trend to include (default 30).") +@pass_context +def overview(ctx, trend_days) -> None: + """Composed risk overview for the org. + + \b + Example: + limacharlie cloudsec overview --trend-days 90 + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.get_overview(trend_days=trend_days)) + + +@group.command("changes") +@click.option("--limit", default=None, type=int, help="Max change events (default 50).") +@pass_context +def changes(ctx, limit) -> None: + """Recent finding lifecycle changes (created/closed), newest first. + + \b + Example: + limacharlie cloudsec changes --limit 100 + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.list_changes(limit=limit)) + + +@group.command("risk-trend") +@click.option("--trend-days", default=None, type=int, help="Days of score trend to include (default 30).") +@pass_context +def risk_trend(ctx, trend_days) -> None: + """The org's risk-score history, oldest first. + + \b + Example: + limacharlie cloudsec risk-trend --trend-days 90 + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.get_risk_trend(trend_days=trend_days)) + + +@group.command("scan-status") +@click.option("--provider", default=None, + help="Cloud provider (e.g. gcp, aws, azure, okta, 1password, " + "google_workspace, cloudflare); validated server-side; default gcp.") +@pass_context +def scan_status(ctx, provider) -> None: + """Cloud-collection run status for a provider. + + \b + Example: + limacharlie cloudsec scan-status --provider aws + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.get_scan_status(provider=provider)) + + +# --------------------------------------------------------------------------- +# finding subgroup +# --------------------------------------------------------------------------- + +@group.group("finding") +def finding_group() -> None: + """Findings worklist: list, facets, get, and triage writes.""" + + +@finding_group.command("list") +@_finding_filter_options +@_sort_options +@_paging_options +@pass_context +def finding_list(ctx, severities, finding_classes, statuses, accounts, + reachable, kev, q, sort, order, cursor, limit) -> None: + """List the merged, risk-ranked cloud-security findings. + + \b + Examples: + limacharlie cloudsec finding list --severity CRITICAL --severity HIGH + limacharlie cloudsec finding list --class public_exposure --kev + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.list_findings( + severity=list(severities) or None, + finding_class=list(finding_classes) or None, + status=list(statuses) or None, + account=list(accounts) or None, + reachable=reachable, + kev=kev, + q=q, + sort=sort, + order=order, + cursor=cursor, + limit=limit, + )) + + +@finding_group.command("facets") +@_finding_filter_options +@pass_context +def finding_facets(ctx, severities, finding_classes, statuses, accounts, + reachable, kev, q) -> None: + """Cross-filtered facet counts for the findings worklist. + + \b + Example: + limacharlie cloudsec finding facets --severity CRITICAL + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.get_finding_facets( + severity=list(severities) or None, + finding_class=list(finding_classes) or None, + status=list(statuses) or None, + account=list(accounts) or None, + reachable=reachable, + kev=kev, + q=q, + )) + + +@finding_group.command("get") +@click.argument("finding_id") +@pass_context +def finding_get(ctx, finding_id) -> None: + """Get a single finding by FINDING_ID. + + \b + Example: + limacharlie cloudsec finding get fnd_0123abcd + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.get_finding(finding_id)) + + +@finding_group.command("resolve") +@click.argument("finding_id") +@click.option("--kind", required=True, type=_KIND_CHOICES, + help="Resolution kind; 'open' reopens the finding.") +@click.option("--reason", default=None, help="Optional operator note.") +@click.option("--expires-at", default=None, type=int, + help="Unix seconds; only meaningful with --kind accepted.") +@pass_context +def finding_resolve(ctx, finding_id, kind, reason, expires_at) -> None: + """Disposition (or reopen, with --kind open) FINDING_ID. + + \b + Examples: + limacharlie cloudsec finding resolve fnd_abc --kind mitigated + limacharlie cloudsec finding resolve fnd_abc --kind open + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.set_finding_status( + finding_id, kind, reason=reason, expires_at=expires_at, + )) + + +@finding_group.command("bulk-resolve") +@click.option("--finding-id", "finding_ids", multiple=True, required=True, + help="Finding id; repeat for each finding.") +@click.option("--kind", required=True, type=_BULK_KIND_CHOICES, + help="Resolution kind (reopen is single-finding only: use 'finding resolve --kind open').") +@click.option("--reason", default=None, help="Optional operator note.") +@click.option("--expires-at", default=None, type=int, + help="Unix seconds; only meaningful with --kind accepted.") +@pass_context +def finding_bulk_resolve(ctx, finding_ids, kind, reason, expires_at) -> None: + """Apply one resolution to many findings. + + \b + Example: + limacharlie cloudsec finding bulk-resolve --finding-id fnd_a --finding-id fnd_b --kind mitigated + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.bulk_set_finding_status( + list(finding_ids), kind, reason=reason, expires_at=expires_at, + )) + + +@finding_group.command("set-owner") +@click.argument("finding_id") +@click.option("--owner", default=None, help="The owner to assign.") +@click.option("--clear", is_flag=True, default=False, help="Clear the owner.") +@pass_context +def finding_set_owner(ctx, finding_id, owner, clear) -> None: + """Assign (or clear, with --clear) the owner of FINDING_ID. + + \b + Example: + limacharlie cloudsec finding set-owner fnd_abc --owner alice@corp.com + """ + if clear == (owner is not None): + raise click.UsageError("provide exactly one of --owner or --clear") + cs = _get_cloudsec(ctx) + _output(ctx, cs.set_finding_owner(finding_id, "" if clear else owner)) + + +@finding_group.command("set-ticket") +@click.argument("finding_id") +@click.option("--ticket", default=None, help="The ticket id/url to link.") +@click.option("--clear", is_flag=True, default=False, help="Clear the ticket.") +@pass_context +def finding_set_ticket(ctx, finding_id, ticket, clear) -> None: + """Link (or clear, with --clear) a ticket to FINDING_ID. + + \b + Example: + limacharlie cloudsec finding set-ticket fnd_abc --ticket JIRA-123 + """ + if clear == (ticket is not None): + raise click.UsageError("provide exactly one of --ticket or --clear") + cs = _get_cloudsec(ctx) + _output(ctx, cs.set_finding_ticket(finding_id, "" if clear else ticket)) + + +# --------------------------------------------------------------------------- +# attack-path subgroup +# --------------------------------------------------------------------------- + +@group.group("attack-path") +def attack_path_group() -> None: + """Headline toxic-combination attack paths.""" + + +@attack_path_group.command("list") +@click.option("--severity", "severities", multiple=True, help="Filter by severity; repeatable (OR).") +@click.option("--account", "accounts", multiple=True, help="Filter by cloud account; repeatable (OR).") +@click.option("--status", "statuses", multiple=True, help="Filter by status; repeatable (OR).") +@click.option("-q", "--search", "q", default=None, help="Substring search.") +@pass_context +def attack_path_list(ctx, severities, accounts, statuses, q) -> None: + """List the headline toxic-combination attack paths. + + \b + Example: + limacharlie cloudsec attack-path list --severity CRITICAL + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.list_attack_paths( + severity=list(severities) or None, + account=list(accounts) or None, + status=list(statuses) or None, + q=q, + )) + + +# --------------------------------------------------------------------------- +# ciem subgroup +# --------------------------------------------------------------------------- + +@group.group("ciem") +def ciem_group() -> None: + """CIEM identity access views.""" + + +@ciem_group.command("public-access") +@pass_context +def ciem_public_access(ctx) -> None: + """Public/external access to sensitive resources. + + \b + Example: + limacharlie cloudsec ciem public-access + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.get_public_access()) + + +@ciem_group.command("facets") +@pass_context +def ciem_facets(ctx) -> None: + """CIEM identity facet counts. + + \b + Example: + limacharlie cloudsec ciem facets + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.get_identity_facets()) + + +# --------------------------------------------------------------------------- +# inventory subgroup +# --------------------------------------------------------------------------- + +@group.group("inventory") +def inventory_group() -> None: + """Cloud resource inventory.""" + + +@inventory_group.command("list") +@click.option("--type", "resource_type", default=None, help="Filter by resource type.") +@click.option("--account", default=None, help="Filter by cloud account.") +@click.option("--region", default=None, help="Filter by region.") +@click.option("-q", "--search", "q", default=None, help="Substring search.") +@_paging_options +@pass_context +def inventory_list(ctx, resource_type, account, region, q, cursor, limit) -> None: + """List the cloud resource inventory. + + \b + Examples: + limacharlie cloudsec inventory list --type gcp_bucket + limacharlie cloudsec inventory list -q prod --limit 50 + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.list_inventory( + resource_type=resource_type, account=account, region=region, + q=q, cursor=cursor, limit=limit, + )) + + +@inventory_group.command("facets") +@pass_context +def inventory_facets(ctx) -> None: + """Inventory facet counts by type/account/region. + + \b + Example: + limacharlie cloudsec inventory facets + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.get_inventory_facets()) + + +# --------------------------------------------------------------------------- +# data-security subgroup +# --------------------------------------------------------------------------- + +@group.group("data-security") +def data_security_group() -> None: + """DSPM data-store views.""" + + +@data_security_group.command("facets") +@pass_context +def data_security_facets(ctx) -> None: + """DSPM data-store facet counts. + + \b + Example: + limacharlie cloudsec data-security facets + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.get_data_security_facets()) + + +# --------------------------------------------------------------------------- +# resource subgroup +# --------------------------------------------------------------------------- + +@group.group("resource") +def resource_group() -> None: + """Point lookups for canonical resource records.""" + + +@resource_group.command("get") +@click.argument("urn") +@pass_context +def resource_get(ctx, urn) -> None: + """Get the canonical record for URN (null when unknown). + + \b + Example: + limacharlie cloudsec resource get "lcrn:gcp:...:bucket/prod-data" + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.get_resource(urn)) + + +# --------------------------------------------------------------------------- +# graph subgroup +# --------------------------------------------------------------------------- + +@group.group("graph") +def graph_group() -> None: + """Security graph expansion.""" + + +@graph_group.command("neighbors") +@click.argument("urn") +@click.option("--limit", default=None, type=int, help="Max neighbors (default 200, cap 500).") +@pass_context +def graph_neighbors(ctx, urn, limit) -> None: + """Expand URN's 1-hop neighborhood in the security graph. + + \b + Example: + limacharlie cloudsec graph neighbors "lcrn:...instance/web-1" --limit 500 + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.get_graph_neighbors(urn, limit=limit)) + + +# --------------------------------------------------------------------------- +# query subgroup +# --------------------------------------------------------------------------- + +@group.group("query") +def query_group() -> None: + """Graph queries: list the query pack, run a query.""" + + +@query_group.command("list") +@pass_context +def query_list(ctx) -> None: + """List the named graph queries in the query pack. + + \b + Example: + limacharlie cloudsec query list + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.list_queries()) + + +@query_group.command("run") +@click.option("--named", default=None, help="A query-pack name (see 'query list').") +@click.option("--text", default=None, help="A text query.") +@click.option("--query-json", default=None, help="A raw query DSL object as JSON.") +@click.option("--project", default=None, + help="Comma-separated aliases to project into the rows.") +@pass_context +def query_run(ctx, named, text, query_json, project) -> None: + """Run a graph query (one of --named / --text / --query-json). + + \b + Examples: + limacharlie cloudsec query run --named public-buckets + limacharlie cloudsec query run --text "public bucket with sensitive data" + """ + _one_of("the query", named=named, text=text, query_json=query_json) + query = None + if query_json: + query = _parse_json_opt(query_json, "--query-json") + # Unconditional: a JSON `null` must not slip through as "no query". + if not isinstance(query, dict): + raise click.BadParameter("must decode to a JSON object", param_hint="--query-json") + project_list = [p.strip() for p in project.split(",") if p.strip()] if project else None + cs = _get_cloudsec(ctx) + _output(ctx, cs.run_query( + named=named, text=text, query=query, project=project_list, + )) + + +# --------------------------------------------------------------------------- +# compliance subgroup +# --------------------------------------------------------------------------- + +@group.group("compliance") +def compliance_group() -> None: + """Compliance assessment: report, frameworks, assignments.""" + + +@compliance_group.command("report") +@click.option("--framework", default=None, + help="Framework id (default cis-gcp); ignored when --assignment is set.") +@click.option("--assignment", default=None, + help="Named scoped assignment to evaluate instead of the whole estate.") +@pass_context +def compliance_report(ctx, framework, assignment) -> None: + """Per-control pass/fail compliance assessment. + + \b + Examples: + limacharlie cloudsec compliance report --framework cis-aws + limacharlie cloudsec compliance report --assignment prod-scope + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.get_compliance(framework=framework, assignment=assignment)) + + +@compliance_group.command("frameworks") +@pass_context +def compliance_frameworks(ctx) -> None: + """List the selectable compliance frameworks. + + \b + Example: + limacharlie cloudsec compliance frameworks + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.list_compliance_frameworks()) + + +@compliance_group.command("assignments") +@pass_context +def compliance_assignments(ctx) -> None: + """List the org's scoped compliance assignments. + + \b + Example: + limacharlie cloudsec compliance assignments + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.list_compliance_assignments()) + + +# --------------------------------------------------------------------------- +# chokepoint subgroup +# --------------------------------------------------------------------------- + +@group.group("chokepoint") +def chokepoint_group() -> None: + """Estate-wide chokepoints: list, dismiss, restore.""" + + +@chokepoint_group.command("list") +@pass_context +def chokepoint_list(ctx) -> None: + """Estate-wide chokepoints ranked by attack paths broken. + + \b + Example: + limacharlie cloudsec chokepoint list + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.list_chokepoints()) + + +@chokepoint_group.command("dismiss") +@click.argument("urn") +@click.option("--reason", default=None, help="Optional reason recorded with the dismissal.") +@pass_context +def chokepoint_dismiss(ctx, urn, reason) -> None: + """Dismiss the choke point at URN from the risk overview. + + \b + Example: + limacharlie cloudsec chokepoint dismiss "lcrn:..." --reason "planned decom" + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.dismiss_chokepoint(urn, reason=reason)) + + +@chokepoint_group.command("restore") +@click.argument("urn") +@pass_context +def chokepoint_restore(ctx, urn) -> None: + """Restore (un-dismiss) the choke point at URN. + + \b + Example: + limacharlie cloudsec chokepoint restore "lcrn:..." + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.restore_chokepoint(urn)) + + +# --------------------------------------------------------------------------- +# resolve subgroup +# --------------------------------------------------------------------------- + +@group.group("resolve") +def resolve_group() -> None: + """Sensor <-> cloud asset resolution (fusion mapping).""" + + +@resolve_group.command("sensors") +@click.argument("sids", nargs=-1, required=True) +@pass_context +def resolve_sensors(ctx, sids) -> None: + """Resolve SIDS to the cloud asset each sensor runs on. + + Any number of SIDs — requests are chunked automatically. + + \b + Example: + limacharlie cloudsec resolve sensors + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.resolve_sensors(list(sids))) + + +@resolve_group.command("assets") +@click.argument("urns", nargs=-1, required=True) +@pass_context +def resolve_assets(ctx, urns) -> None: + """Resolve asset URNS to the sensors running on each. + + Any number of URNs — requests are chunked automatically. + + \b + Example: + limacharlie cloudsec resolve assets "lcrn:...instance/web-1" + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.resolve_assets(list(urns))) + + +# --------------------------------------------------------------------------- +# caasm subgroup +# --------------------------------------------------------------------------- + +@group.group("caasm") +def caasm_group() -> None: + """CAASM: third-party asset inventory, coverage gaps, policy, ingest.""" + + +@caasm_group.command("assets") +@click.option("-q", "--search", "q", default=None, help="Substring filter over asset urn/name.") +@_paging_options +@pass_context +def caasm_assets(ctx, q, cursor, limit) -> None: + """The merged third-party asset inventory. + + \b + Example: + limacharlie cloudsec caasm assets -q laptop --limit 50 + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.list_caasm_assets(q=q, cursor=cursor, limit=limit)) + + +@caasm_group.command("coverage") +@click.option("--status", "statuses", multiple=True, help="Filter by finding status; repeatable (OR).") +@click.option("--severity", "severities", multiple=True, help="Filter by severity; repeatable (OR).") +@click.option("-q", "--search", "q", default=None, help="Substring search.") +@_sort_options +@_paging_options +@pass_context +def caasm_coverage(ctx, statuses, severities, q, sort, order, cursor, limit) -> None: + """Coverage-gap findings (assets missing a required tool). + + \b + Example: + limacharlie cloudsec caasm coverage --status open --severity HIGH + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.list_caasm_coverage( + status=list(statuses) or None, + severity=list(severities) or None, + q=q, sort=sort, order=order, cursor=cursor, limit=limit, + )) + + +@caasm_group.group("policy") +def caasm_policy_group() -> None: + """The expected-coverage policy.""" + + +@caasm_policy_group.command("get") +@pass_context +def caasm_policy_get(ctx) -> None: + """Get the stored expected-coverage policy. + + \b + Example: + limacharlie cloudsec caasm policy get + """ + cs = _get_cloudsec(ctx) + _output(ctx, cs.get_caasm_policy()) + + +@caasm_policy_group.command("set") +@click.option("--policy-json", default=None, help="The policy as inline JSON ({expect:[...]}).") +@click.option("--input-file", default=None, type=click.Path(exists=True, dir_okay=False), + help="Read the policy from a JSON or YAML file.") +@pass_context +def caasm_policy_set(ctx, policy_json, input_file) -> None: + """Set (upsert) the expected-coverage policy. + + The policy can also be piped via stdin (JSON or YAML). + + \b + Example: + limacharlie cloudsec caasm policy set --input-file policy.yaml + """ + policy = _load_json_object_arg( + inline=policy_json, inline_hint="--policy-json", + input_file=input_file, file_hint="--input-file", + what="the policy", + shape_msg="must decode to a JSON object ({expect:[...]})", + ) + cs = _get_cloudsec(ctx) + _output(ctx, cs.set_caasm_policy(policy)) + + +@caasm_group.command("ingest") +@click.option("--source", required=True, + help="The CAASM source the records come from (e.g. sentinelone, " + "crowdstrike, defender, okta, entraid, ms_graph, wiz); " + "validated server-side.") +@click.option("--records-file", default=None, type=click.Path(exists=True, dir_okay=False), + help="JSON or YAML file holding an array of raw vendor-shaped records.") +@click.option("--record-json", default=None, help="A single raw record as inline JSON.") +@click.option("--policy-json", default=None, help="Optional inline coverage policy override (JSON).") +@pass_context +def caasm_ingest(ctx, source, records_file, record_json, policy_json) -> None: + """Ingest raw third-party asset records into the merged inventory. + + \b + Examples: + limacharlie cloudsec caasm ingest --source okta --records-file users.json + limacharlie cloudsec caasm ingest --source crowdstrike --record-json '{...}' + """ + _one_of("the records", records_file=records_file, record_json=record_json) + records = None + record = None + if records_file: + records = _load_file(records_file, "--records-file") + # Unconditional: a file whose content is `null` (or any non-array) + # must not slip through as a silent records-less ingest. + if not isinstance(records, list): + raise click.BadParameter( + "must decode to a JSON array of records", param_hint="--records-file", + ) + else: + record = _parse_json_opt(record_json, "--record-json") + if not isinstance(record, dict): + raise click.BadParameter("must decode to a JSON object", param_hint="--record-json") + policy = None + if policy_json: + policy = _parse_json_opt(policy_json, "--policy-json") + if not isinstance(policy, dict): + raise click.BadParameter("must decode to a JSON object", param_hint="--policy-json") + cs = _get_cloudsec(ctx) + _output(ctx, cs.caasm_ingest( + source, records=records, record=record, policy=policy, + )) + + +# --------------------------------------------------------------------------- +# provider subgroup +# --------------------------------------------------------------------------- + +@group.group("provider") +def provider_group() -> None: + """Provider credential preflight (configs live in the cloudsec_provider hive).""" + + +@provider_group.command("test") +@click.option("--provider-json", default=None, + help="The provider record (cloudsec_provider hive shape) as inline JSON.") +@click.option("--input-file", default=None, type=click.Path(exists=True, dir_okay=False), + help="Read the provider record from a JSON or YAML file.") +@pass_context +def provider_test(ctx, provider_json, input_file) -> None: + """Preflight a provider configuration (credentials are never stored). + + The record can also be piped via stdin (JSON or YAML). + + \b + Example: + limacharlie cloudsec provider test --input-file provider.yaml + """ + provider = _load_json_object_arg( + inline=provider_json, inline_hint="--provider-json", + input_file=input_file, file_hint="--input-file", + what="the provider record", + shape_msg="must decode to a JSON object (cloudsec_provider record shape)", + ) + cs = _get_cloudsec(ctx) + _output(ctx, cs.test_provider(provider)) diff --git a/limacharlie/sdk/cloudsec.py b/limacharlie/sdk/cloudsec.py new file mode 100644 index 00000000..c4a6d256 --- /dev/null +++ b/limacharlie/sdk/cloudsec.py @@ -0,0 +1,693 @@ +"""Cloud Security (CNAPP) SDK for LimaCharlie v2. + +Wraps the ``/cloudsec/{oid}/...`` REST routes served by the API +gateway: the merged, risk-ranked findings worklist (CSPM + attack +paths + CIEM), the resource inventory and security graph, compliance +assessment, the risk overview, CAASM (third-party asset attack +surface), sensor<->cloud-asset resolution, and the finding triage +writes. + +Reads require the ``cloudsec.get`` permission and writes require +``cloudsec.set``; every route additionally requires the org to be +subscribed to the ``ext-cloud-inventory`` extension (403 otherwise). + +Provider credentials/config and the cloudsec policies are hive +records (``cloudsec_provider``, ``cloudsec_policy``, ``cloudsec_query`` +hives) managed through the standard Hive API; the one provider +operation here is the pre-save credential preflight +(:meth:`CloudSec.test_provider`). +""" + +from __future__ import annotations + +import json +from typing import Any, TYPE_CHECKING + +if TYPE_CHECKING: + from .organization import Organization + + +def _add_pairs( + pairs: list[tuple[str, str]], + key: str, + values: list[str] | tuple[str, ...] | None, +) -> None: + """Append one ``(key, value)`` pair per value (repeatable query param).""" + if not values: + return + for v in values: + pairs.append((key, str(v))) + + +def _add_scalar( + pairs: list[tuple[str, str]], + key: str, + value: Any, +) -> None: + """Append a single ``(key, value)`` pair when the caller set a value. + + Booleans are lowered to ``true``/``false`` (the gateway parses them + with ``strconv.ParseBool``). + """ + if value is None: + return + if isinstance(value, bool): + pairs.append((key, "true" if value else "false")) + else: + pairs.append((key, str(value))) + + +def _query_pairs(**params: Any) -> list[tuple[str, str]]: + """Build the query-pair list from keyword selectors, skipping unset keys. + + List/tuple values become repeated keys (OR within a key, AND across + keys, matching the gateway contract); scalars go through + :func:`_add_scalar`. Kwarg order is preserved so the emitted query + string is deterministic. + """ + pairs: list[tuple[str, str]] = [] + for key, value in params.items(): + if isinstance(value, (list, tuple)): + _add_pairs(pairs, key, value) + else: + _add_scalar(pairs, key, value) + return pairs + + +def _finding_query_pairs( + *, + severity: list[str] | None = None, + finding_class: list[str] | None = None, + status: list[str] | None = None, + account: list[str] | None = None, + reachable: bool | None = None, + kev: bool | None = None, + q: str | None = None, + sort: str | None = None, + order: str | None = None, + cursor: str | None = None, + limit: int | None = None, +) -> list[tuple[str, str]]: + """Assemble the findings worklist selectors shared by list/facets.""" + return _query_pairs( + severity=severity, finding_class=finding_class, status=status, + account=account, reachable=reachable, kev=kev, q=q, + sort=sort, order=order, cursor=cursor, limit=limit, + ) + + +# Chunk size for the bulk sensor<->asset resolution GETs: ids ride as repeated +# query params and the platform load balancer caps URLs at ~8KB, so one request +# can only carry ~190 UUIDs. 100 per request (~4KB) leaves comfortable headroom; +# the gateway's own per-request cap is 500. +_RESOLVE_CHUNK_SIZE = 100 + + +class CloudSec: + """Cloud Security (CNAPP) client for LimaCharlie.""" + + def __init__(self, org: Organization) -> None: + self._org = org + + @property + def oid(self) -> str: + return self._org.oid + + # ------------------------------------------------------------------ + # Low-level helpers + # ------------------------------------------------------------------ + + def _get( + self, + path: str, + query_params: list[tuple[str, str]] | None = None, + ) -> dict[str, Any]: + return self._org.client.request( + "GET", + f"cloudsec/{self.oid}/{path}", + query_params=query_params or None, + ) + + def _post(self, path: str, body: dict[str, Any]) -> dict[str, Any]: + return self._org.client.request( + "POST", + f"cloudsec/{self.oid}/{path}", + raw_body=json.dumps(body).encode(), + content_type="application/json", + ) + + # ------------------------------------------------------------------ + # Findings worklist + # ------------------------------------------------------------------ + + def list_findings( + self, + *, + severity: list[str] | None = None, + finding_class: list[str] | None = None, + status: list[str] | None = None, + account: list[str] | None = None, + reachable: bool | None = None, + kev: bool | None = None, + q: str | None = None, + sort: str | None = None, + order: str | None = None, + cursor: str | None = None, + limit: int | None = None, + ) -> dict[str, Any]: + """List the merged, risk-ranked cloud-security findings. + + Args: + severity: Filter values (CRITICAL/HIGH/MEDIUM/LOW/INFO), OR'd. + finding_class: Filter values (toxic_combination, public_exposure, + ciem_risk, privilege_escalation, vulnerability, misconfig, + malware, secret, scan_finding, coverage_gap), OR'd. + status: Filter values (open/resolved), OR'd. + account: Cloud account filter values, OR'd. + reachable: Only findings on (non-)reachable resources. + kev: Only findings with (without) a KEV vulnerability. + q: Substring search. + sort, order: Server-side ordering selectors. + cursor: Keyset-pagination token from a previous page. + limit: Page size (server clamps to 1000). + + Returns: + ``{"findings": [...], "next_cursor": str}``. + """ + return self._get("findings", _finding_query_pairs( + severity=severity, finding_class=finding_class, status=status, + account=account, reachable=reachable, kev=kev, q=q, + sort=sort, order=order, cursor=cursor, limit=limit, + )) + + def get_finding_facets( + self, + *, + severity: list[str] | None = None, + finding_class: list[str] | None = None, + status: list[str] | None = None, + account: list[str] | None = None, + reachable: bool | None = None, + kev: bool | None = None, + q: str | None = None, + ) -> dict[str, Any]: + """Cross-filtered facet counts for the findings worklist. + + Takes the same filter selectors as :meth:`list_findings`; each + facet dimension is counted against the other active filters. + + Returns: + ``{"facets": {...}}``. + """ + return self._get("findings/facets", _finding_query_pairs( + severity=severity, finding_class=finding_class, status=status, + account=account, reachable=reachable, kev=kev, q=q, + )) + + def get_finding(self, finding_id: str) -> dict[str, Any]: + """Get one finding by id (e.g. ``fnd_``). + + Returns: + ``{"finding": {...}}``. + """ + return self._get(f"findings/{finding_id}") + + # ------------------------------------------------------------------ + # Finding triage writes (cloudsec.set) + # ------------------------------------------------------------------ + + def set_finding_status( + self, + finding_id: str, + kind: str, + *, + reason: str | None = None, + expires_at: int | None = None, + ) -> dict[str, Any]: + """Disposition (or reopen) a finding. + + Args: + finding_id: The finding to disposition. + kind: ``mitigated``, ``accepted``, ``false_positive``, or + ``open`` to clear the disposition and reopen the finding + (owner/ticket are kept). + reason: Optional operator note. + expires_at: Unix seconds; only meaningful for ``accepted``. + + Returns: + ``{"ok": bool}``. + """ + resolution: dict[str, Any] = {"kind": kind} + if reason is not None: + resolution["reason"] = reason + if expires_at is not None: + resolution["expires_at"] = expires_at + return self._post( + f"findings/{finding_id}/status", {"resolution": resolution}, + ) + + def bulk_set_finding_status( + self, + finding_ids: list[str], + kind: str, + *, + reason: str | None = None, + expires_at: int | None = None, + ) -> dict[str, Any]: + """Apply one resolution to many findings at once. + + ``kind`` must be ``mitigated``, ``accepted``, or + ``false_positive`` — unlike :meth:`set_finding_status`, the bulk + endpoint does NOT accept ``open`` (reopen findings one at a + time). + + Returns: + ``{"updated": int}``. + """ + resolution: dict[str, Any] = {"kind": kind} + if reason is not None: + resolution["reason"] = reason + if expires_at is not None: + resolution["expires_at"] = expires_at + return self._post("findings/bulk/status", { + "finding_ids": list(finding_ids), + "resolution": resolution, + }) + + def set_finding_owner(self, finding_id: str, owner: str) -> dict[str, Any]: + """Assign (or clear, with an empty string) the owner of a finding. + + Returns: + ``{"ok": bool}``. + """ + return self._post(f"findings/{finding_id}/owner", {"owner": owner}) + + def set_finding_ticket(self, finding_id: str, ticket: str) -> dict[str, Any]: + """Link (or clear, with an empty string) a ticket id/url to a finding. + + Returns: + ``{"ok": bool}``. + """ + return self._post(f"findings/{finding_id}/ticket", {"ticket": ticket}) + + # ------------------------------------------------------------------ + # Attack paths / CIEM + # ------------------------------------------------------------------ + + def list_attack_paths( + self, + *, + severity: list[str] | None = None, + account: list[str] | None = None, + status: list[str] | None = None, + q: str | None = None, + ) -> dict[str, Any]: + """Headline toxic-combination attack paths. + + Returns: + ``{"paths": [...]}``. + """ + return self._get("attack-paths", _query_pairs( + severity=severity, account=account, status=status, q=q, + )) + + def get_public_access(self) -> dict[str, Any]: + """CIEM: public/external access to sensitive resources. + + Returns: + ``{"access": [...]}``. + """ + return self._get("ciem/public-access") + + def get_identity_facets(self) -> dict[str, Any]: + """CIEM identity facet counts. + + Returns: + ``{"facets": {...}}``. + """ + return self._get("ciem/facets") + + # ------------------------------------------------------------------ + # Inventory / resources / data security + # ------------------------------------------------------------------ + + def list_inventory( + self, + *, + resource_type: str | None = None, + account: str | None = None, + region: str | None = None, + q: str | None = None, + cursor: str | None = None, + limit: int | None = None, + ) -> dict[str, Any]: + """List the cloud resource inventory. + + Args: + resource_type: Filter by resource type (the ``type`` selector). + account, region: Scalar filters. + q: Substring search. + cursor, limit: Keyset pagination. + + Returns: + ``{"resources": [...], "next_cursor": str}``. + """ + return self._get("inventory", _query_pairs( + type=resource_type, account=account, region=region, + q=q, 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_data_security_facets(self) -> dict[str, Any]: + """DSPM data-store facet counts (total/sensitive/public, store kinds). + + Returns: + ``{"facets": {...}}``. + """ + return self._get("data-security/facets") + + def get_resource(self, urn: str) -> dict[str, Any]: + """Get the canonical record for any urn the graph knows. + + Returns: + ``{"resource": {...}}`` or ``{"resource": null}`` when unknown. + """ + return self._get("resource", _query_pairs(urn=urn)) + + # ------------------------------------------------------------------ + # Security graph + # ------------------------------------------------------------------ + + def get_graph_neighbors( + self, urn: str, *, limit: int | None = None, + ) -> dict[str, Any]: + """Expand a resource's 1-hop neighborhood in the security graph. + + Args: + urn: The anchor resource. + limit: Max neighbors (default 200, hard cap 500). + + Returns: + ``{"graph": {"nodes": [...], "edges": [...]}}`` with ``truncated``. + """ + return self._get("graph/neighbors", _query_pairs(urn=urn, limit=limit)) + + def list_queries(self) -> dict[str, Any]: + """List the named graph queries in the query pack. + + Returns: + ``{"queries": [{"name","title","description","query"}, ...]}``. + """ + return self._get("queries") + + def run_query( + self, + *, + named: str | None = None, + text: str | None = None, + query: dict[str, Any] | None = None, + project: list[str] | None = None, + ) -> dict[str, Any]: + """Run a graph query. Provide exactly one of named / text / query. + + Args: + named: A query-pack name (see :meth:`list_queries`). + text: A text query. + query: A raw DSL object. + project: Optional aliases to project into the rows. + + Returns: + ``{"rows": [{alias: urn, ...}, ...]}``. + """ + body: dict[str, Any] = {} + if named is not None: + body["named"] = named + if text is not None: + body["text"] = text + if query is not None: + body["query"] = query + if project is not None: + body["project"] = project + return self._post("query", body) + + # ------------------------------------------------------------------ + # Compliance + # ------------------------------------------------------------------ + + def get_compliance( + self, + *, + framework: str | None = None, + assignment: str | None = None, + ) -> dict[str, Any]: + """Per-control pass/fail compliance assessment. + + Args: + framework: Framework id (default cis-gcp server-side); + ignored when ``assignment`` is set. + assignment: Named scoped assignment to evaluate instead. + + Returns: + ``{"report": {...}}``. + """ + return self._get("compliance", _query_pairs( + framework=framework, assignment=assignment, + )) + + def list_compliance_frameworks(self) -> dict[str, Any]: + """List selectable compliance frameworks. + + Returns: + ``{"frameworks": [{"id","name","version","control_count"}, ...]}``. + """ + return self._get("compliance/frameworks") + + def list_compliance_assignments(self) -> dict[str, Any]: + """List the org's scoped compliance assignments (with scores).""" + return self._get("compliance/assignments") + + # ------------------------------------------------------------------ + # Overview / trends / chokepoints + # ------------------------------------------------------------------ + + def get_overview(self, *, trend_days: int | None = None) -> dict[str, Any]: + """Composed risk overview (score, severity distribution, top paths, + coverage, trend, recent changes) in one round-trip.""" + return self._get("overview", _query_pairs(trend_days=trend_days)) + + def list_chokepoints(self) -> dict[str, Any]: + """Estate-wide chokepoints ranked by attack paths broken. + + Returns: + ``{"chokepoints": [...], "total_paths": int}``. + """ + return self._get("chokepoints") + + def dismiss_chokepoint( + self, urn: str, *, reason: str | None = None, + ) -> dict[str, Any]: + """Dismiss an estate-wide choke point from the risk overview. + + Returns: + ``{"ok": bool}``. + """ + body: dict[str, Any] = {"urn": urn} + if reason is not None: + body["reason"] = reason + return self._post("chokepoints/dismiss", body) + + def restore_chokepoint(self, urn: str) -> dict[str, Any]: + """Restore (un-dismiss) a previously dismissed choke point.""" + return self._post("chokepoints/restore", {"urn": urn}) + + def list_changes(self, *, limit: int | None = None) -> dict[str, Any]: + """Recent finding lifecycle changes (created/closed), newest first.""" + return self._get("changes", _query_pairs(limit=limit)) + + def get_risk_trend(self, *, trend_days: int | None = None) -> dict[str, Any]: + """The org risk-score history, oldest first.""" + return self._get("risk-trend", _query_pairs(trend_days=trend_days)) + + def get_scan_status(self, *, provider: str | None = None) -> dict[str, Any]: + """Cloud-collection run status for a provider. + + Args: + provider: Provider id (e.g. ``gcp`` — the server default — + ``aws``, ``azure``, ``okta``, ...; validated server-side). + Lowered before sending: the backend scan-state lookup is a + case-sensitive read keyed on lowercase provider ids, so a + raw ``"AWS"`` would silently read as never-scanned. + + Returns: + ``{"status": {...}}``. + """ + if provider is not None: + provider = provider.strip().lower() + return self._get("scan-status", _query_pairs(provider=provider)) + + # ------------------------------------------------------------------ + # Sensor <-> cloud asset resolution + # ------------------------------------------------------------------ + + def _resolve_chunked( + self, path: str, key: str, values: list[str], + ) -> dict[str, Any]: + """Run a bulk resolve as URL-safe chunks and merge the responses. + + The ids ride as repeated query params, so an unbounded batch would + blow the ~8KB load-balancer URL limit long before the gateway's + 500-per-request cap — chunking makes any batch size work. + """ + resolved: list[Any] = [] + unresolved: list[Any] = [] + values = list(values) + for i in range(0, len(values), _RESOLVE_CHUNK_SIZE): + chunk = values[i:i + _RESOLVE_CHUNK_SIZE] + resp = self._get(path, _query_pairs(**{key: chunk})) + resolved.extend(resp.get("resolved") or []) + unresolved.extend(resp.get("unresolved") or []) + return {"resolved": resolved, "unresolved": unresolved} + + def resolve_sensors(self, sids: list[str]) -> dict[str, Any]: + """Resolve sensor ids to the cloud asset each runs on. + + Any batch size works — requests are chunked (100 ids each) to + stay within URL limits, and the per-chunk responses are merged. + + Returns: + ``{"resolved": [...], "unresolved": [...]}``. + """ + return self._resolve_chunked("resolve/sensors", "sid", sids) + + def resolve_assets(self, urns: list[str]) -> dict[str, Any]: + """Resolve cloud asset URNs to the sensors running on each. + + Any batch size works — requests are chunked (100 URNs each) to + stay within URL limits, and the per-chunk responses are merged. + + Returns: + ``{"resolved": [...], "unresolved": [...]}``. + """ + return self._resolve_chunked("resolve/assets", "urn", urns) + + # ------------------------------------------------------------------ + # CAASM (third-party asset attack surface) + # ------------------------------------------------------------------ + + def list_caasm_assets( + self, + *, + q: str | None = None, + cursor: str | None = None, + limit: int | None = None, + ) -> dict[str, Any]: + """The merged third-party asset inventory (EDR/IdP/MDM/scanner sources). + + Returns: + ``{"resources": [...], "next_cursor": str}``. + """ + return self._get("caasm/assets", _query_pairs( + q=q, cursor=cursor, limit=limit, + )) + + def list_caasm_coverage( + self, + *, + status: list[str] | None = None, + severity: list[str] | None = None, + q: str | None = None, + sort: str | None = None, + order: str | None = None, + cursor: str | None = None, + limit: int | None = None, + ) -> dict[str, Any]: + """Coverage-gap findings (assets missing a required tool). + + Same shape as :meth:`list_findings` with the ``coverage_gap`` + class stamped server-side. + + Returns: + ``{"findings": [...], "next_cursor": str}``. + """ + return self._get("caasm/coverage", _query_pairs( + status=status, severity=severity, q=q, sort=sort, + order=order, cursor=cursor, limit=limit, + )) + + def get_caasm_policy(self) -> dict[str, Any]: + """The stored expected-coverage policy. + + Returns: + The standard resource-list shape: ``resources`` holds zero + rows (no policy declared) or one row whose ``props`` object + is the policy (``{"expect": [...]}``). + """ + return self._get("caasm/policy") + + def set_caasm_policy(self, policy: dict[str, Any]) -> dict[str, Any]: + """Set (upsert) the expected-coverage policy. + + Args: + policy: e.g. ``{"expect": [{"label": "edr-on-devices", + "capability": "edr", "kinds": ["device"]}]}``. Validated + server-side; an invalid policy is rejected loudly. + + Returns: + ``{"ok": bool}``. + """ + return self._post("caasm/policy", {"policy": policy}) + + def caasm_ingest( + self, + source: str, + *, + records: list[dict[str, Any]] | None = None, + record: dict[str, Any] | None = None, + policy: dict[str, Any] | None = None, + ) -> dict[str, Any]: + """Ingest raw third-party asset records into the merged inventory. + + Args: + source: One of sentinelone|crowdstrike|defender|okta|entraid| + ms_graph|wiz. + records: Raw vendor-shaped JSON objects (batch). Chunk large + imports — the request body is capped at 1 MiB. + record: A single object (alternative to ``records``). + policy: Optional inline coverage policy override. + + Returns: + ``{"result": {"received","normalized","skipped","assets", + "created","updated","deleted"}}``. + """ + body: dict[str, Any] = {"source": source} + if records is not None: + body["records"] = records + if record is not None: + body["record"] = record + if policy is not None: + body["policy"] = policy + return self._post("caasm/ingest", body) + + # ------------------------------------------------------------------ + # Provider preflight + # ------------------------------------------------------------------ + + def test_provider(self, provider: dict[str, Any]) -> dict[str, Any]: + """Preflight a cloud provider configuration before saving it. + + Connects to the provider with the given credentials (ephemeral — + never stored) and probes every permission surface collection + needs. ``credentials`` may be inline plaintext or a + ``hive://secret/`` reference to an already-saved secret. + + Args: + provider: A ``cloudsec_provider`` hive record shape. + + Returns: + ``{"supported": bool, "report": {"provider", "ok", + "checks": [{"id","name","required","ok","detail"}, ...]}}``. + """ + return self._post("providers/test", {"provider": provider}) diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index e3076e15..9f7d5e9a 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -1,6 +1,8 @@ import os import sys +import pytest + # Get the directory of the current conftest.py file current_dir = os.path.dirname(os.path.abspath(__file__)) @@ -9,4 +11,21 @@ # Insert the project root at the beginning of sys.path if project_root not in sys.path: - sys.path.insert(0, project_root) \ No newline at end of file + sys.path.insert(0, project_root) + + +@pytest.fixture(autouse=True) +def _no_real_network(monkeypatch): + """Unit tests must never reach the network. + + Local dev machines carry live credentials in ~/.limacharlie.d, so an + unmocked CLI/SDK invocation that slips past argument validation would + otherwise perform a REAL API call (including writes) against a live + org. Tests that need HTTP behavior patch ``limacharlie.client.urlopen`` + themselves (unittest.mock.patch overrides this guard for their scope). + """ + def _blocked(*_a, **_k): + raise AssertionError( + "unit test attempted a real network call (limacharlie.client.urlopen)" + ) + monkeypatch.setattr("limacharlie.client.urlopen", _blocked) \ No newline at end of file diff --git a/tests/unit/test_cli_cloudsec.py b/tests/unit/test_cli_cloudsec.py new file mode 100644 index 00000000..0b463dab --- /dev/null +++ b/tests/unit/test_cli_cloudsec.py @@ -0,0 +1,693 @@ +"""Tests for limacharlie cloudsec CLI commands.""" + +import json + +from unittest.mock import patch, MagicMock + +from click.testing import CliRunner + +from limacharlie.cli import cli + + +def _patches(): + return ( + patch("limacharlie.commands.cloudsec.Client"), + patch("limacharlie.commands.cloudsec.Organization"), + patch("limacharlie.commands.cloudsec.CloudSec"), + ) + + +def _invoke(args, mock_cs_cls, return_value=None, stdin=None): + """Run the CLI with a mocked CloudSec instance.""" + inst = MagicMock() + mock_cs_cls.return_value = inst + if return_value is None: + return_value = {"ok": True} + # MagicMock: every SDK method returns the same renderable value. + inst.configure_mock(**{ + f"{name}.return_value": return_value + for name in [ + "get_overview", "list_changes", "get_risk_trend", "get_scan_status", + "list_findings", "get_finding_facets", "get_finding", + "set_finding_status", "bulk_set_finding_status", + "set_finding_owner", "set_finding_ticket", + "list_attack_paths", "get_public_access", "get_identity_facets", + "list_inventory", "get_inventory_facets", "get_data_security_facets", + "get_resource", "get_graph_neighbors", "list_queries", "run_query", + "get_compliance", "list_compliance_frameworks", + "list_compliance_assignments", + "list_chokepoints", "dismiss_chokepoint", "restore_chokepoint", + "resolve_sensors", "resolve_assets", + "list_caasm_assets", "list_caasm_coverage", + "get_caasm_policy", "set_caasm_policy", "caasm_ingest", + "test_provider", + ] + }) + runner = CliRunner() + result = runner.invoke(cli, ["--output", "json"] + args, input=stdin) + return result, inst + + +# --------------------------------------------------------------------------- +# Help +# --------------------------------------------------------------------------- + + +class TestCloudSecHelp: + def test_root_help_lists_subcommands(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "--help"]) + assert result.exit_code == 0 + for cmd in [ + "overview", "changes", "risk-trend", "scan-status", + "finding", "attack-path", "ciem", "inventory", "data-security", + "resource", "graph", "query", "compliance", "chokepoint", + "resolve", "caasm", "provider", + ]: + assert cmd in result.output + + def test_finding_subgroup_help(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "finding", "--help"]) + assert result.exit_code == 0 + for cmd in ["list", "facets", "get", "resolve", "bulk-resolve", + "set-owner", "set-ticket"]: + assert cmd in result.output + + def test_caasm_subgroup_help(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "caasm", "--help"]) + assert result.exit_code == 0 + for cmd in ["assets", "coverage", "policy", "ingest"]: + assert cmd in result.output + + def test_caasm_policy_subgroup_help(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "caasm", "policy", "--help"]) + assert result.exit_code == 0 + for cmd in ["get", "set"]: + assert cmd in result.output + + def test_attack_path_subgroup_help(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "attack-path", "--help"]) + assert result.exit_code == 0 + assert "list" in result.output + + def test_ciem_subgroup_help(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "ciem", "--help"]) + assert result.exit_code == 0 + for cmd in ["public-access", "facets"]: + assert cmd in result.output + + def test_inventory_subgroup_help(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "inventory", "--help"]) + assert result.exit_code == 0 + for cmd in ["list", "facets"]: + assert cmd in result.output + + def test_data_security_subgroup_help(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "data-security", "--help"]) + assert result.exit_code == 0 + assert "facets" in result.output + + def test_resource_subgroup_help(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "resource", "--help"]) + assert result.exit_code == 0 + assert "get" in result.output + + def test_graph_subgroup_help(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "graph", "--help"]) + assert result.exit_code == 0 + assert "neighbors" in result.output + + def test_query_subgroup_help(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "query", "--help"]) + assert result.exit_code == 0 + for cmd in ["list", "run"]: + assert cmd in result.output + + def test_compliance_subgroup_help(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "compliance", "--help"]) + assert result.exit_code == 0 + for cmd in ["report", "frameworks", "assignments"]: + assert cmd in result.output + + def test_chokepoint_subgroup_help(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "chokepoint", "--help"]) + assert result.exit_code == 0 + for cmd in ["list", "dismiss", "restore"]: + assert cmd in result.output + + def test_resolve_subgroup_help(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "resolve", "--help"]) + assert result.exit_code == 0 + for cmd in ["sensors", "assets"]: + assert cmd in result.output + + def test_provider_subgroup_help(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "provider", "--help"]) + assert result.exit_code == 0 + assert "test" in result.output + + +# --------------------------------------------------------------------------- +# Top-level reads +# --------------------------------------------------------------------------- + + +class TestTopLevel: + def test_overview(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "overview", "--trend-days", "90"], cls, + return_value={"score": 42}, + ) + assert result.exit_code == 0, result.output + inst.get_overview.assert_called_once_with(trend_days=90) + + def test_changes(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "changes", "--limit", "5"], cls, + return_value={"changes": []}, + ) + assert result.exit_code == 0, result.output + inst.list_changes.assert_called_once_with(limit=5) + + def test_scan_status_forwards_unlisted_provider(self): + # The provider registry grows server-side; the CLI must not pin it. + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "scan-status", "--provider", "cloudflare"], cls, + return_value={"status": {}}, + ) + assert result.exit_code == 0, result.output + inst.get_scan_status.assert_called_once_with(provider="cloudflare") + + def test_scan_status_provider(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "scan-status", "--provider", "aws"], cls, + return_value={"status": {}}, + ) + assert result.exit_code == 0, result.output + inst.get_scan_status.assert_called_once_with(provider="aws") + + +# --------------------------------------------------------------------------- +# finding +# --------------------------------------------------------------------------- + + +class TestFindingCommands: + def test_list_repeatable_filters(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + [ + "cloudsec", "finding", "list", + "--severity", "CRITICAL", "--severity", "HIGH", + "--class", "toxic_combination", + "--kev", "--reachable", + "-q", "prod", "--limit", "50", + ], + cls, + return_value={"findings": []}, + ) + assert result.exit_code == 0, result.output + inst.list_findings.assert_called_once_with( + severity=["CRITICAL", "HIGH"], + finding_class=["toxic_combination"], + status=None, + account=None, + reachable=True, + kev=True, + q="prod", + sort=None, + order=None, + cursor=None, + limit=50, + ) + + def test_list_no_kev_flag(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "finding", "list", "--no-kev"], cls, + return_value={"findings": []}, + ) + assert result.exit_code == 0, result.output + assert inst.list_findings.call_args[1]["kev"] is False + + def test_get(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "finding", "get", "fnd_abc"], cls, + return_value={"finding": {}}, + ) + assert result.exit_code == 0, result.output + inst.get_finding.assert_called_once_with("fnd_abc") + + def test_resolve(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + [ + "cloudsec", "finding", "resolve", "fnd_abc", + "--kind", "accepted", "--reason", "known", + "--expires-at", "1767225600", + ], + cls, + ) + assert result.exit_code == 0, result.output + inst.set_finding_status.assert_called_once_with( + "fnd_abc", "accepted", reason="known", expires_at=1767225600, + ) + + def test_resolve_requires_kind(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "finding", "resolve", "fnd_abc"]) + assert result.exit_code != 0 + + def test_resolve_rejects_bad_kind(self): + runner = CliRunner() + result = runner.invoke( + cli, ["cloudsec", "finding", "resolve", "fnd_abc", "--kind", "wontfix"], + ) + assert result.exit_code != 0 + + def test_bulk_resolve_rejects_open(self): + # The bulk API does not accept 'open' (reopen is single-finding only); + # the CLI must reject it at parse time instead of a server error. + runner = CliRunner() + result = runner.invoke( + cli, ["cloudsec", "finding", "bulk-resolve", + "--finding-id", "fnd_a", "--kind", "open"], + ) + assert result.exit_code != 0 + assert "open" in result.output + + def test_bulk_resolve(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + [ + "cloudsec", "finding", "bulk-resolve", + "--finding-id", "fnd_a", "--finding-id", "fnd_b", + "--kind", "mitigated", + ], + cls, + return_value={"updated": 2}, + ) + assert result.exit_code == 0, result.output + inst.bulk_set_finding_status.assert_called_once_with( + ["fnd_a", "fnd_b"], "mitigated", reason=None, expires_at=None, + ) + + def test_set_owner(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "finding", "set-owner", "fnd_abc", + "--owner", "alice@corp.com"], + cls, + ) + assert result.exit_code == 0, result.output + inst.set_finding_owner.assert_called_once_with("fnd_abc", "alice@corp.com") + + def test_set_owner_clear(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "finding", "set-owner", "fnd_abc", "--clear"], cls, + ) + assert result.exit_code == 0, result.output + inst.set_finding_owner.assert_called_once_with("fnd_abc", "") + + def test_set_owner_requires_exactly_one(self): + runner = CliRunner() + # Neither flag. + result = runner.invoke(cli, ["cloudsec", "finding", "set-owner", "fnd_abc"]) + assert result.exit_code != 0 + # Both flags. + result = runner.invoke( + cli, ["cloudsec", "finding", "set-owner", "fnd_abc", + "--owner", "x", "--clear"], + ) + assert result.exit_code != 0 + + def test_set_ticket_clear(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "finding", "set-ticket", "fnd_abc", "--clear"], cls, + ) + assert result.exit_code == 0, result.output + inst.set_finding_ticket.assert_called_once_with("fnd_abc", "") + + +# --------------------------------------------------------------------------- +# graph / query +# --------------------------------------------------------------------------- + + +class TestGraphAndQuery: + def test_neighbors(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "graph", "neighbors", "lcrn:x", "--limit", "500"], + cls, + return_value={"graph": {}}, + ) + assert result.exit_code == 0, result.output + inst.get_graph_neighbors.assert_called_once_with("lcrn:x", limit=500) + + def test_query_run_named(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "query", "run", "--named", "public-buckets"], cls, + return_value={"rows": []}, + ) + assert result.exit_code == 0, result.output + inst.run_query.assert_called_once_with( + named="public-buckets", text=None, query=None, project=None, + ) + + def test_query_run_dsl_with_project(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + [ + "cloudsec", "query", "run", + "--query-json", '{"match": "x"}', + "--project", "a, b", + ], + cls, + return_value={"rows": []}, + ) + assert result.exit_code == 0, result.output + inst.run_query.assert_called_once_with( + named=None, text=None, query={"match": "x"}, project=["a", "b"], + ) + + def test_query_run_requires_exactly_one_source(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "query", "run"]) + assert result.exit_code != 0 + result = runner.invoke( + cli, ["cloudsec", "query", "run", "--named", "n", "--text", "t"], + ) + assert result.exit_code != 0 + + def test_query_run_rejects_bad_json(self): + runner = CliRunner() + result = runner.invoke( + cli, ["cloudsec", "query", "run", "--query-json", "{not json"], + ) + assert result.exit_code != 0 + + def test_query_run_rejects_null_json(self): + # `--query-json null` parses to None and must not slip past the + # object check into an empty POST body. + runner = CliRunner() + result = runner.invoke( + cli, ["cloudsec", "query", "run", "--query-json", "null"], + ) + assert result.exit_code != 0 + + def test_query_run_rejects_empty_text(self): + # An explicit empty string is not a query; fail client-side + # instead of round-tripping to the server. + runner = CliRunner() + result = runner.invoke( + cli, ["cloudsec", "query", "run", "--text", ""], + ) + assert result.exit_code != 0 + + def test_query_run_rejects_empty_alongside_real_option(self): + # `--text foo --query-json ""` is an ambiguous invocation: the + # empty option must not be silently ignored in favor of the other. + runner = CliRunner() + result = runner.invoke( + cli, ["cloudsec", "query", "run", "--text", "foo", "--query-json", ""], + ) + assert result.exit_code != 0 + + +# --------------------------------------------------------------------------- +# resolve +# --------------------------------------------------------------------------- + + +class TestResolve: + def test_sensors_bulk(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "resolve", "sensors", "sid-1", "sid-2"], cls, + return_value={"resolved": []}, + ) + assert result.exit_code == 0, result.output + inst.resolve_sensors.assert_called_once_with(["sid-1", "sid-2"]) + + def test_sensors_requires_at_least_one(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "resolve", "sensors"]) + assert result.exit_code != 0 + + def test_assets(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "resolve", "assets", "lcrn:a"], cls, + return_value={"resolved": []}, + ) + assert result.exit_code == 0, result.output + inst.resolve_assets.assert_called_once_with(["lcrn:a"]) + + +# --------------------------------------------------------------------------- +# caasm +# --------------------------------------------------------------------------- + + +class TestCaasm: + def test_policy_set_from_json(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + policy = {"expect": [{"label": "edr", "capability": "edr", + "kinds": ["device"]}]} + result, inst = _invoke( + ["cloudsec", "caasm", "policy", "set", + "--policy-json", json.dumps(policy)], + cls, + ) + assert result.exit_code == 0, result.output + inst.set_caasm_policy.assert_called_once_with(policy) + + def test_policy_set_from_file(self, tmp_path): + p1, p2, p3 = _patches() + policy = {"expect": []} + f = tmp_path / "policy.json" + f.write_text(json.dumps(policy)) + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "caasm", "policy", "set", "--input-file", str(f)], + cls, + ) + assert result.exit_code == 0, result.output + inst.set_caasm_policy.assert_called_once_with(policy) + + def test_policy_set_from_yaml_file(self, tmp_path): + p1, p2, p3 = _patches() + f = tmp_path / "policy.yaml" + f.write_text("expect:\n - label: edr-on-devices\n capability: edr\n kinds: [device]\n") + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "caasm", "policy", "set", "--input-file", str(f)], + cls, + ) + assert result.exit_code == 0, result.output + inst.set_caasm_policy.assert_called_once_with({ + "expect": [{"label": "edr-on-devices", "capability": "edr", + "kinds": ["device"]}], + }) + + def test_policy_set_from_stdin(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "caasm", "policy", "set"], cls, + stdin='{"expect": []}', + ) + assert result.exit_code == 0, result.output + inst.set_caasm_policy.assert_called_once_with({"expect": []}) + + def test_policy_set_requires_input(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "caasm", "policy", "set"]) + assert result.exit_code != 0 + + def test_policy_set_rejects_null_json(self): + runner = CliRunner() + result = runner.invoke( + cli, ["cloudsec", "caasm", "policy", "set", "--policy-json", "null"], + ) + assert result.exit_code != 0 + + def test_policy_set_malformed_stdin_is_clean_error(self): + # Piped input that is invalid YAML AND invalid JSON must produce + # a usage error, not a raw json.JSONDecodeError traceback. (Note + # '{[unclosed' fails both parsers; something like '{"a": }' is + # VALID YAML — {'a': None} — and would proceed.) + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, _ = _invoke( + ["cloudsec", "caasm", "policy", "set"], cls, + stdin="{[unclosed", + ) + assert result.exit_code != 0 + assert result.exception is None or isinstance(result.exception, SystemExit) + assert "neither valid YAML nor JSON" in result.output + + def test_ingest_records_file(self, tmp_path): + p1, p2, p3 = _patches() + records = [{"id": "u1"}, {"id": "u2"}] + f = tmp_path / "records.json" + f.write_text(json.dumps(records)) + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "caasm", "ingest", "--source", "okta", + "--records-file", str(f)], + cls, + return_value={"result": {}}, + ) + assert result.exit_code == 0, result.output + inst.caasm_ingest.assert_called_once_with( + "okta", records=records, record=None, policy=None, + ) + + def test_ingest_single_record(self): + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "caasm", "ingest", "--source", "crowdstrike", + "--record-json", '{"device_id": "d1"}'], + cls, + return_value={"result": {}}, + ) + assert result.exit_code == 0, result.output + inst.caasm_ingest.assert_called_once_with( + "crowdstrike", records=None, record={"device_id": "d1"}, policy=None, + ) + + def test_ingest_forwards_unlisted_source(self): + # The CAASM source registry grows server-side; the CLI must not pin it. + p1, p2, p3 = _patches() + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "caasm", "ingest", "--source", "new-edr", + "--record-json", '{"id": "d1"}'], + cls, + return_value={"result": {}}, + ) + assert result.exit_code == 0, result.output + inst.caasm_ingest.assert_called_once_with( + "new-edr", records=None, record={"id": "d1"}, policy=None, + ) + + def test_ingest_rejects_non_array_records_file(self, tmp_path): + f = tmp_path / "records.json" + f.write_text('{"not": "an array"}') + runner = CliRunner() + result = runner.invoke( + cli, ["cloudsec", "caasm", "ingest", "--source", "okta", + "--records-file", str(f)], + ) + assert result.exit_code != 0 + + def test_ingest_rejects_null_records_file(self, tmp_path): + # A file whose JSON content is `null` must error, not silently + # produce a records-less ingest. + f = tmp_path / "records.json" + f.write_text("null") + runner = CliRunner() + result = runner.invoke( + cli, ["cloudsec", "caasm", "ingest", "--source", "okta", + "--records-file", str(f)], + ) + assert result.exit_code != 0 + + def test_ingest_rejects_null_record_json(self): + runner = CliRunner() + result = runner.invoke( + cli, ["cloudsec", "caasm", "ingest", "--source", "okta", + "--record-json", "null"], + ) + assert result.exit_code != 0 + + def test_ingest_records_file_accepts_yaml(self, tmp_path): + p1, p2, p3 = _patches() + f = tmp_path / "records.yaml" + f.write_text("- id: u1\n- id: u2\n") + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "caasm", "ingest", "--source", "okta", + "--records-file", str(f)], + cls, + return_value={"result": {}}, + ) + assert result.exit_code == 0, result.output + inst.caasm_ingest.assert_called_once_with( + "okta", records=[{"id": "u1"}, {"id": "u2"}], record=None, policy=None, + ) + + +# --------------------------------------------------------------------------- +# provider +# --------------------------------------------------------------------------- + + +class TestProvider: + def test_test_from_file(self, tmp_path): + p1, p2, p3 = _patches() + provider = {"provider_type": "gcp", "credentials": "hive://secret/gcp-sa"} + f = tmp_path / "provider.json" + f.write_text(json.dumps(provider)) + with p1, p2, p3 as cls: + result, inst = _invoke( + ["cloudsec", "provider", "test", "--input-file", str(f)], cls, + return_value={"supported": True, "report": {"ok": True}}, + ) + assert result.exit_code == 0, result.output + inst.test_provider.assert_called_once_with(provider) + + def test_test_requires_input(self): + runner = CliRunner() + result = runner.invoke(cli, ["cloudsec", "provider", "test"]) + assert result.exit_code != 0 + + def test_test_rejects_non_object(self): + runner = CliRunner() + result = runner.invoke( + cli, ["cloudsec", "provider", "test", "--provider-json", "[1,2]"], + ) + assert result.exit_code != 0 diff --git a/tests/unit/test_cli_lazy_loading_regression.py b/tests/unit/test_cli_lazy_loading_regression.py index 88689e4b..30b9bb3b 100644 --- a/tests/unit/test_cli_lazy_loading_regression.py +++ b/tests/unit/test_cli_lazy_loading_regression.py @@ -46,7 +46,7 @@ # Every top-level command/group that must be registered on cli. EXPECTED_TOP_LEVEL_COMMANDS = frozenset({ "ai", "ai-cost-model", "ai-memory", "ai-skill", "api", "api-key", "app", "arl", "artifact", - "audit", "auth", "billing", "case", "cloud-adapter", "completion", "config", + "audit", "auth", "billing", "case", "cloud-adapter", "cloudsec", "completion", "config", "detection", "download", "dr", "endpoint-policy", "event", "exfil", "extension", "external-adapter", "feedback", "fp", "group", "help", "hive", "ingestion-key", "installation-key", "integrity", "ioc", "job", "logging", @@ -73,6 +73,7 @@ "billing": ("group", "billing"), "case_cmd": ("group", "case"), "cloud_sensor": ("group", "cloud-adapter"), + "cloudsec": ("group", "cloudsec"), "completion": ("cmd", "completion"), "config_cmd": ("group", "config"), "detection": ("group", "detection"), @@ -147,6 +148,12 @@ "tag", "telemetry", "update", "update-note", }), "cloud-adapter": frozenset({"delete", "disable", "enable", "get", "list", "list-types", "schema", "sensors", "set", "tag"}), + "cloudsec": frozenset({ + "overview", "changes", "risk-trend", "scan-status", + "finding", "attack-path", "ciem", "inventory", "data-security", + "resource", "graph", "query", "compliance", "chokepoint", + "resolve", "caasm", "provider", + }), "detection": frozenset({"get", "list"}), "download": frozenset({"adapter", "list", "sensor"}), "dr": frozenset({ diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index 0d236210..df4a1681 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -195,6 +195,30 @@ def test_successful_get_request(self, mock_urlopen): assert result == {"sensors": []} + @patch("limacharlie.client.urlopen") + def test_query_params_sequences_expand_to_repeated_keys(self, mock_urlopen): + # doseq: a dict-of-lists (or list-of-tuples) must encode as + # repeated keys, not the Python list repr. + jwt_response = MagicMock() + jwt_response.read.return_value = json.dumps({"jwt": "test-jwt"}).encode() + jwt_response.close = MagicMock() + + api_response = MagicMock() + api_response.read.return_value = json.dumps({}).encode() + api_response.close = MagicMock() + api_response.getheaders.return_value = [] + + mock_urlopen.side_effect = [jwt_response, api_response] + + client = Client(oid="test-oid", api_key="test-key") + client.request( + "GET", "things", + query_params={"severity": ["HIGH", "LOW"], "q": "prod"}, + ) + + sent = mock_urlopen.call_args_list[-1][0][0] + assert sent.full_url.endswith("?severity=HIGH&severity=LOW&q=prod") + @patch("limacharlie.client.urlopen") def test_auto_refreshes_jwt_on_401(self, mock_urlopen): from urllib.error import HTTPError diff --git a/tests/unit/test_sdk_cloudsec.py b/tests/unit/test_sdk_cloudsec.py new file mode 100644 index 00000000..80773d03 --- /dev/null +++ b/tests/unit/test_sdk_cloudsec.py @@ -0,0 +1,400 @@ +"""Tests for limacharlie.sdk.cloudsec module.""" + +import json + +from unittest.mock import MagicMock + +import pytest + +from limacharlie.sdk.cloudsec import CloudSec + + +OID = "11111111-2222-3333-4444-555555555555" + + +@pytest.fixture +def mock_org(): + org = MagicMock() + org.oid = OID + org.client = MagicMock() + return org + + +@pytest.fixture +def cs(mock_org): + return CloudSec(mock_org) + + +def _get_call(mock_org): + """Pull (url, query_pairs) out of a mocked GET client.request call.""" + args, kwargs = mock_org.client.request.call_args + assert args[0] == "GET" + return args[1], kwargs.get("query_params") + + +def _post_call(mock_org): + """Pull (url, decoded_json_body) out of a mocked POST client.request call.""" + args, kwargs = mock_org.client.request.call_args + assert args[0] == "POST" + assert kwargs["content_type"] == "application/json" + return args[1], json.loads(kwargs["raw_body"]) + + +class TestBasics: + def test_oid_property(self, cs): + assert cs.oid == OID + + def test_get_omits_empty_query(self, cs, mock_org): + mock_org.client.request.return_value = {"chokepoints": []} + cs.list_chokepoints() + url, qp = _get_call(mock_org) + assert url == f"cloudsec/{OID}/chokepoints" + assert qp is None + + +class TestFindings: + def test_list_findings_defaults(self, cs, mock_org): + mock_org.client.request.return_value = {"findings": []} + cs.list_findings() + url, qp = _get_call(mock_org) + assert url == f"cloudsec/{OID}/findings" + assert qp is None + + def test_list_findings_all_selectors(self, cs, mock_org): + mock_org.client.request.return_value = {} + cs.list_findings( + severity=["CRITICAL", "HIGH"], + finding_class=["toxic_combination"], + status=["open"], + account=["acct-1"], + reachable=True, + kev=False, + q="prod", + sort="lc_risk", + order="desc", + cursor="c1", + limit=50, + ) + _, qp = _get_call(mock_org) + # Repeatable keys appear once per value; booleans are lowered. + assert qp == [ + ("severity", "CRITICAL"), + ("severity", "HIGH"), + ("finding_class", "toxic_combination"), + ("status", "open"), + ("account", "acct-1"), + ("reachable", "true"), + ("kev", "false"), + ("q", "prod"), + ("sort", "lc_risk"), + ("order", "desc"), + ("cursor", "c1"), + ("limit", "50"), + ] + + def test_finding_facets_takes_filters_only(self, cs, mock_org): + mock_org.client.request.return_value = {} + cs.get_finding_facets(severity=["LOW"], kev=True) + url, qp = _get_call(mock_org) + assert url == f"cloudsec/{OID}/findings/facets" + assert qp == [("severity", "LOW"), ("kev", "true")] + + def test_get_finding(self, cs, mock_org): + mock_org.client.request.return_value = {"finding": {}} + cs.get_finding("fnd_abc") + url, qp = _get_call(mock_org) + assert url == f"cloudsec/{OID}/findings/fnd_abc" + assert qp is None + + +class TestFindingWrites: + def test_set_finding_status(self, cs, mock_org): + mock_org.client.request.return_value = {"ok": True} + cs.set_finding_status( + "fnd_abc", "accepted", reason="known", expires_at=1767225600, + ) + url, body = _post_call(mock_org) + assert url == f"cloudsec/{OID}/findings/fnd_abc/status" + assert body == {"resolution": { + "kind": "accepted", "reason": "known", "expires_at": 1767225600, + }} + + def test_set_finding_status_reopen_omits_optionals(self, cs, mock_org): + mock_org.client.request.return_value = {"ok": True} + cs.set_finding_status("fnd_abc", "open") + _, body = _post_call(mock_org) + assert body == {"resolution": {"kind": "open"}} + + def test_bulk_set_finding_status(self, cs, mock_org): + mock_org.client.request.return_value = {"updated": 2} + cs.bulk_set_finding_status(["fnd_a", "fnd_b"], "mitigated", reason="fixed") + url, body = _post_call(mock_org) + assert url == f"cloudsec/{OID}/findings/bulk/status" + assert body == { + "finding_ids": ["fnd_a", "fnd_b"], + "resolution": {"kind": "mitigated", "reason": "fixed"}, + } + + def test_set_finding_owner_empty_clears(self, cs, mock_org): + mock_org.client.request.return_value = {"ok": True} + cs.set_finding_owner("fnd_abc", "") + url, body = _post_call(mock_org) + assert url == f"cloudsec/{OID}/findings/fnd_abc/owner" + assert body == {"owner": ""} + + def test_set_finding_ticket(self, cs, mock_org): + mock_org.client.request.return_value = {"ok": True} + cs.set_finding_ticket("fnd_abc", "JIRA-123") + url, body = _post_call(mock_org) + assert url == f"cloudsec/{OID}/findings/fnd_abc/ticket" + assert body == {"ticket": "JIRA-123"} + + +class TestAttackPathsAndCiem: + def test_list_attack_paths_selectors(self, cs, mock_org): + mock_org.client.request.return_value = {"paths": []} + cs.list_attack_paths(severity=["CRITICAL"], q="db") + url, qp = _get_call(mock_org) + assert url == f"cloudsec/{OID}/attack-paths" + assert qp == [("severity", "CRITICAL"), ("q", "db")] + + def test_public_access(self, cs, mock_org): + mock_org.client.request.return_value = {"access": []} + cs.get_public_access() + url, _ = _get_call(mock_org) + assert url == f"cloudsec/{OID}/ciem/public-access" + + def test_identity_facets(self, cs, mock_org): + mock_org.client.request.return_value = {"facets": {}} + cs.get_identity_facets() + url, _ = _get_call(mock_org) + assert url == f"cloudsec/{OID}/ciem/facets" + + +class TestInventoryAndResources: + def test_list_inventory_maps_type_selector(self, cs, mock_org): + mock_org.client.request.return_value = {"resources": []} + cs.list_inventory(resource_type="gcp_bucket", region="us-central1", limit=10) + url, qp = _get_call(mock_org) + assert url == f"cloudsec/{OID}/inventory" + # resource_type is sent as the gateway's `type` selector. + assert qp == [ + ("type", "gcp_bucket"), ("region", "us-central1"), ("limit", "10"), + ] + + def test_inventory_facets(self, cs, mock_org): + mock_org.client.request.return_value = {} + cs.get_inventory_facets() + url, _ = _get_call(mock_org) + assert url == f"cloudsec/{OID}/inventory/facets" + + def test_data_security_facets(self, cs, mock_org): + mock_org.client.request.return_value = {} + cs.get_data_security_facets() + url, _ = _get_call(mock_org) + assert url == f"cloudsec/{OID}/data-security/facets" + + def test_get_resource(self, cs, mock_org): + mock_org.client.request.return_value = {"resource": None} + cs.get_resource("lcrn:x") + url, qp = _get_call(mock_org) + assert url == f"cloudsec/{OID}/resource" + assert qp == [("urn", "lcrn:x")] + + +class TestGraphAndQueries: + def test_graph_neighbors(self, cs, mock_org): + mock_org.client.request.return_value = {"graph": {}} + cs.get_graph_neighbors("lcrn:x", limit=500) + url, qp = _get_call(mock_org) + assert url == f"cloudsec/{OID}/graph/neighbors" + assert qp == [("urn", "lcrn:x"), ("limit", "500")] + + def test_list_queries(self, cs, mock_org): + mock_org.client.request.return_value = {"queries": []} + cs.list_queries() + url, _ = _get_call(mock_org) + assert url == f"cloudsec/{OID}/queries" + + def test_run_query_named(self, cs, mock_org): + mock_org.client.request.return_value = {"rows": []} + cs.run_query(named="public-buckets") + url, body = _post_call(mock_org) + assert url == f"cloudsec/{OID}/query" + assert body == {"named": "public-buckets"} + + def test_run_query_dsl_with_projection(self, cs, mock_org): + mock_org.client.request.return_value = {"rows": []} + cs.run_query(query={"match": "x"}, project=["a", "b"]) + _, body = _post_call(mock_org) + assert body == {"query": {"match": "x"}, "project": ["a", "b"]} + + +class TestCompliance: + def test_report_default(self, cs, mock_org): + mock_org.client.request.return_value = {"report": {}} + cs.get_compliance() + url, qp = _get_call(mock_org) + assert url == f"cloudsec/{OID}/compliance" + assert qp is None + + def test_report_assignment(self, cs, mock_org): + mock_org.client.request.return_value = {"report": {}} + cs.get_compliance(assignment="prod-scope") + _, qp = _get_call(mock_org) + assert qp == [("assignment", "prod-scope")] + + def test_frameworks_and_assignments(self, cs, mock_org): + mock_org.client.request.return_value = {} + cs.list_compliance_frameworks() + assert _get_call(mock_org)[0] == f"cloudsec/{OID}/compliance/frameworks" + cs.list_compliance_assignments() + assert _get_call(mock_org)[0] == f"cloudsec/{OID}/compliance/assignments" + + +class TestOverviewTrendsChokepoints: + def test_overview_trend_days(self, cs, mock_org): + mock_org.client.request.return_value = {"score": 0} + cs.get_overview(trend_days=90) + url, qp = _get_call(mock_org) + assert url == f"cloudsec/{OID}/overview" + assert qp == [("trend_days", "90")] + + def test_changes_limit(self, cs, mock_org): + mock_org.client.request.return_value = {"changes": []} + cs.list_changes(limit=100) + url, qp = _get_call(mock_org) + assert url == f"cloudsec/{OID}/changes" + assert qp == [("limit", "100")] + + def test_risk_trend(self, cs, mock_org): + mock_org.client.request.return_value = {"trend": []} + cs.get_risk_trend() + url, qp = _get_call(mock_org) + assert url == f"cloudsec/{OID}/risk-trend" + assert qp is None + + def test_scan_status_provider(self, cs, mock_org): + mock_org.client.request.return_value = {"status": {}} + cs.get_scan_status(provider="aws") + url, qp = _get_call(mock_org) + assert url == f"cloudsec/{OID}/scan-status" + assert qp == [("provider", "aws")] + + def test_scan_status_provider_is_case_normalized(self, cs, mock_org): + # The backend scan-state read is a case-sensitive lookup keyed on + # lowercase provider ids; "AWS" would silently read as never-scanned. + mock_org.client.request.return_value = {"status": {}} + cs.get_scan_status(provider=" AWS ") + _, qp = _get_call(mock_org) + assert qp == [("provider", "aws")] + + def test_dismiss_chokepoint_with_reason(self, cs, mock_org): + mock_org.client.request.return_value = {"ok": True} + cs.dismiss_chokepoint("lcrn:x", reason="decom") + url, body = _post_call(mock_org) + assert url == f"cloudsec/{OID}/chokepoints/dismiss" + assert body == {"urn": "lcrn:x", "reason": "decom"} + + def test_restore_chokepoint(self, cs, mock_org): + mock_org.client.request.return_value = {"ok": True} + cs.restore_chokepoint("lcrn:x") + url, body = _post_call(mock_org) + assert url == f"cloudsec/{OID}/chokepoints/restore" + assert body == {"urn": "lcrn:x"} + + +class TestResolution: + def test_resolve_sensors_bulk(self, cs, mock_org): + mock_org.client.request.return_value = {"resolved": []} + cs.resolve_sensors(["sid-1", "sid-2"]) + url, qp = _get_call(mock_org) + assert url == f"cloudsec/{OID}/resolve/sensors" + assert qp == [("sid", "sid-1"), ("sid", "sid-2")] + + def test_resolve_assets_bulk(self, cs, mock_org): + mock_org.client.request.return_value = {"resolved": []} + cs.resolve_assets(["lcrn:a", "lcrn:b"]) + url, qp = _get_call(mock_org) + assert url == f"cloudsec/{OID}/resolve/assets" + assert qp == [("urn", "lcrn:a"), ("urn", "lcrn:b")] + + def test_resolve_sensors_chunks_large_batches(self, cs, mock_org): + # Ids ride as repeated query params, so large batches must be + # split to stay within the ~8KB URL limit, and the per-chunk + # responses merged. + sids = [f"sid-{i}" for i in range(250)] + mock_org.client.request.side_effect = [ + {"resolved": [{"sid": "a"}], "unresolved": ["x"]}, + {"resolved": [{"sid": "b"}], "unresolved": []}, + {"resolved": [], "unresolved": ["y"]}, + ] + out = cs.resolve_sensors(sids) + calls = mock_org.client.request.call_args_list + assert len(calls) == 3 + sent = [kwargs["query_params"] for _, kwargs in calls] + assert [len(qp) for qp in sent] == [100, 100, 50] + # Every id is sent exactly once, in order. + assert [v for qp in sent for _, v in qp] == sids + assert out == { + "resolved": [{"sid": "a"}, {"sid": "b"}], + "unresolved": ["x", "y"], + } + + def test_resolve_assets_empty_batch_makes_no_request(self, cs, mock_org): + out = cs.resolve_assets([]) + mock_org.client.request.assert_not_called() + assert out == {"resolved": [], "unresolved": []} + + +class TestCaasm: + def test_list_assets(self, cs, mock_org): + mock_org.client.request.return_value = {"resources": []} + cs.list_caasm_assets(q="laptop", limit=50) + url, qp = _get_call(mock_org) + assert url == f"cloudsec/{OID}/caasm/assets" + assert qp == [("q", "laptop"), ("limit", "50")] + + def test_list_coverage(self, cs, mock_org): + mock_org.client.request.return_value = {"findings": []} + cs.list_caasm_coverage(status=["open"], severity=["HIGH"], cursor="c") + url, qp = _get_call(mock_org) + assert url == f"cloudsec/{OID}/caasm/coverage" + assert qp == [("status", "open"), ("severity", "HIGH"), ("cursor", "c")] + + def test_policy_get(self, cs, mock_org): + mock_org.client.request.return_value = {"resources": []} + cs.get_caasm_policy() + url, _ = _get_call(mock_org) + assert url == f"cloudsec/{OID}/caasm/policy" + + def test_policy_set(self, cs, mock_org): + mock_org.client.request.return_value = {"ok": True} + policy = {"expect": [{"label": "edr", "capability": "edr", "kinds": ["device"]}]} + cs.set_caasm_policy(policy) + url, body = _post_call(mock_org) + assert url == f"cloudsec/{OID}/caasm/policy" + assert body == {"policy": policy} + + def test_ingest_records(self, cs, mock_org): + mock_org.client.request.return_value = {"result": {}} + cs.caasm_ingest("okta", records=[{"id": "u1"}]) + url, body = _post_call(mock_org) + assert url == f"cloudsec/{OID}/caasm/ingest" + assert body == {"source": "okta", "records": [{"id": "u1"}]} + + def test_ingest_single_record(self, cs, mock_org): + mock_org.client.request.return_value = {"result": {}} + cs.caasm_ingest("crowdstrike", record={"device_id": "d1"}) + _, body = _post_call(mock_org) + assert body == {"source": "crowdstrike", "record": {"device_id": "d1"}} + + +class TestProvider: + def test_test_provider(self, cs, mock_org): + mock_org.client.request.return_value = {"supported": True, "report": {}} + provider = {"provider_type": "gcp", "credentials": "hive://secret/gcp-sa"} + cs.test_provider(provider) + url, body = _post_call(mock_org) + assert url == f"cloudsec/{OID}/providers/test" + assert body == {"provider": provider}