Extensible Security-Event Pipeline: Rule Suppression and Plugin-Based Enrichment

Context

Modern cloud security platforms ingest massive streams of telemetry events (e.g., login attempts, file downloads, API key usage). Before triggering an alert, each event must pass through a processing pipeline that dynamically enriches the event with contextual threat intelligence and evaluates suppression rules to filter out known false positives or benign behaviors.

Different tenants (organizations) require different configurations: some enrichers and suppression rules apply globally, while others are tenant-specific. In addition, new enrichers and rules must be pluggable at runtime without requiring core pipeline redeployments.

Requirements

Implement an extensible, in-memory pipeline engine consisting of:

1. Event: An object containing at least tenant_id: str, event_type: str, timestamp: int, and attributes: dict.
2. Enricher (Interface/Base Class): Defines a method enrich(event: Event) -> None that mutates the event's attributes in-place or records additional metadata.
3. SuppressionRule (Interface/Base Class): Defines a method should_suppress(event: Event) -> bool that returns True if the event should be dropped/suppressed.
4. SecurityPipeline: Manages registration and execution of enrichers and suppression rules:
   • register_enricher(enricher: Enricher, tenant_id: Optional[str] = None) -> None: Registers an enricher. If tenant_id is None, it applies globally to all tenants. Otherwise, it only applies to the specified tenant.
   • register_suppression_rule(rule: SuppressionRule, tenant_id: Optional[str] = None) -> None: Registers a suppression rule globally or per tenant.
   • process(event: Event) -> Optional[Event]: Executes all applicable global enrichers followed by tenant-specific enrichers. Then, evaluates global suppression rules followed by tenant suppression rules. If any suppression rule returns True, processing stops and None is returned. If no suppression rule triggers, return the enriched Event.

Error Handling & Edge Cases

• If an enricher raises an exception, the pipeline should log or capture the error in event.attributes["pipeline_errors"] and continue processing rather than crashing the pipeline.
• Order of enricher execution should preserve the order in which they were registered.
• Suppression rules should be evaluated in order; the first rule that returns True short-circuits further suppression evaluations.

Example

class IPReputationEnricher:
    def enrich(self, event: Event) -> None:
        ip = event.attributes.get("source_ip")
        if ip == "192.168.1.100":
            event.attributes["is_internal"] = True
        else:
            event.attributes["is_internal"] = False

class InternalIPSuppressionRule:
    def should_suppress(self, event: Event) -> bool:
        return event.attributes.get("is_internal") is True

pipeline = SecurityPipeline()
pipeline.register_enricher(IPReputationEnricher())  # Global
pipeline.register_suppression_rule(InternalIPSuppressionRule(), tenant_id="tenant_a")

event_1 = Event(
    tenant_id="tenant_a",
    event_type="login",
    timestamp=1600000000,
    attributes={"source_ip": "192.168.1.100"}
)

result_1 = pipeline.process(event_1)
# Output: None (Enriched with is_internal=True, then suppressed by tenant_a rule)

event_2 = Event(
    tenant_id="tenant_b",
    event_type="login",
    timestamp=1600000001,
    attributes={"source_ip": "192.168.1.100"}
)

result_2 = pipeline.process(event_2)
# Output: Event with attributes: {"source_ip": "192.168.1.100", "is_internal": True}
# (Suppression rule was only registered for tenant_a)