diff --git a/limacharlie/commands/_adapter_types.py b/limacharlie/commands/_adapter_types.py index 3ccb50c0..b25ee675 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,15 +38,6 @@ 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. @@ -123,32 +72,25 @@ def _types_from_schema(schema: Any) -> list[str] | 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. - 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} 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 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..206ab851 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,37 @@ 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) + 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: + def __init__(self, org, hive_name): + pass + def get_schema(self): + return {"schema": {"$ref": "#/$defs/R", "$defs": {"R": {"properties": { + "threatlocker": {}, "webhook": {}, "sensor_type": {}, + }}}}} + + with patch.object(at, "Hive", _FakeHive): + rows = at.adapter_types(None) types = {r["type"] for r in rows} - assert "threatlocker" in types - assert "webhook" in types + assert types == {"threatlocker", "webhook"} + + 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 +457,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 +471,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