Skip to content

Commit 7fcd555

Browse files
stevefulme1claude
stevefulme1
and
claude
authored
feat(eda): add webhook, polling, and Kafka event source plugins (#11)
* feat(eda): add webhook, polling, and Kafka event source plugins Add three EDA source plugins for event-driven automation: - webhook: receive Edwin AI alerts via HTTP webhook - alerts: poll Edwin AI API for active alerts - kafka: consume alert events from Kafka topics All sources emit a consistent normalized event schema under the 'edwin_ai' key. Includes example rulebooks for each source. * fix(eda): address static analysis security findings - Validate portal name against allowlist pattern to prevent SSRF - Add explicit verify=True on requests calls for SSL cert validation - Validate webhook bind host and port inputs - Add payload size limit to webhook endpoint - Validate deserialized payload types before processing - Use logger.exception() instead of logging raw exception strings - Add minimum polling interval guard Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * fix: harden existing modules for static analysis compliance - Add portal name validation to _rest_methods.py to prevent SSRF - Add explicit timeout and verify=True on all requests calls - Add raise_for_status() to auth token request - Import _validate_portal in query_api.py module Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1 parent c257254 commit 7fcd555

9 files changed

Lines changed: 589 additions & 7 deletions

File tree

‎extensions/eda/plugins/event_source/__init__.py‎

Whitespace-only changes.

‎extensions/eda/plugins/event_source/alerts.py‎

Lines changed: 174 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,174 @@
1+
"""Poll Edwin AI / LogicMonitor for new alerts.
2+
3+
Periodically queries the Edwin AI API for active alerts and emits
4+
new or changed alerts as events. Tracks seen alerts to avoid
5+
duplicate events.
6+
7+
Arguments:
8+
portal: Edwin AI portal name (e.g. 'mycompany')
9+
access_id: API access ID
10+
access_key: API access key
11+
interval: Polling interval in seconds (default: 60)
12+
severity: Minimum severity to emit (default: 'warning')
13+
resource_group: Optional resource group filter
14+
"""
15+
16+
import asyncio
17+
import logging
18+
import re
19+
from typing import Any
20+
21+
IMPORT_ERRORS = []
22+
try:
23+
import requests
24+
except ImportError as ie:
25+
IMPORT_ERRORS.append(ie)
26+
27+
logger = logging.getLogger(__name__)
28+
29+
SEVERITY_LEVELS = {
30+
"critical": 4,
31+
"error": 3,
32+
"warning": 2,
33+
"info": 1,
34+
}
35+
36+
_VALID_PORTAL = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9._-]{0,62}$")
37+
38+
39+
def _validate_portal(portal: str) -> str:
40+
if not _VALID_PORTAL.match(portal):
41+
raise ValueError(
42+
f"Invalid portal name: {portal!r}. "
43+
"Must be alphanumeric with hyphens/dots/underscores."
44+
)
45+
return portal
46+
47+
48+
async def main(queue: asyncio.Queue, args: dict[str, Any]) -> None:
49+
"""Poll Edwin AI for alerts and forward to the EDA rulebook."""
50+
for exc in IMPORT_ERRORS:
51+
raise exc
52+
53+
portal = _validate_portal(args["portal"])
54+
access_id = args["access_id"]
55+
access_key = args["access_key"]
56+
interval = int(args.get("interval", 60))
57+
if interval < 10:
58+
raise ValueError("Polling interval must be at least 10 seconds")
59+
min_severity = args.get("severity", "warning").lower()
60+
resource_group = args.get("resource_group", "")
61+
min_level = SEVERITY_LEVELS.get(min_severity, 2)
62+
63+
seen_alerts: dict[str, str] = {}
64+
65+
while True:
66+
try:
67+
token = _get_token(portal, access_id, access_key)
68+
alerts = _fetch_alerts(portal, token, resource_group)
69+
70+
current_ids = set()
71+
for alert in alerts:
72+
alert_id = str(alert.get("id", ""))
73+
current_ids.add(alert_id)
74+
severity = alert.get(
75+
"severity", alert.get("level", "info")
76+
).lower()
77+
level = SEVERITY_LEVELS.get(severity, 1)
78+
79+
if level < min_level:
80+
continue
81+
82+
status_key = f"{alert_id}:{alert.get('status', '')}"
83+
if seen_alerts.get(alert_id) == status_key:
84+
continue
85+
seen_alerts[alert_id] = status_key
86+
87+
event = _normalize_event(alert)
88+
await queue.put(event)
89+
90+
cleared = set(seen_alerts.keys()) - current_ids
91+
for alert_id in cleared:
92+
await queue.put({
93+
"edwin_ai": {
94+
"event_type": "alert_cleared",
95+
"alert_id": alert_id,
96+
"status": "cleared",
97+
}
98+
})
99+
del seen_alerts[alert_id]
100+
101+
except Exception:
102+
logger.exception("Error polling Edwin AI alerts")
103+
104+
await asyncio.sleep(interval)
105+
106+
107+
def _get_token(portal: str, access_id: str, access_key: str) -> str:
108+
portal = _validate_portal(portal)
109+
response = requests.post(
110+
f"https://{portal}.dexda.ai/auth/token",
111+
data={
112+
"grant_type": "client_credentials",
113+
"client_id": access_id,
114+
"client_secret": access_key,
115+
},
116+
timeout=30,
117+
verify=True,
118+
)
119+
response.raise_for_status()
120+
return response.json()["access_token"]
121+
122+
123+
def _fetch_alerts(
124+
portal: str, token: str, resource_group: str = ""
125+
) -> list[dict]:
126+
portal = _validate_portal(portal)
127+
url = f"https://{portal}.dexda.ai/api/v1/alerts"
128+
params: dict[str, Any] = {"status": "active", "size": 100}
129+
if resource_group:
130+
params["resourceGroup"] = resource_group
131+
132+
response = requests.get(
133+
url,
134+
headers={"Authorization": f"Bearer {token}"},
135+
params=params,
136+
timeout=30,
137+
verify=True,
138+
)
139+
response.raise_for_status()
140+
data = response.json()
141+
return data.get("items", data.get("data", []))
142+
143+
144+
def _normalize_event(alert: dict) -> dict:
145+
return {
146+
"edwin_ai": {
147+
"event_type": alert.get("type", "alert"),
148+
"alert_id": str(alert.get("id", "")),
149+
"severity": alert.get(
150+
"severity", alert.get("level", "unknown")
151+
),
152+
"host": alert.get(
153+
"host",
154+
alert.get("resourceName", alert.get("device", "")),
155+
),
156+
"message": alert.get(
157+
"message",
158+
alert.get("description", alert.get("summary", "")),
159+
),
160+
"metric": alert.get("metric", alert.get("datapoint", "")),
161+
"value": alert.get("value", alert.get("currentValue", "")),
162+
"threshold": alert.get("threshold", ""),
163+
"resource_group": alert.get(
164+
"resourceGroup", alert.get("group", "")
165+
),
166+
"timestamp": alert.get(
167+
"timestamp", alert.get("startEpoch", "")
168+
),
169+
"status": alert.get(
170+
"status", alert.get("alertStatus", "active")
171+
),
172+
"raw": alert,
173+
}
174+
}

