From 07426187476750993934767c14778b04f07b0c1b Mon Sep 17 00:00:00 2001 From: Maxime Lamothe-Brassard Date: Fri, 19 Jun 2026 16:53:04 -0700 Subject: [PATCH 1/2] Remove _FALLBACK_ADAPTER_TYPES curated fallback adapter list-types now derives the supported type list (and per-type description) solely from the live cloud_sensor/external_adapter hive JSON-Schema. The hard-coded fallback only ever masked backend downtime, during which every other command fails anyway, so a stale list would only mislead. If the schema can't be fetched the error now surfaces. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01EfrssJQLhNEFhvaUr1EPzv --- limacharlie/commands/_adapter_types.py | 102 +++++++------------------ tests/unit/test_cli_ergonomics.py | 63 ++++++++++++--- 2 files changed, 81 insertions(+), 84 deletions(-) diff --git a/limacharlie/commands/_adapter_types.py b/limacharlie/commands/_adapter_types.py index 3ccb50c0..a560335c 100644 --- a/limacharlie/commands/_adapter_types.py +++ b/limacharlie/commands/_adapter_types.py @@ -3,8 +3,7 @@ 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. +stale relative to the backend. """ from __future__ import annotations @@ -27,47 +26,6 @@ "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("#/"): @@ -80,17 +38,8 @@ def _resolve_ref(root: dict, ref: str) -> dict | None: 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. +def _types_from_schema(schema: Any) -> dict[str, str] | None: + """Derive adapter types 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 @@ -98,8 +47,11 @@ def _types_from_schema(schema: Any) -> list[str] | None: (``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. + helper structs (``ClientOptions``, ``AckBufferOptions``, …). + + Returns a ``{type_name: description}`` mapping (description taken from the + schema property's own ``description``, blank if it has none), or ``None`` + if the schema does not expose any usable type names. """ if isinstance(schema, dict) and isinstance(schema.get("schema"), dict): schema = schema["schema"] @@ -119,36 +71,36 @@ def _types_from_schema(schema: Any) -> list[str] | None: return None names = {n for n in props if n and not n.startswith("_")} - _NON_TYPE_FIELDS - return sorted(names) or None + if not names: + return None + out: dict[str, str] = {} + for name in sorted(names): + node = props[name] + desc = node.get("description", "") if isinstance(node, dict) else "" + out[name] = desc if isinstance(desc, str) else "" + return out 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 solely from the live hive schema so it always tracks the backend. """ - 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()) - ] + schema = Hive(org, hive_name).get_schema() + derived = _types_from_schema(schema) + if not derived: + raise click.ClickException( + f"The '{hive_name}' hive schema did not advertise any adapter types." + ) + return [{"type": name, "description": desc} for name, desc in derived.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 list is derived from the live adapter hive JSON-Schema (so it +tracks the backend). 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 diff --git a/tests/unit/test_cli_ergonomics.py b/tests/unit/test_cli_ergonomics.py index ce3ca301..3ca5cde2 100644 --- a/tests/unit/test_cli_ergonomics.py +++ b/tests/unit/test_cli_ergonomics.py @@ -9,6 +9,7 @@ import json from unittest.mock import patch, MagicMock +import pytest from click.testing import CliRunner from limacharlie.cli import cli @@ -374,13 +375,41 @@ def test_detect_with_input_file_errors(self, mock_hive_cls, _org, _client, tmp_p # --------------------------------------------------------------------------- class TestAdapterListTypes: - def test_fallback_includes_threatlocker(self): - from limacharlie.commands._adapter_types import adapter_types - # Passing an org whose schema fetch raises forces the fallback path. - rows = adapter_types(None) - types = {r["type"] for r in rows} - assert "threatlocker" in types - assert "webhook" in types + def test_derived_carries_description(self): + # adapter_types pulls the type list (and per-type description) straight + # from the live hive schema — there is no hard-coded fallback. + from limacharlie.commands import _adapter_types as at + + class _FakeHive: + def __init__(self, org, hive_name): + pass + def get_schema(self): + return {"schema": {"$ref": "#/$defs/R", "$defs": {"R": {"properties": { + "threatlocker": {"description": "ThreatLocker unified audit"}, + "webhook": {}, + "sensor_type": {}, + }}}}} + + with patch.object(at, "Hive", _FakeHive): + rows = at.adapter_types(None) + by_type = {r["type"]: r["description"] for r in rows} + assert by_type["threatlocker"] == "ThreatLocker unified audit" + assert by_type["webhook"] == "" + assert "sensor_type" not in by_type + + def test_raises_when_schema_unavailable(self): + # No fallback: a failing schema fetch must surface, not return a stale list. + from limacharlie.commands import _adapter_types as at + + class _FailHive: + def __init__(self, org, hive_name): + pass + def get_schema(self): + raise Exception("no schema") + + with patch.object(at, "Hive", _FailHive): + with pytest.raises(Exception): + at.adapter_types(None) def test_derived_from_ref_rooted_schema(self): # Mirrors the real reflected shape: a {"schema": {...}} wrapper whose @@ -432,9 +461,11 @@ def get_schema(self): @patch("limacharlie.commands._adapter_types.Hive") @patch("limacharlie.commands._adapter_types.Client", create=True) def test_cli_list_types(self, _client, mock_hive_cls): - # When the schema fetch fails, the command still returns the curated list. + # The command lists the types advertised by the live hive schema. mock_hive = MagicMock() - mock_hive.get_schema.side_effect = Exception("no schema") + mock_hive.get_schema.return_value = {"schema": {"$ref": "#/$defs/R", "$defs": {"R": { + "properties": {"threatlocker": {}, "s3": {}, "sensor_type": {}}, + }}}} mock_hive_cls.return_value = mock_hive runner = CliRunner() @@ -444,6 +475,20 @@ def test_cli_list_types(self, _client, mock_hive_cls): parsed = json.loads(result.output) assert any(r["type"] == "threatlocker" for r in parsed) + @patch("limacharlie.commands._adapter_types.Hive") + @patch("limacharlie.commands._adapter_types.Client", create=True) + def test_cli_list_types_fails_without_schema(self, _client, mock_hive_cls): + # No fallback: if the schema can't be fetched the command errors out + # rather than printing a stale hard-coded list. + mock_hive = MagicMock() + mock_hive.get_schema.side_effect = Exception("no schema") + mock_hive_cls.return_value = mock_hive + + runner = CliRunner() + with patch("limacharlie.client.Client"): + result = runner.invoke(cli, ["--output", "json", "cloud-adapter", "list-types"]) + assert result.exit_code != 0 + # --------------------------------------------------------------------------- # hive schema flat rendering From 5e552cd68dc2b24f86abc518ee8ed6ec2d3b4824 Mon Sep 17 00:00:00 2001 From: Maxime Lamothe-Brassard Date: Fri, 19 Jun 2026 16:56:55 -0700 Subject: [PATCH 2/2] Drop the now-empty description column from list-types The live hive schema does not carry per-type descriptions; descriptions only ever came from the curated fallback dict that this PR removes. Rather than emit a permanently-empty "description" column, list-types now returns just the type name. Reverts _types_from_schema to its plain list return as well. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01EfrssJQLhNEFhvaUr1EPzv --- limacharlie/commands/_adapter_types.py | 26 ++++++++------------------ tests/unit/test_cli_ergonomics.py | 16 ++++++---------- 2 files changed, 14 insertions(+), 28 deletions(-) diff --git a/limacharlie/commands/_adapter_types.py b/limacharlie/commands/_adapter_types.py index a560335c..b25ee675 100644 --- a/limacharlie/commands/_adapter_types.py +++ b/limacharlie/commands/_adapter_types.py @@ -38,8 +38,8 @@ def _resolve_ref(root: dict, ref: str) -> dict | None: return node if isinstance(node, dict) else None -def _types_from_schema(schema: Any) -> dict[str, str] | None: - """Derive adapter types from an adapter hive JSON-Schema. +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 @@ -47,11 +47,8 @@ def _types_from_schema(schema: Any) -> dict[str, str] | None: (``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 a ``{type_name: description}`` mapping (description taken from the - schema property's own ``description``, blank if it has none), or ``None`` - if the schema does not expose any usable type names. + 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"] @@ -71,18 +68,11 @@ def _types_from_schema(schema: Any) -> dict[str, str] | None: return None names = {n for n in props if n and not n.startswith("_")} - _NON_TYPE_FIELDS - if not names: - return None - out: dict[str, str] = {} - for name in sorted(names): - node = props[name] - desc = node.get("description", "") if isinstance(node, dict) else "" - out[name] = desc if isinstance(desc, str) else "" - return out + 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. + """Return the supported adapter types for a hive as name rows. Derived solely from the live hive schema so it always tracks the backend. """ @@ -92,11 +82,11 @@ def adapter_types(org: Any, hive_name: str = "cloud_sensor") -> list[dict[str, s raise click.ClickException( f"The '{hive_name}' hive schema did not advertise any adapter types." ) - return [{"type": name, "description": desc} for name, desc in derived.items()] + return [{"type": name} for name in derived] _EXPLAIN_LIST_TYPES = """\ -List the supported adapter/sensor type names with a short description. +List the supported adapter/sensor type names. The list is derived from the live adapter hive JSON-Schema (so it tracks the backend). Use a type name as the top-level sensor_type when diff --git a/tests/unit/test_cli_ergonomics.py b/tests/unit/test_cli_ergonomics.py index 3ca5cde2..206ab851 100644 --- a/tests/unit/test_cli_ergonomics.py +++ b/tests/unit/test_cli_ergonomics.py @@ -375,9 +375,9 @@ def test_detect_with_input_file_errors(self, mock_hive_cls, _org, _client, tmp_p # --------------------------------------------------------------------------- class TestAdapterListTypes: - def test_derived_carries_description(self): - # adapter_types pulls the type list (and per-type description) straight - # from the live hive schema — there is no hard-coded fallback. + def test_derived_from_live_schema(self): + # adapter_types pulls the type list straight from the live hive schema — + # there is no hard-coded fallback. from limacharlie.commands import _adapter_types as at class _FakeHive: @@ -385,17 +385,13 @@ def __init__(self, org, hive_name): pass def get_schema(self): return {"schema": {"$ref": "#/$defs/R", "$defs": {"R": {"properties": { - "threatlocker": {"description": "ThreatLocker unified audit"}, - "webhook": {}, - "sensor_type": {}, + "threatlocker": {}, "webhook": {}, "sensor_type": {}, }}}}} with patch.object(at, "Hive", _FakeHive): rows = at.adapter_types(None) - by_type = {r["type"]: r["description"] for r in rows} - assert by_type["threatlocker"] == "ThreatLocker unified audit" - assert by_type["webhook"] == "" - assert "sensor_type" not in by_type + types = {r["type"] for r in rows} + assert types == {"threatlocker", "webhook"} def test_raises_when_schema_unavailable(self): # No fallback: a failing schema fetch must surface, not return a stale list.