diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 069a55bb..e8e54c51 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -4,11 +4,9 @@ on: push: branches: - master - - cli-v2 pull_request: branches: - master - - cli-v2 permissions: contents: read diff --git a/doc/cli/hive-data.md b/doc/cli/hive-data.md index c8b3ebc6..f7ad3890 100644 --- a/doc/cli/hive-data.md +++ b/doc/cli/hive-data.md @@ -64,6 +64,18 @@ limacharlie external-adapter list limacharlie cloud-adapter list ``` +### app + +User-authored, AI-generated mini web apps (a self-contained HTML document +rendered in a sandboxed iframe by the web UI). + +```bash +limacharlie app list +limacharlie app get --key my-app +limacharlie app set --key my-app --input-file app.yaml +limacharlie app delete --key my-app --confirm +``` + ## extension ```bash diff --git a/doc/cli/platform-admin.md b/doc/cli/platform-admin.md index 63e747f9..440a91ac 100644 --- a/doc/cli/platform-admin.md +++ b/doc/cli/platform-admin.md @@ -9,6 +9,7 @@ Commands for organization management, users, groups, API keys, ingestion keys, b ```bash limacharlie org info # Name, sensor count, version, quotas limacharlie org stats # Usage statistics +limacharlie org quota-usage # Enforced sensor quota usage + breakdown limacharlie org urls # Service URLs (for firewall rules) limacharlie org errors # Platform errors limacharlie org dismiss-error --component diff --git a/limacharlie/ai_help.py b/limacharlie/ai_help.py index 4a75ea5f..c9673772 100644 --- a/limacharlie/ai_help.py +++ b/limacharlie/ai_help.py @@ -140,6 +140,8 @@ def _top_level_help(cli: click.Group) -> str: lines.append("--wide Don't truncate table columns") lines.append("--filter JMESPATH Filter/transform output") lines.append("--fields f1,f2 Select specific output fields") + lines.append("--sort-by FIELD Sort list output by a field") + lines.append("--reverse Reverse sorted order (with --sort-by)") lines.append("--quiet Suppress non-data output") lines.append("--debug Print HTTP request details") lines.append("--env NAME Use a named environment from config") diff --git a/limacharlie/cli.py b/limacharlie/cli.py index 21f123cb..56df3d72 100644 --- a/limacharlie/cli.py +++ b/limacharlie/cli.py @@ -53,6 +53,9 @@ class LimaCharlieContext: wide: bool = False no_warnings: bool = False filter_expr: str | None = None + fields: list[str] | None = None + sort_by: str | None = None + reverse: bool = False profile: str | None = None environment: str | None = None @@ -99,6 +102,7 @@ def _config_no_warnings() -> bool: "ai-skill": ("ai_skill", "group"), "api": ("api_cmd", "cmd"), "api-key": ("api_key", "group"), + "app": ("app", "group"), "arl": ("arl", "group"), "artifact": ("artifact", "group"), "audit": ("audit", "group"), @@ -346,11 +350,14 @@ def _find_shadowed_opts( @click.option("--wide", "-W", is_flag=True, default=False, help="Disable table value truncation (show full values).") @click.option("--no-warnings", is_flag=True, default=False, help="Suppress advisory warnings (cost notices, memory hints, checkpoint suggestions).") @click.option("--filter", "filter_expr", default=None, help="JMESPath expression to filter/transform output (e.g. 'user_perms', 'keys(@)').") +@click.option("--fields", "fields", default=None, help="Comma-separated field names to keep in output (e.g. 'sid,hostname'). Applied to list/record output.") +@click.option("--sort-by", "sort_by", default=None, help="Field name to sort list output by.") +@click.option("--reverse", "reverse", is_flag=True, default=False, help="Reverse the order of sorted list output (use with --sort-by).") @click.option("--profile", default=None, help="Named credential profile to use.") @click.option("--env", "environment", default=None, help="Named environment from config file.") @click.version_option(version=__version__, prog_name="limacharlie") @click.pass_context -def cli(ctx: click.Context, oid: str | None, output_format: str | None, debug: bool, debug_full: bool, debug_curl: bool, quiet: bool, wide: bool, no_warnings: bool, filter_expr: str | None, profile: str | None, environment: str | None) -> None: +def cli(ctx: click.Context, oid: str | None, output_format: str | None, debug: bool, debug_full: bool, debug_curl: bool, quiet: bool, wide: bool, no_warnings: bool, filter_expr: str | None, fields: str | None, sort_by: str | None, reverse: bool, profile: str | None, environment: str | None) -> None: """LimaCharlie CLI - Endpoint Detection & Response platform. Manage sensors, detection rules, hive data, and more from the command line. @@ -368,14 +375,22 @@ def cli(ctx: click.Context, oid: str | None, output_format: str | None, debug: b lc_ctx.wide = wide lc_ctx.no_warnings = no_warnings or _config_no_warnings() lc_ctx.filter_expr = filter_expr + # Parse --fields into a clean list of names (drop blanks/whitespace). + field_list = [f.strip() for f in fields.split(",") if f.strip()] if fields else None + lc_ctx.fields = field_list + lc_ctx.sort_by = sort_by + lc_ctx.reverse = reverse lc_ctx.profile = profile lc_ctx.environment = environment # Lazy import: output pulls in jmespath, tabulate, yaml, csv (~14ms). # Deferring to here avoids that cost for fast paths like --help, --version, # and --ai-help that never render command output. - from .output import set_filter_expr, set_wide_mode + from .output import set_filter_expr, set_wide_mode, set_fields, set_sort_by, set_reverse set_wide_mode(wide) set_filter_expr(filter_expr) + set_fields(field_list) + set_sort_by(sort_by) + set_reverse(reverse) # Inject --ai-help on the root cli group itself (subcommands get it lazily diff --git a/limacharlie/commands/_adapter_types.py b/limacharlie/commands/_adapter_types.py new file mode 100644 index 00000000..3ccb50c0 --- /dev/null +++ b/limacharlie/commands/_adapter_types.py @@ -0,0 +1,360 @@ +"""Shared helpers for listing supported adapter/sensor types. + +Both the cloud-adapter and external-adapter command groups expose a +``list-types`` subcommand. The list of supported types is derived at +runtime from the ``cloud_sensor`` hive's JSON-Schema so it cannot go +stale relative to the backend; a curated fallback is used only if the +schema cannot be fetched or does not advertise the per-type sub-structs. +""" + +from __future__ import annotations + +import re +from typing import Any + +import click + +from ..sdk.hive import Hive + +# Fields that appear at the top level of an adapter config but are NOT +# adapter type names (they are shared across every adapter type). These +# are filtered out when deriving the type list from the schema. +_NON_TYPE_FIELDS = { + "sensor_type", + "client_options", + "mapping", + "mappings", + "indexing", +} + +# Curated fallback list. IMPORTANT: this is only used when the live +# cloud_sensor schema cannot be fetched or parsed. It MUST be kept in +# sync with the backend's supported adapter types; prefer the schema- +# derived list, which cannot go stale. +_FALLBACK_ADAPTER_TYPES: dict[str, str] = { + "1password": "1Password audit events", + "azure_event_hub": "Azure Event Hub stream", + "carbon_black": "VMware Carbon Black events", + "cato": "Cato Networks events", + "crowdstrike": "CrowdStrike Falcon Data Replicator", + "duo": "Cisco Duo authentication logs", + "entraid": "Microsoft Entra ID (Azure AD) logs", + "file": "Tail a local file", + "gcs": "Google Cloud Storage objects", + "github": "GitHub audit log", + "google_workspace": "Google Workspace activity", + "guardduty": "AWS GuardDuty findings", + "imap": "IMAP mailbox ingestion", + "itglue": "IT Glue records", + "k8s_pods": "Kubernetes pod logs", + "mac_unified_logging": "macOS unified logging", + "mimecast": "Mimecast email security logs", + "ms_graph": "Microsoft Graph API", + "office365": "Microsoft Office 365 management activity", + "okta": "Okta system log", + "pubsub": "Google Cloud Pub/Sub", + "s3": "AWS S3 objects", + "sentinelone": "SentinelOne events", + "simulation": "Simulated/test events", + "slack": "Slack audit logs", + "sophos": "Sophos Central events", + "sqs": "AWS SQS queue", + "stdin": "Read events from stdin (external adapter only)", + "syslog": "Syslog over TCP/UDP", + "threatlocker": "ThreatLocker unified audit", + "webhook": "Inbound HTTP webhook", + "wel": "Windows Event Log (external adapter only)", + "wiz": "Wiz cloud security findings", +} + + +def _resolve_ref(root: dict, ref: str) -> dict | None: + """Resolve a local JSON-Schema ``$ref`` (e.g. ``#/$defs/CloudSensorRecord``).""" + if not isinstance(ref, str) or not ref.startswith("#/"): + return None + node: Any = root + for part in ref[2:].split("/"): + if not isinstance(node, dict): + return None + node = node.get(part) + return node if isinstance(node, dict) else None + + +def _describe(name: str) -> str: + """Best-effort human description for a derived type name (blank if unknown).""" + return ( + _FALLBACK_ADAPTER_TYPES.get(name) + or _FALLBACK_ADAPTER_TYPES.get(name.replace("_", "")) + or "" + ) + + +def _types_from_schema(schema: Any) -> list[str] | None: + """Derive adapter type names from an adapter hive JSON-Schema. + + The reflected schema is a root that ``$ref``s into ``$defs`` (e.g. + ``CloudSensorRecord`` / ``ExternalAdapterConfig``); the per-adapter config + lives in that record as a property keyed by the adapter type name + (``s3``, ``office365``, ``threatlocker``, …), alongside the ``sensor_type`` + discriminator. The type names are therefore the record's ``properties`` + minus the shared/non-type fields — NOT the bare ``$defs`` keys, which are + helper structs (``ClientOptions``, ``AckBufferOptions``, …). Returns + ``None`` if the schema does not expose any usable type names. + """ + if isinstance(schema, dict) and isinstance(schema.get("schema"), dict): + schema = schema["schema"] + if not isinstance(schema, dict): + return None + + # Follow a top-level $ref into the record definition. + record = schema + ref = schema.get("$ref") + if isinstance(ref, str): + resolved = _resolve_ref(schema, ref) + if resolved is not None: + record = resolved + + props = record.get("properties") + if not isinstance(props, dict): + return None + + names = {n for n in props if n and not n.startswith("_")} - _NON_TYPE_FIELDS + return sorted(names) or None + + +def adapter_types(org: Any, hive_name: str = "cloud_sensor") -> list[dict[str, str]]: + """Return the supported adapter types for a hive as name/description rows. + + Prefers the live hive schema (so it tracks the backend); falls back to the + curated constant when the schema is unavailable or does not advertise types. + """ + derived: list[str] | None = None + try: + schema = Hive(org, hive_name).get_schema() + derived = _types_from_schema(schema) + except Exception: + derived = None + + if derived: + return [{"type": name, "description": _describe(name)} for name in derived] + return [ + {"type": name, "description": desc} + for name, desc in sorted(_FALLBACK_ADAPTER_TYPES.items()) + ] + + +_EXPLAIN_LIST_TYPES = """\ +List the supported adapter/sensor type names with a short description. + +The list is derived from the live adapter hive JSON-Schema when +available (so it tracks the backend), with a curated fallback otherwise. +Use a type name as the top-level sensor_type when calling 'set'. + +The cloud-adapter and external-adapter type sets differ: cloud adapters +run in LimaCharlie's infrastructure, external (on-prem) adapters add +types like syslog, file, stdin and wel. Consult the set --ai-help for +the per-type config shape. +""" + + +_UUID_RE = re.compile(r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$") + + +def _record_root(schema: Any) -> tuple[dict | None, dict | None]: + """Return (root_schema, record_node) for an adapter hive schema. + + Unwraps the {"schema": {...}} envelope and follows the root $ref into the + record definition (CloudSensorRecord / ExternalAdapterConfig). + """ + if isinstance(schema, dict) and isinstance(schema.get("schema"), dict): + schema = schema["schema"] + if not isinstance(schema, dict): + return None, None + record = schema + ref = schema.get("$ref") + if isinstance(ref, str): + resolved = _resolve_ref(schema, ref) + if resolved is not None: + record = resolved + return schema, record + + +def adapter_type_schema(org: Any, hive_name: str, type_name: str) -> tuple[Any, dict | None]: + """Resolve one adapter type's config sub-schema. + + Returns (root_schema, type_node): type_node is the JSON-Schema node for the + per-type config (e.g. the threatlocker sub-struct), or None if the type is + unknown. root_schema is returned so the caller can flatten/render with $ref + resolution against $defs. + """ + schema = Hive(org, hive_name).get_schema() + root, record = _record_root(schema) + if record is None: + return schema, None + props = record.get("properties") + if not isinstance(props, dict) or type_name not in props: + return root, None + return root, props[type_name] + + +def adapter_sensors(org: Any, hive_name: str, key: str) -> dict[str, Any]: + """Find the live sensor(s) belonging to an adapter record. + + Reads the adapter record, extracts its installation key (the sensor iid), + and returns the sensors whose iid matches. Falls back to matching the + adapter's configured hostname when the installation key isn't a bare iid + (UUID). The sensors list is empty when the adapter hasn't registered a + sensor yet (e.g. it has not delivered any events). + """ + record = Hive(org, hive_name).get(key) + data = getattr(record, "data", None) or {} + sensor_type = data.get("sensor_type") + sub = data.get(sensor_type, {}) if sensor_type else {} + client_options = sub.get("client_options", {}) if isinstance(sub, dict) else {} + identity = client_options.get("identity", {}) if isinstance(client_options, dict) else {} + install_key = identity.get("installation_key") + hostname = client_options.get("hostname") + + if isinstance(install_key, str) and _UUID_RE.match(install_key): + match_by, match_val = "iid", install_key + elif hostname: + match_by, match_val = "hostname", hostname + else: + return {"adapter": key, "match_by": None, "match_value": None, "selector": None, "sensors": [], + "note": "Adapter record has no resolvable installation_key (iid) or hostname to match on."} + + # Filter server-side with a sensor selector (both iid and hostname are + # supported selector fields), so this scales to large fleets instead of + # paging every sensor and filtering client-side. + selector = f'{match_by} == "{match_val}"' + sensors = [ + {"sid": s.get("sid"), "hostname": s.get("hostname"), "iid": s.get("iid"), + "is_online": s.get("is_online"), "last_seen": s.get("alive")} + for s in org.list_sensors(selector=selector) + ] + out = {"adapter": key, "match_by": match_by, "match_value": match_val, "selector": selector, "sensors": sensors} + if not sensors: + out["note"] = ("No sensor registered for this adapter yet — it has not delivered any " + "events (a cloud adapter materializes a sensor on first event). Normal " + "for a freshly-created adapter.") + return out + + +_EXPLAIN_SCHEMA = """\ +Show the configuration schema for ONE adapter/sensor type as a flat field +listing (field | type | required | notes), resolved from the live adapter hive +JSON-Schema. Use this before 'set' to learn the exact field set and where each +field lives (e.g. that hostname goes under client_options, not at the top +level). Pass --output json for the raw JSON-Schema node. + +Run 'list-types' first to see the valid --type values. +""" + +_EXPLAIN_SENSORS = """\ +List the live sensor(s) this adapter has produced. Reads the adapter record's +installation key (iid) and returns sensors whose iid matches — the reliable way +to get a cloud/external adapter's SID without decoding the installation key. + +An empty result means the adapter has not registered a sensor yet (it has not +delivered any events); that is expected for a freshly-created adapter. +""" + + +def add_schema(group: click.Group, command_path: str, hive_name: str = "cloud_sensor") -> None: + """Attach a ``schema --type `` subcommand to an adapter command group.""" + from ..cli import pass_context + from ..client import Client + from ..sdk.organization import Organization + from ..output import format_output, detect_output_format + from ..discovery import register_explain + from .hive import _flatten_schema + + @group.command("schema", help="Show one adapter type's config schema.") + @click.option("--type", "type_name", required=True, help="Adapter type (see list-types).") + @pass_context + def schema_cmd(ctx, type_name) -> None: + 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) + root, node = adapter_type_schema(org, hive_name, type_name) + if node is None: + types = ", ".join(t["type"] for t in adapter_types(org, hive_name)) + raise click.UsageError(f"Unknown adapter type '{type_name}'. Valid types: {types}") + if ctx.obj.quiet: + return + fmt = ctx.obj.output_format or detect_output_format() + if fmt == "table" and isinstance(root, dict): + rows = _flatten_schema(node, root) + if rows: + click.echo(format_output(rows, fmt)) + return + # Raw JSON-Schema node for machine formats: resolve a top-level $ref and + # carry the root $defs so nested $refs in the node stay resolvable. + resolved = _resolve_ref(root, node["$ref"]) if isinstance(node, dict) and "$ref" in node and isinstance(root, dict) else node + if isinstance(resolved, dict) and isinstance(root, dict) and isinstance(root.get("$defs"), dict): + resolved = {**resolved, "$defs": root["$defs"]} + click.echo(format_output(resolved, fmt)) + + register_explain(command_path, _EXPLAIN_SCHEMA) + + +def add_sensors(group: click.Group, command_path: str, hive_name: str = "cloud_sensor") -> None: + """Attach a ``sensors --key `` subcommand to an adapter command group.""" + from ..cli import pass_context + from ..client import Client + from ..sdk.organization import Organization + from ..output import format_output, detect_output_format + from ..discovery import register_explain + + @group.command("sensors", help="List the live sensor(s) this adapter produced.") + @click.option("--key", required=True, help="Adapter record key.") + @pass_context + def sensors_cmd(ctx, key) -> None: + 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) + data = adapter_sensors(org, hive_name, key) + if not ctx.obj.quiet: + fmt = ctx.obj.output_format or detect_output_format() + click.echo(format_output(data, fmt)) + + register_explain(command_path, _EXPLAIN_SENSORS) + + +def add_list_types(group: click.Group, command_path: str, hive_name: str = "cloud_sensor") -> None: + """Attach a ``list-types`` subcommand to an adapter command group. + + ``hive_name`` selects which adapter hive's schema to enumerate + (``cloud_sensor`` vs ``external_adapter``) so each group lists its own + supported types. + """ + from ..cli import pass_context + from ..client import Client + from ..sdk.organization import Organization + from ..output import format_output, detect_output_format + from ..discovery import register_explain + + @group.command("list-types", help="List supported adapter/sensor types.") + @pass_context + def list_types_cmd(ctx) -> None: + 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) + data = adapter_types(org, hive_name) + if not ctx.obj.quiet: + fmt = ctx.obj.output_format or detect_output_format() + click.echo(format_output(data, fmt)) + + register_explain(command_path, _EXPLAIN_LIST_TYPES) diff --git a/limacharlie/commands/_hive_shortcut.py b/limacharlie/commands/_hive_shortcut.py index 4659a991..1696e537 100644 --- a/limacharlie/commands/_hive_shortcut.py +++ b/limacharlie/commands/_hive_shortcut.py @@ -28,7 +28,7 @@ def _output(ctx: click.Context, data: Any) -> None: click.echo(format_output(data, fmt)) -def make_hive_group(group_name: str, hive_name: str, noun_singular: str, noun_plural: str | None = None) -> click.Group: +def make_hive_group(group_name: str, hive_name: str, noun_singular: str, noun_plural: str | None = None, value_key: str | None = None) -> click.Group: """Create a Click group for a specific hive type. Args: @@ -36,6 +36,13 @@ def make_hive_group(group_name: str, hive_name: str, noun_singular: str, noun_pl hive_name: Hive backend name (e.g., "secret"). noun_singular: Human-readable singular (e.g., "secret"). noun_plural: Human-readable plural (defaults to noun_singular + "s"). + value_key: Name of the single scalar field in the record's ``data`` + payload for hives whose value is one scalar (e.g. ``"secret"`` for + the secret hive, whose data is ``{secret: }``). When set, the + 'set' command gains a ``--value`` convenience flag that wraps the + value as ``{data: {: }}``. Leave None for hives + whose data is structured (lookup, fp, playbook, …) — they have no + single value field and do not get ``--value``. Returns: click.Group: The configured group with list, get, set, delete commands. @@ -47,7 +54,21 @@ def make_hive_group(group_name: str, hive_name: str, noun_singular: str, noun_pl explain_list = f"List all {noun_plural} stored in the '{hive_name}' hive." explain_get = f"Get a specific {noun_singular} by its key name from the '{hive_name}' hive." - explain_set = f"Create or update {article} {noun_singular} in the '{hive_name}' hive. Provide data via --input-file (JSON/YAML) or stdin." + value_hint = ( + f"Or use --value to set the {noun_singular} value directly " + f"(wrapped as {{data: {{{value_key}: }}}}); --value exposes the value " + f"on the command line and in shell history, so stdin or --input-file is the " + f"recommended path for humans. " + if value_key else "" + ) + explain_set = ( + f"Create or update {article} {noun_singular} in the '{hive_name}' hive. " + f"Provide data via --input-file (JSON/YAML) or stdin. " + f"{value_hint}" + f"--tag (repeatable) and --comment populate usr_mtd. " + f"New hive records default to disabled — pass --enabled to create-and-enable in one shot, " + f"or include usr_mtd.enabled: true in the input file." + ) explain_delete = f"Delete {article} {noun_singular} from the '{hive_name}' hive. Requires --confirm for safety." @click.group(group_name) @@ -74,40 +95,75 @@ def get_cmd(ctx, key) -> None: record = hive.get(key) _output(ctx, record.to_dict()) + def _value_option(fn): + # Only hives with a single scalar value field (value_key set, e.g. + # the secret hive) expose --value; structured-data hives do not, so + # there is no misleading flag and no hardcoded data-key assumption. + if value_key is None: + return fn + return click.option( + "--value", default=None, + help=f"Set the {noun_singular} value directly (wraps into {{data: {{{value_key}: }}}}). " + "WARNING: this exposes the value on the command line and in shell history; " + "stdin or --input-file remains the recommended path for humans.", + )(fn) + @grp.command("set", help=f"Create or update {article} {noun_singular}.") @click.option("--key", required=True, help="Record key name.") @click.option("--input-file", type=click.Path(exists=True), default=None, help="JSON or YAML file with record data.") + @_value_option + @click.option("--tag", "tags", multiple=True, help="Tag to set in usr_mtd (repeatable).") + @click.option("--comment", default=None, help="Set usr_mtd.comment on the record.") + @click.option( + "--enabled/--disabled", "enabled", default=None, + help=f"Set usr_mtd.enabled on the {noun_singular}. Overrides any value in the input file. Records default to disabled if neither this flag nor usr_mtd.enabled is provided.", + ) @pass_context - def set_cmd(ctx, key, input_file) -> None: - if input_file: - with open(input_file, "r") as f: - content = f.read() - elif not sys.stdin.isatty(): - content = sys.stdin.read() - else: - raise click.UsageError("Provide data via --input-file or pipe to stdin.") - - # Parse input as YAML first (YAML is a superset of JSON) - try: - data = yaml.safe_load(content) - except Exception: - data = json.loads(content) - - # Support the same format as 'hive set': if the input has a - # "data" key, use it as the record data and extract usr_mtd. - if isinstance(data, dict) and "data" in data: - # Build a raw dict matching the API format so HiveRecord - # picks up usr_mtd and etag correctly. - raw = { - "data": data["data"], - "usr_mtd": data.get("usr_mtd", {}), - "sys_mtd": {}, - } - if data.get("etag"): - raw["sys_mtd"]["etag"] = data["etag"] - record = HiveRecord.from_raw(key, raw) + def set_cmd(ctx, key, input_file, tags, comment, enabled, value=None) -> None: + if value is not None: + if input_file: + raise click.UsageError("--value is mutually exclusive with --input-file/stdin.") + # Convenience wrapper so a single-value record (value_key set) can + # be set without a file or stdin. --value is explicit intent, so + # stdin is ignored in this mode rather than consulted. + record = HiveRecord(key, data={value_key: value}) else: - record = HiveRecord(key, data=data) + if input_file: + with open(input_file, "r") as f: + content = f.read() + elif not sys.stdin.isatty(): + content = sys.stdin.read() + else: + hint = "--value, --input-file, or pipe to stdin" if value_key else "--input-file or pipe to stdin" + raise click.UsageError(f"Provide data via {hint}.") + + # Parse input as YAML first (YAML is a superset of JSON) + try: + data = yaml.safe_load(content) + except Exception: + data = json.loads(content) + + # Support the same format as 'hive set': if the input has a + # "data" key, use it as the record data and extract usr_mtd. + if isinstance(data, dict) and "data" in data: + # Build a raw dict matching the API format so HiveRecord + # picks up usr_mtd and etag correctly. + raw = { + "data": data["data"], + "usr_mtd": data.get("usr_mtd", {}), + "sys_mtd": {}, + } + if data.get("etag"): + raw["sys_mtd"]["etag"] = data["etag"] + record = HiveRecord.from_raw(key, raw) + else: + record = HiveRecord(key, data=data) + if tags: + record.tags = list(tags) + if comment is not None: + record.comment = comment + if enabled is not None: + record.enabled = enabled org = _get_org(ctx) hive = Hive(org, hive_name) result = hive.set(record) @@ -147,6 +203,57 @@ def disable_cmd(ctx, key) -> None: result = hive.set(record) _output(ctx, result) + @grp.group("tag", help=f"Manage tags on {noun_plural}.") + def tag_group() -> None: + pass + + def _merge_tags(existing: list[str] | None, changes: tuple[str, ...], remove: bool) -> list[str]: + if remove: + to_remove = {t.lower() for t in changes} + return [t for t in (existing or []) if t.lower() not in to_remove] + seen: dict[str, str] = {} + for tag in (existing or []) + list(changes): + k = tag.lower() + if k not in seen: + seen[k] = tag + return list(seen.values()) + + @tag_group.command("set", help=f"Replace all tags on {article} {noun_singular}.") + @click.option("--key", required=True, help="Record key name.") + @click.option("--tag", "-t", "tags", multiple=True, required=True, help="Tag value (repeatable).") + @pass_context + def tag_set_cmd(ctx, key, tags) -> None: + org = _get_org(ctx) + hive = Hive(org, hive_name) + record = hive.get_metadata(key) + record.tags = list(tags) + result = hive.set(record) + _output(ctx, result) + + @tag_group.command("add", help=f"Add tags to {article} {noun_singular} (merged with existing).") + @click.option("--key", required=True, help="Record key name.") + @click.option("--tag", "-t", "tags", multiple=True, required=True, help="Tag value to add (repeatable).") + @pass_context + def tag_add_cmd(ctx, key, tags) -> None: + org = _get_org(ctx) + hive = Hive(org, hive_name) + record = hive.get_metadata(key) + record.tags = _merge_tags(record.tags, tags, remove=False) + result = hive.set(record) + _output(ctx, result) + + @tag_group.command("rm", help=f"Remove tags from {article} {noun_singular}.") + @click.option("--key", required=True, help="Record key name.") + @click.option("--tag", "-t", "tags", multiple=True, required=True, help="Tag value to remove (repeatable).") + @pass_context + def tag_rm_cmd(ctx, key, tags) -> None: + org = _get_org(ctx) + hive = Hive(org, hive_name) + record = hive.get_metadata(key) + record.tags = _merge_tags(record.tags, tags, remove=True) + result = hive.set(record) + _output(ctx, result) + # Register explain texts. register_explain(f"{group_name}.list", explain_list) register_explain(f"{group_name}.get", explain_get) @@ -154,5 +261,8 @@ def disable_cmd(ctx, key) -> None: register_explain(f"{group_name}.delete", explain_delete) register_explain(f"{group_name}.enable", f"Enable {article} {noun_singular} by key (sets usr_mtd.enabled to true).") register_explain(f"{group_name}.disable", f"Disable {article} {noun_singular} by key (sets usr_mtd.enabled to false).") + register_explain(f"{group_name}.tag.set", f"Replace all tags on {article} {noun_singular} (fetch metadata, set tags).") + register_explain(f"{group_name}.tag.add", f"Add tags to {article} {noun_singular}, merged additively with existing tags.") + register_explain(f"{group_name}.tag.rm", f"Remove tags from {article} {noun_singular}, keeping the rest.") return grp diff --git a/limacharlie/commands/adapter.py b/limacharlie/commands/adapter.py index a24093dc..ca901117 100644 --- a/limacharlie/commands/adapter.py +++ b/limacharlie/commands/adapter.py @@ -3,9 +3,13 @@ from __future__ import annotations from ._hive_shortcut import make_hive_group +from ._adapter_types import add_list_types, add_schema, add_sensors from ..discovery import register_explain group = make_hive_group("external-adapter", "external_adapter", "external adapter") +add_list_types(group, "external-adapter.list-types", "external_adapter") +add_schema(group, "external-adapter.schema", "external_adapter") +add_sensors(group, "external-adapter.sensors", "external_adapter") # Override the generic hive explains with adapter-specific documentation. @@ -16,8 +20,9 @@ Each record contains the adapter type and its connection settings. Common adapter types: syslog, file, s3, gcs, pubsub, webhook, stdin, -office365, 1password, crowdstrike, carbon_black, duo, sophos, and many -others. +office365, 1password, crowdstrike, carbon_black, duo, sophos, +threatlocker, and more. Run 'limacharlie external-adapter list-types' +for the full, up-to-date list. Use --output json for the full config including connection details. """) diff --git a/limacharlie/commands/ai.py b/limacharlie/commands/ai.py index 2817c130..13a28878 100644 --- a/limacharlie/commands/ai.py +++ b/limacharlie/commands/ai.py @@ -271,6 +271,10 @@ def summarize_detection(ctx, detection_id) -> None: so on. Any --option flag below overrides the matching field from the template; the rest of the template is used as-is. +--definition accepts BOTH a bare record key (e.g. "my-agent") and the +"hive://ai_agent/" URI form (the form the D&R "start ai agent" +action uses); both resolve to the same ai_agent record. + This lets you reuse one ai_agent definition as a starting point and vary only the bits you need per-run (swap the prompt, cap the budget, change the model, add an env var, etc.). @@ -354,7 +358,7 @@ def _parse_env_kv(items: tuple[str, ...]) -> dict[str, str] | None: @group.command("start-session") @click.option("--definition", required=True, - help="Name of the ai_agent hive record to use as template.") + help="ai_agent hive record to use as template. Accepts a bare key (my-agent) or the hive://ai_agent/ URI form.") @click.option("--prompt", default=None, help="Replace the prompt from the definition.") @click.option("--name", default=None, help="Replace the session name.") @click.option("--idempotent-key", default=None, help="Deduplication key for the session.") diff --git a/limacharlie/commands/api_key.py b/limacharlie/commands/api_key.py index 7b4db46d..4472b554 100644 --- a/limacharlie/commands/api_key.py +++ b/limacharlie/commands/api_key.py @@ -56,6 +56,13 @@ def group() -> None: List all API keys in the organization. Each key entry shows the key name, hash, creation date, and associated permissions. +With --output json the result is an OBJECT keyed by key-hash, whose +values carry the key's name and permissions (it is not a list). + +Use --name to filter the result down to the single +matching key entry; the output keeps the same key-hash-keyed shape +(an object with one entry, or an empty object if no key matches). + The actual secret key value is only returned at creation time and cannot be retrieved later. Use --output json to get the full key metadata for auditing purposes. @@ -63,11 +70,24 @@ def group() -> None: register_explain("api-key.list", _EXPLAIN_LIST) +def _key_name(entry: Any) -> str | None: + """Extract the human name from an API key entry value.""" + if isinstance(entry, dict): + # The API has used both 'name' and 'key_name' over time; accept either. + return entry.get("name") or entry.get("key_name") + return None + + @group.command("list") +@click.option("--name", "name", default=None, help="Filter to the single API key with this name (output keeps the key-hash-keyed object shape).") @pass_context -def list_keys(ctx) -> None: +def list_keys(ctx, name) -> None: org = _get_org(ctx) data = org.get_api_keys() + if name is not None and isinstance(data, dict): + # Keep the raw shape (object keyed by key-hash); just narrow it down + # to the matching entry/entries for back-compat with json consumers. + data = {h: v for h, v in data.items() if _key_name(v) == name} _output(ctx, data) @@ -108,8 +128,19 @@ def list_keys(ctx) -> None: "--permissions", required=True, help="Comma-separated list of permissions (e.g., 'dr.list,dr.set').", ) +@click.option( + "--store-secret", "store_secret", default=None, + help="Also store the freshly-minted key value into the secret hive under this " + "key name ({data: {secret: }}), so it can be referenced as " + "hive://secret/. The value is written directly without transiting " + "intermediate files — collapses the mint -> store -> reference chain.", +) +@click.option( + "--store-secret-tag", "store_secret_tags", multiple=True, + help="Tag to apply to the stored secret record (repeatable). Only used with --store-secret.", +) @pass_context -def create(ctx, name, permissions) -> None: +def create(ctx, name, permissions, store_secret, store_secret_tags) -> None: perm_list = [p.strip() for p in permissions.split(",") if p.strip()] if not perm_list: click.echo("Error: At least one permission is required.", err=True) @@ -120,6 +151,37 @@ def create(ctx, name, permissions) -> None: data = org.add_api_key(name, perm_list) if not ctx.obj.quiet: click.echo(f"API key '{name}' created.") + + if store_secret: + # The key value is only shown once; persist it into the secret hive in + # the same step so callers never have to capture and re-pipe it. + value = data.get("api_key") or data.get("secret") or data.get("key") + if not value: + # The key was already created; surface it so it isn't lost, then fail. + click.echo( + "Error: API key created but no key value was returned to store as a secret.", + err=True, + ) + _output(ctx, data) + ctx.exit(4) + return + from ..sdk.hive import Hive, HiveRecord + secret_hive = Hive(org, "secret") + # If a secret of this name already exists, carry its etag so the write is + # a conditional update (no lost update on a concurrent change) and we can + # report create-vs-overwrite instead of silently clobbering it. + existing_etag = None + try: + existing_etag = secret_hive.get_metadata(store_secret).etag + except Exception: + existing_etag = None # not found -> creating a new secret + record = HiveRecord(store_secret, data={"secret": value}, + tags=list(store_secret_tags), enabled=True, etag=existing_etag) + secret_hive.set(record) + if not ctx.obj.quiet: + verb = "Updated existing" if existing_etag else "Stored key value in new" + click.echo(f"{verb} secret '{store_secret}' (reference it as hive://secret/{store_secret}).") + _output(ctx, data) diff --git a/limacharlie/commands/app.py b/limacharlie/commands/app.py new file mode 100644 index 00000000..18eb214c --- /dev/null +++ b/limacharlie/commands/app.py @@ -0,0 +1,62 @@ +"""App commands for LimaCharlie CLI v2.""" + +from __future__ import annotations + +from ._hive_shortcut import make_hive_group +from ..discovery import register_explain + +group = make_hive_group("app", "app", "app") + +# Override the generic hive explains with app-specific documentation. + +register_explain("app.list", """\ +List all apps stored in the 'app' hive. An app is a user-authored, +AI-generated mini web application: a single self-contained HTML +document that the LimaCharlie web UI renders inside a sandboxed +