Add Cloud Security (CNAPP) support: 'limacharlie cloudsec' CLI + SDK

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

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

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,
+        )