Skip to content

Remove _FALLBACK_ADAPTER_TYPES curated fallback #307

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
maximelb merged 2 commits into from
Jun 20, 2026
Merged

Remove _FALLBACK_ADAPTER_TYPES curated fallback #307

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0742618
Remove _FALLBACK_ADAPTER_TYPES curated fallback
maximelb Jun 19, 2026
5e552cd
Drop the now-empty description column from list-types
maximelb Jun 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 14 additions & 72 deletions limacharlie/commands/_adapter_types.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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("#/"):
Expand All @@ -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.

Expand Down Expand Up @@ -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
Expand Down
57 changes: 49 additions & 8 deletions tests/unit/test_cli_ergonomics.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
import json
from unittest.mock import patch, MagicMock

import pytest
from click.testing import CliRunner

from limacharlie.cli import cli
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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()
Expand All @@ -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
Expand Down