‎extensions/eda/plugins/event_source/kafka.py‎

Lines changed: 123 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,123 @@
1+
"""Consume Edwin AI events from Apache Kafka.
2+
3+
For high-volume enterprise deployments where Edwin AI publishes
4+
alert events to a Kafka topic. The EDA controller consumes from
5+
the configured topic and emits normalized events.
6+
7+
Arguments:
8+
topic: Kafka topic to consume from (default: 'edwin-ai-alerts')
9+
bootstrap_servers: Kafka bootstrap servers (default: 'localhost:9092')
10+
group_id: Consumer group ID (default: 'eda-edwin-ai')
11+
security_protocol: Kafka security protocol (default: 'PLAINTEXT')
12+
sasl_mechanism: SASL mechanism if using SASL auth
13+
sasl_username: SASL username
14+
sasl_password: SASL password
15+
ssl_cafile: Path to CA certificate file
16+
"""
17+
18+
import asyncio
19+
import json
20+
import logging
21+
from typing import Any
22+
23+
IMPORT_ERRORS = []
24+
try:
25+
from aiokafka import AIOKafkaConsumer
26+
except ImportError as ie:
27+
IMPORT_ERRORS.append(ie)
28+
29+
logger = logging.getLogger(__name__)
30+
31+
32+
async def main(queue: asyncio.Queue, args: dict[str, Any]) -> None:
33+
"""Consume Edwin AI events from Kafka and forward to the EDA rulebook."""
34+
for exc in IMPORT_ERRORS:
35+
raise ImportError(
36+
"The 'aiokafka' package is required for the Kafka event source. "
37+
"Install it with: pip install aiokafka"
38+
) from exc
39+
40+
topic = args.get("topic", "edwin-ai-alerts")
41+
bootstrap_servers = args.get("bootstrap_servers", "localhost:9092")
42+
group_id = args.get("group_id", "eda-edwin-ai")
43+
44+
consumer_kwargs: dict[str, Any] = {
45+
"bootstrap_servers": bootstrap_servers,
46+
"group_id": group_id,
47+
"auto_offset_reset": "latest",
48+
"enable_auto_commit": True,
49+
"value_deserializer": lambda m: json.loads(m.decode("utf-8")),
50+
}
51+
52+
security_protocol = args.get("security_protocol", "PLAINTEXT")
53+
consumer_kwargs["security_protocol"] = security_protocol
54+
55+
if args.get("sasl_mechanism"):
56+
consumer_kwargs["sasl_mechanism"] = args["sasl_mechanism"]
57+
consumer_kwargs["sasl_plain_username"] = args.get(
58+
"sasl_username", ""
59+
)
60+
consumer_kwargs["sasl_plain_password"] = args.get(
61+
"sasl_password", ""
62+
)
63+
64+
if args.get("ssl_cafile"):
65+
consumer_kwargs["ssl_cafile"] = args["ssl_cafile"]
66+
67+
consumer = AIOKafkaConsumer(topic, **consumer_kwargs)
68+
69+
await consumer.start()
70+
logger.info(
71+
"Edwin AI Kafka consumer started on topic '%s' at %s",
72+
topic,
73+
bootstrap_servers,
74+
)
75+
76+
try:
77+
async for message in consumer:
78+
try:
79+
data = message.value
80+
if isinstance(data, str):
81+
data = json.loads(data)
82+
if not isinstance(data, dict):
83+
logger.warning("Skipping non-dict Kafka message")
84+
continue
85+
event = _normalize_event(data)
86+
await queue.put(event)
87+
except (json.JSONDecodeError, TypeError):
88+
logger.warning("Skipping malformed Kafka message")
89+
finally:
90+
await consumer.stop()
91+
92+
93+
def _normalize_event(data: dict) -> dict:
94+
return {
95+
"edwin_ai": {
96+
"event_type": data.get("type", data.get("eventType", "alert")),
97+
"alert_id": str(data.get("alertId", data.get("id", ""))),
98+
"severity": data.get(
99+
"severity", data.get("level", "unknown")
100+
),
101+
"host": data.get(
102+
"host",
103+
data.get("resourceName", data.get("device", "")),
104+
),
105+
"message": data.get(
106+
"message",
107+
data.get("description", data.get("summary", "")),
108+
),
109+
"metric": data.get("metric", data.get("datapoint", "")),
110+
"value": data.get("value", data.get("currentValue", "")),
111+
"threshold": data.get("threshold", ""),
112+
"resource_group": data.get(
113+
"resourceGroup", data.get("group", "")
114+
),
115+
"timestamp": data.get(
116+
"timestamp", data.get("startEpoch", "")
117+
),
118+
"status": data.get(
119+
"status", data.get("alertStatus", "active")
120+
),
121+
"raw": data,
122+
}
123+
}

0 commit comments

Comments
 (0)