Python API

The generated API reference is intentionally limited to supported plugin and model extension surfaces. Search commands are documented separately because their public interface is argparse-based.

Crawler plugins

Base class for crawler plugins.

BaseCrawlerPlugin

Bases: ABC

Abstract base class for crawler plugins. Plugins should inherit from this and implement the run method.

Source code in crawlers/plugins/base.py

class BaseCrawlerPlugin(ABC):
    """
    Abstract base class for crawler plugins.
    Plugins should inherit from this and implement the run method.
    """

    def __init__(self, config=None):
        self.config = config or {}
        self.name = self.config.get('name', self.__class__.__name__)

    @abstractmethod
    def run(self):
        """
        Run the crawler logic.
        For daemon mode, should loop continuously.
        For scheduled, run once.
        """
        raise NotImplementedError("Subclasses must implement the run method")

    def get_queryset(self, model_class=None, **filters):
        """
        Helper to get a queryset from the configured database.
        """
        from events.models import Event
        model = model_class or Event
        db_alias = self.config.get('db_alias', 'default')
        return model.objects.using(db_alias).filter(**filters)

    def create_finding(self, event, rule_name, description, severity='medium', mitre_tactic=None, mitre_technique=None):
        """
        Create a finding for an event.
        Checks for cooldown based on config before creating.
        """
        from crawlers.models import Finding
        cooldown = self.config.get('realert_cooldown')
        if not Finding.can_create_finding(event, rule_name, cooldown):
            logger.debug("Skipping finding creation due to cooldown: %s for event %s", rule_name, event.id)
            return None

        finding = Finding.objects.create(
            event=event,
            rule_name=rule_name,
            description=description,
            severity=severity,
            mitre_tactic=mitre_tactic,
            mitre_technique=mitre_technique,
        )
        logger.info("Created finding: %s", finding)
        self.send_alerts(finding)
        return finding

    def send_alerts(self, finding):
        """
        Send alerts for the finding using configured alerting plugins.
        """
        alerting_plugins = self.config.get('alerting_plugins', [])
        if not alerting_plugins:
            return
        from django.conf import settings
        alerting_plugin_paths = getattr(settings, 'ALERTING_PLUGINS', [])
        alerting_configs = getattr(settings, 'ALERTING_CONFIGS', {})
        # Load plugin classes if not already
        if not hasattr(self, '_alerting_classes'):
            import importlib
            self._alerting_classes = {}
            for plugin_path in alerting_plugin_paths:
                try:
                    module_path, class_name = plugin_path.rsplit('.', 1)
                    module = importlib.import_module(module_path)
                    plugin_class = getattr(module, class_name)
                    name = getattr(plugin_class, 'name', class_name.lower())
                    self._alerting_classes[name] = plugin_class
                except Exception as e:
                    logger.error(f"Failed to load alerting plugin {plugin_path}: {e}")
        for plugin_name in alerting_plugins:
            if plugin_name in self._alerting_classes:
                config = alerting_configs.get(plugin_name, {})
                plugin_instance = self._alerting_classes[plugin_name](config)
                try:
                    plugin_instance.send_alert(finding)
                except Exception as e:
                    logger.error(f"Failed to send alert with {plugin_name}: {e}")
            else:
                logger.warning(f"Alerting plugin {plugin_name} not loaded")

create_finding(event, rule_name, description, severity='medium', mitre_tactic=None, mitre_technique=None)

Create a finding for an event. Checks for cooldown based on config before creating.

Source code in crawlers/plugins/base.py

def create_finding(self, event, rule_name, description, severity='medium', mitre_tactic=None, mitre_technique=None):
    """
    Create a finding for an event.
    Checks for cooldown based on config before creating.
    """
    from crawlers.models import Finding
    cooldown = self.config.get('realert_cooldown')
    if not Finding.can_create_finding(event, rule_name, cooldown):
        logger.debug("Skipping finding creation due to cooldown: %s for event %s", rule_name, event.id)
        return None

    finding = Finding.objects.create(
        event=event,
        rule_name=rule_name,
        description=description,
        severity=severity,
        mitre_tactic=mitre_tactic,
        mitre_technique=mitre_technique,
    )
    logger.info("Created finding: %s", finding)
    self.send_alerts(finding)
    return finding

get_queryset(model_class=None, **filters)

Helper to get a queryset from the configured database.

Source code in crawlers/plugins/base.py

def get_queryset(self, model_class=None, **filters):
    """
    Helper to get a queryset from the configured database.
    """
    from events.models import Event
    model = model_class or Event
    db_alias = self.config.get('db_alias', 'default')
    return model.objects.using(db_alias).filter(**filters)

run() abstractmethod

Run the crawler logic. For daemon mode, should loop continuously. For scheduled, run once.

Source code in crawlers/plugins/base.py

@abstractmethod
def run(self):
    """
    Run the crawler logic.
    For daemon mode, should loop continuously.
    For scheduled, run once.
    """
    raise NotImplementedError("Subclasses must implement the run method")

send_alerts(finding)

Send alerts for the finding using configured alerting plugins.

Source code in crawlers/plugins/base.py

def send_alerts(self, finding):
    """
    Send alerts for the finding using configured alerting plugins.
    """
    alerting_plugins = self.config.get('alerting_plugins', [])
    if not alerting_plugins:
        return
    from django.conf import settings
    alerting_plugin_paths = getattr(settings, 'ALERTING_PLUGINS', [])
    alerting_configs = getattr(settings, 'ALERTING_CONFIGS', {})
    # Load plugin classes if not already
    if not hasattr(self, '_alerting_classes'):
        import importlib
        self._alerting_classes = {}
        for plugin_path in alerting_plugin_paths:
            try:
                module_path, class_name = plugin_path.rsplit('.', 1)
                module = importlib.import_module(module_path)
                plugin_class = getattr(module, class_name)
                name = getattr(plugin_class, 'name', class_name.lower())
                self._alerting_classes[name] = plugin_class
            except Exception as e:
                logger.error(f"Failed to load alerting plugin {plugin_path}: {e}")
    for plugin_name in alerting_plugins:
        if plugin_name in self._alerting_classes:
            config = alerting_configs.get(plugin_name, {})
            plugin_instance = self._alerting_classes[plugin_name](config)
            try:
                plugin_instance.send_alert(finding)
            except Exception as e:
                logger.error(f"Failed to send alert with {plugin_name}: {e}")
        else:
            logger.warning(f"Alerting plugin {plugin_name} not loaded")

Bases: Model

Model to store findings generated by crawlers.

Source code in crawlers/models.py

class Finding(models.Model):
    """
    Model to store findings generated by crawlers.
    """
    class Status(models.TextChoices):
        NEW = 'new', 'New'
        ACKNOWLEDGED = 'acknowledged', 'Acknowledged'
        IN_PROGRESS = 'in_progress', 'In progress'
        RESOLVED = 'resolved', 'Resolved'
        FALSE_POSITIVE = 'false_positive', 'False positive'

    @classmethod
    def actionable_statuses(cls):
        """Return statuses that require preservation of the linked event."""
        return (
            cls.Status.NEW,
            cls.Status.ACKNOWLEDGED,
            cls.Status.IN_PROGRESS,
        )

    event = models.ForeignKey(Event, on_delete=models.CASCADE, related_name='findings')
    rule_name = models.CharField(max_length=255)
    description = models.TextField()
    severity = models.CharField(max_length=50, choices=[
        ('low', 'Low'),
        ('medium', 'Medium'),
        ('high', 'High'),
        ('critical', 'Critical'),
    ], default='medium')
    mitre_tactic = models.CharField(max_length=255, blank=True, null=True)
    mitre_technique = models.CharField(max_length=255, blank=True, null=True)
    status = models.CharField(
        max_length=20,
        choices=Status.choices,
        default=Status.NEW,
        db_index=True,
    )
    assignee = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        blank=True,
        null=True,
        on_delete=models.SET_NULL,
        related_name='assigned_findings',
    )
    notes = models.TextField(blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    def __str__(self):
        return f"{self.rule_name}: {self.description[:50]}"

    @classmethod
    def can_create_finding(cls, event, rule_name, cooldown_seconds=None):
        """
        Check if a finding can be created for the given event and rule within the cooldown period.
        Returns True if no recent finding exists, False otherwise.
        """
        if cooldown_seconds is None:
            return True
        from django.utils import timezone
        from datetime import timedelta
        cooldown_since = timezone.now() - timedelta(seconds=cooldown_seconds)
        return not cls.objects.filter(
            event=event,
            rule_name=rule_name,
            created_at__gte=cooldown_since
        ).exists()

    class Meta:
        ordering = ['-created_at']

actionable_statuses() classmethod

Return statuses that require preservation of the linked event.

Source code in crawlers/models.py

@classmethod
def actionable_statuses(cls):
    """Return statuses that require preservation of the linked event."""
    return (
        cls.Status.NEW,
        cls.Status.ACKNOWLEDGED,
        cls.Status.IN_PROGRESS,
    )

can_create_finding(event, rule_name, cooldown_seconds=None) classmethod

Check if a finding can be created for the given event and rule within the cooldown period. Returns True if no recent finding exists, False otherwise.

Source code in crawlers/models.py

@classmethod
def can_create_finding(cls, event, rule_name, cooldown_seconds=None):
    """
    Check if a finding can be created for the given event and rule within the cooldown period.
    Returns True if no recent finding exists, False otherwise.
    """
    if cooldown_seconds is None:
        return True
    from django.utils import timezone
    from datetime import timedelta
    cooldown_since = timezone.now() - timedelta(seconds=cooldown_seconds)
    return not cls.objects.filter(
        event=event,
        rule_name=rule_name,
        created_at__gte=cooldown_since
    ).exists()

Agent plugins

Cross-platform host security posture inventory.

HostSecurityPosturePlugin

Collect stable host inventory and best-effort security-control state.

Portable data comes from Python and psutil. Small platform adapters add firewall, disk-encryption, secure-boot, and endpoint-protection state when the operating system exposes it. Missing tools or insufficient privileges are represented as unknown instead of stopping the collector.

Source code in agent/plugins/host_security_posture_plugin.py

class HostSecurityPosturePlugin:
    """Collect stable host inventory and best-effort security-control state.

    Portable data comes from Python and psutil. Small platform adapters add
    firewall, disk-encryption, secure-boot, and endpoint-protection state when
    the operating system exposes it. Missing tools or insufficient privileges
    are represented as ``unknown`` instead of stopping the collector.
    """

    def __init__(self, config, event_queue, stop_event):
        self.event_queue = event_queue
        self.stop_event = stop_event
        self.poll_interval = float(config.get('poll_interval', 900.0))
        self.status_interval = float(config.get('status_interval', 3600.0))
        self.command_timeout = float(config.get('command_timeout', 10.0))
        if self.poll_interval <= 0:
            raise ValueError('poll_interval must be greater than zero')
        if self.status_interval <= 0:
            raise ValueError('status_interval must be greater than zero')
        if self.command_timeout <= 0:
            raise ValueError('command_timeout must be greater than zero')

        self.collect_local_accounts = bool(
            config.get('collect_local_accounts', True)
        )
        self.index = config.get('index', 'host_security_posture')
        self.host = config.get('host', socket.gethostname())
        self.source = config.get('source', 'host_security_posture')
        self.sourcetype = config.get('sourcetype', 'json')
        self.db_alias = config.get('db_alias')

        self._last_snapshot = {}
        self._last_status_emit = None
        self._last_health = None

    @staticmethod
    def _is_privileged():
        if os.name == 'nt':
            try:
                import ctypes

                return bool(ctypes.windll.shell32.IsUserAnAdmin())
            except (AttributeError, OSError):
                return None
        geteuid = getattr(os, 'geteuid', None)
        return geteuid() == 0 if geteuid is not None else None

    def _collect_host_identity(self):
        uname = platform.uname()
        return {
            'hostname': socket.gethostname(),
            'fqdn': socket.getfqdn(),
            'os': uname.system,
            'os_release': uname.release,
            'os_version': uname.version,
            'architecture': uname.machine,
            'processor': uname.processor or None,
            'boot_time': psutil.boot_time(),
            'timezone': datetime.now().astimezone().tzname(),
            'agent_user': getpass.getuser(),
            'agent_privileged': self._is_privileged(),
        }

    @staticmethod
    def _address_family_name(family):
        if family == socket.AF_INET:
            return 'ipv4'
        if family == socket.AF_INET6:
            return 'ipv6'
        if family == psutil.AF_LINK:
            return 'mac'
        return str(getattr(family, 'name', family)).lower()

    def _collect_network_interfaces(self):
        addresses = psutil.net_if_addrs()
        stats = psutil.net_if_stats()
        interfaces = []
        for name in sorted(set(addresses) | set(stats)):
            interface_stats = stats.get(name)
            interface_addresses = []
            for address in addresses.get(name, []):
                interface_addresses.append({
                    'family': self._address_family_name(address.family),
                    'address': address.address,
                    'netmask': address.netmask,
                    'broadcast': address.broadcast,
                    'ptp': address.ptp,
                })
            interface_addresses.sort(
                key=lambda item: (item['family'], item['address'] or '')
            )
            duplex = None
            if interface_stats is not None:
                duplex_names = {
                    psutil.NIC_DUPLEX_FULL: 'full',
                    psutil.NIC_DUPLEX_HALF: 'half',
                    psutil.NIC_DUPLEX_UNKNOWN: 'unknown',
                }
                duplex = duplex_names.get(interface_stats.duplex, 'unknown')
            interfaces.append({
                'name': name,
                'is_up': interface_stats.isup if interface_stats else None,
                'duplex': duplex,
                'speed_mbps': interface_stats.speed if interface_stats else None,
                'mtu': interface_stats.mtu if interface_stats else None,
                'addresses': interface_addresses,
            })
        return interfaces

    @staticmethod
    def _collect_user_sessions():
        sessions = []
        for user in psutil.users():
            sessions.append({
                'username': user.name,
                'terminal': user.terminal,
                'remote_host': user.host,
                'started': user.started,
                'pid': getattr(user, 'pid', None),
            })
        return sorted(
            sessions,
            key=lambda item: (
                item['username'] or '',
                item['terminal'] or '',
                item['started'] or 0,
            ),
        )

    @staticmethod
    def _collect_filesystems():
        filesystems = []
        for partition in psutil.disk_partitions(all=False):
            filesystems.append({
                'device': partition.device,
                'mountpoint': partition.mountpoint,
                'filesystem': partition.fstype,
                'options': sorted(
                    option
                    for option in (partition.opts or '').split(',')
                    if option
                ),
            })
        return sorted(
            filesystems,
            key=lambda item: (item['mountpoint'], item['device']),
        )

    def _run(self, command):
        options = {
            'capture_output': True,
            'text': True,
            'errors': 'replace',
            'timeout': self.command_timeout,
            'check': False,
        }
        if os.name == 'nt':
            options['creationflags'] = getattr(
                subprocess,
                'CREATE_NO_WINDOW',
                0,
            )
        completed = subprocess.run(command, **options)
        if completed.returncode != 0:
            raise OSError(
                f'{command[0]} exited with status {completed.returncode}'
            )
        return completed.stdout.strip()

    def _powershell_json(self, script):
        executable = shutil.which('powershell.exe') or shutil.which('pwsh.exe')
        if not executable:
            raise FileNotFoundError('PowerShell is unavailable')
        output = self._run([
            executable,
            '-NoLogo',
            '-NoProfile',
            '-NonInteractive',
            '-Command',
            script,
        ])
        return json.loads(output) if output else None

    def _collect_windows_accounts(self):
        script = r'''
$items = @(Get-LocalUser | ForEach-Object {
    [ordered]@{
        username = $_.Name
        enabled = [bool]$_.Enabled
        sid = [string]$_.SID
        last_logon = if ($_.LastLogon) { $_.LastLogon.ToString('o') } else { $null }
        password_expires = if ($_.PasswordExpires) { $_.PasswordExpires.ToString('o') } else { $null }
        password_required = [bool]$_.PasswordRequired
        user_may_change_password = [bool]$_.UserMayChangePassword
    }
})
ConvertTo-Json -InputObject $items -Depth 4 -Compress
'''
        result = self._powershell_json(script)
        if result is None:
            return []
        accounts = result if isinstance(result, list) else [result]
        return sorted(accounts, key=lambda item: item.get('username') or '')

    @staticmethod
    def _collect_posix_accounts():
        import pwd

        accounts = []
        for account in pwd.getpwall():
            accounts.append({
                'username': account.pw_name,
                'uid': account.pw_uid,
                'gid': account.pw_gid,
                'home': account.pw_dir,
                'shell': account.pw_shell,
                'system_account': account.pw_uid < 1000,
            })
        return sorted(accounts, key=lambda item: (item['uid'], item['username']))

    def _collect_local_account_inventory(self):
        if platform.system() == 'Windows':
            return self._collect_windows_accounts()
        return self._collect_posix_accounts()

    def _collect_windows_controls(self):
        script = r'''
$result = [ordered]@{}
try {
    $result.firewall = [ordered]@{
        state = 'available'
        profiles = @(Get-NetFirewallProfile | ForEach-Object {
            [ordered]@{
                name = $_.Name
                enabled = [bool]$_.Enabled
                default_inbound = [string]$_.DefaultInboundAction
                default_outbound = [string]$_.DefaultOutboundAction
            }
        })
    }
} catch { $result.firewall = [ordered]@{ state = 'unknown' } }
if (Get-Command Get-MpComputerStatus -ErrorAction SilentlyContinue) {
    try {
        $mp = Get-MpComputerStatus
        $result.endpoint_protection = [ordered]@{
            provider = 'Microsoft Defender'
            state = 'available'
            antivirus_enabled = [bool]$mp.AntivirusEnabled
            antispyware_enabled = [bool]$mp.AntispywareEnabled
            realtime_protection_enabled = [bool]$mp.RealTimeProtectionEnabled
            behavior_monitor_enabled = [bool]$mp.BehaviorMonitorEnabled
            signatures_last_updated = if ($mp.AntivirusSignatureLastUpdated) { $mp.AntivirusSignatureLastUpdated.ToString('o') } else { $null }
        }
    } catch { $result.endpoint_protection = [ordered]@{ provider = 'Microsoft Defender'; state = 'unknown' } }
} else { $result.endpoint_protection = [ordered]@{ state = 'unsupported' } }
try {
    $result.secure_boot = [ordered]@{
        state = 'available'
        enabled = [bool](Confirm-SecureBootUEFI -ErrorAction Stop)
    }
} catch { $result.secure_boot = [ordered]@{ state = 'unknown'; enabled = $null } }
if (Get-Command Get-BitLockerVolume -ErrorAction SilentlyContinue) {
    try {
        $result.disk_encryption = [ordered]@{
            provider = 'BitLocker'
            state = 'available'
            volumes = @(Get-BitLockerVolume | ForEach-Object {
                [ordered]@{
                    mount_point = $_.MountPoint
                    volume_status = [string]$_.VolumeStatus
                    protection_status = [string]$_.ProtectionStatus
                    encryption_method = [string]$_.EncryptionMethod
                    encryption_percentage = $_.EncryptionPercentage
                }
            })
        }
    } catch { $result.disk_encryption = [ordered]@{ provider = 'BitLocker'; state = 'unknown' } }
} else { $result.disk_encryption = [ordered]@{ provider = 'BitLocker'; state = 'unsupported' } }
ConvertTo-Json -InputObject $result -Depth 7 -Compress
'''
        controls = self._powershell_json(script) or {}
        firewall = controls.get('firewall') or {}
        if isinstance(firewall.get('profiles'), list):
            firewall['profiles'].sort(key=lambda item: item.get('name') or '')
        encryption = controls.get('disk_encryption') or {}
        if isinstance(encryption.get('volumes'), list):
            encryption['volumes'].sort(
                key=lambda item: item.get('mount_point') or ''
            )
        return controls

    def _linux_firewall(self):
        if shutil.which('ufw'):
            try:
                output = self._run(['ufw', 'status'])
                first_line = output.splitlines()[0] if output else ''
                status_value = first_line.partition(':')[2].strip().lower()
                return {
                    'provider': 'ufw',
                    'state': 'enabled' if status_value == 'active' else 'disabled',
                }
            except (OSError, subprocess.TimeoutExpired):
                return {'provider': 'ufw', 'state': 'unknown'}
        if shutil.which('firewall-cmd'):
            try:
                output = self._run(['firewall-cmd', '--state'])
                return {
                    'provider': 'firewalld',
                    'state': 'enabled' if output.lower() == 'running' else output.lower(),
                }
            except (OSError, subprocess.TimeoutExpired):
                return {'provider': 'firewalld', 'state': 'unknown'}
        if shutil.which('nft'):
            try:
                output = self._run(['nft', 'list', 'ruleset'])
                return {
                    'provider': 'nftables',
                    'state': 'configured' if output else 'empty',
                }
            except (OSError, subprocess.TimeoutExpired):
                return {'provider': 'nftables', 'state': 'unknown'}
        return {'provider': None, 'state': 'unsupported'}

    @staticmethod
    def _linux_secure_boot():
        paths = glob.glob('/sys/firmware/efi/efivars/SecureBoot-*')
        if not paths:
            return {'state': 'unsupported', 'enabled': None}
        try:
            with open(paths[0], 'rb') as secure_boot:
                value = secure_boot.read(5)
            return {
                'state': 'available',
                'enabled': len(value) >= 5 and value[4] == 1,
            }
        except OSError:
            return {'state': 'unknown', 'enabled': None}

    def _linux_disk_encryption(self):
        if not shutil.which('lsblk'):
            return {'provider': 'LUKS', 'state': 'unsupported'}
        try:
            output = self._run([
                'lsblk',
                '--json',
                '--output',
                'NAME,TYPE,FSTYPE,MOUNTPOINTS',
            ])
            devices = json.loads(output).get('blockdevices', [])
        except (OSError, subprocess.TimeoutExpired, json.JSONDecodeError):
            return {'provider': 'LUKS', 'state': 'unknown'}

        encrypted = []

        def walk(items):
            for item in items:
                if str(item.get('fstype') or '').lower() == 'crypto_luks':
                    encrypted.append(item.get('name'))
                walk(item.get('children') or [])

        walk(devices)
        return {
            'provider': 'LUKS',
            'state': 'detected' if encrypted else 'not_detected',
            'encrypted_devices': sorted(filter(None, encrypted)),
        }

    def _collect_linux_controls(self):
        return {
            'firewall': self._linux_firewall(),
            'secure_boot': self._linux_secure_boot(),
            'disk_encryption': self._linux_disk_encryption(),
            'endpoint_protection': {'state': 'not_assessed'},
        }

    def _macos_check(self, command, enabled_text):
        if not os.path.exists(command[0]):
            return {'state': 'unsupported', 'enabled': None}
        try:
            output = self._run(command)
            return {
                'state': 'available',
                'enabled': enabled_text in output.lower(),
                'summary': output.splitlines()[0] if output else '',
            }
        except (OSError, subprocess.TimeoutExpired):
            return {'state': 'unknown', 'enabled': None}

    def _collect_macos_controls(self):
        return {
            'firewall': self._macos_check(
                [
                    '/usr/libexec/ApplicationFirewall/socketfilterfw',
                    '--getglobalstate',
                ],
                'enabled',
            ),
            'disk_encryption': self._macos_check(
                ['/usr/bin/fdesetup', 'status'],
                'filevault is on',
            ),
            'gatekeeper': self._macos_check(
                ['/usr/sbin/spctl', '--status'],
                'assessments enabled',
            ),
            'secure_boot': {'state': 'not_assessed', 'enabled': None},
            'endpoint_protection': {'state': 'not_assessed'},
        }

    def _collect_security_controls(self):
        system = platform.system()
        if system == 'Windows':
            return self._collect_windows_controls()
        if system == 'Linux':
            return self._collect_linux_controls()
        if system == 'Darwin':
            return self._collect_macos_controls()
        return {
            'firewall': {'state': 'unsupported'},
            'disk_encryption': {'state': 'unsupported'},
            'secure_boot': {'state': 'unsupported'},
            'endpoint_protection': {'state': 'unsupported'},
        }

    def _collect_snapshot(self):
        collectors = [
            ('host_identity', self._collect_host_identity),
            ('network_interfaces', self._collect_network_interfaces),
            ('user_sessions', self._collect_user_sessions),
            ('filesystems', self._collect_filesystems),
            ('security_controls', self._collect_security_controls),
        ]
        if self.collect_local_accounts:
            collectors.append(
                ('local_accounts', self._collect_local_account_inventory)
            )

        snapshot = {}
        issues = []
        for component, collector in collectors:
            try:
                snapshot[component] = collector()
            except (
                OSError,
                psutil.Error,
                subprocess.TimeoutExpired,
                json.JSONDecodeError,
            ) as exc:
                logger.warning(
                    'Host posture component %s failed: %s',
                    component,
                    exc,
                )
                issues.append({
                    'component': component,
                    'error_type': type(exc).__name__,
                })
        return snapshot, issues

    def _queue_event(
        self,
        event_type,
        timestamp,
        component,
        data,
        previous=None,
    ):
        event = {
            'type': 'host_security_posture',
            'event_type': event_type,
            'timestamp': timestamp,
            'component': component,
            'data': data,
            'index': self.index,
            'host': self.host,
            'source': self.source,
            'sourcetype': self.sourcetype,
        }
        if previous is not None:
            event['previous'] = previous
        if self.db_alias:
            event['db_alias'] = self.db_alias
        self.event_queue.put(event)

    def _emit_status(self, timestamp, status):
        health = (
            status['state'],
            tuple(
                (issue['component'], issue['error_type'])
                for issue in status['issues']
            ),
        )
        due = (
            self._last_status_emit is None
            or timestamp - self._last_status_emit >= self.status_interval
        )
        if due or health != self._last_health:
            self._queue_event(
                'collection_status',
                timestamp,
                'collector',
                status,
            )
            self._last_status_emit = timestamp
            self._last_health = health

    def collect_once(self, timestamp=None):
        """Collect and enqueue one posture snapshot or its component diffs."""
        timestamp = time.time() if timestamp is None else timestamp
        started = time.monotonic()
        snapshot, issues = self._collect_snapshot()
        for component, current in snapshot.items():
            if component not in self._last_snapshot:
                self._queue_event(
                    'posture_snapshot',
                    timestamp,
                    component,
                    current,
                )
            elif current != self._last_snapshot[component]:
                self._queue_event(
                    'posture_changed',
                    timestamp,
                    component,
                    current,
                    previous=self._last_snapshot[component],
                )
            self._last_snapshot[component] = current

        self._emit_status(
            timestamp,
            {
                'state': 'partial' if issues else 'ok',
                'components_collected': sorted(snapshot),
                'components_failed': sorted(
                    issue['component'] for issue in issues
                ),
                'issues': issues,
                'collection_duration_ms': round(
                    (time.monotonic() - started) * 1000,
                    3,
                ),
            },
        )

    def run(self):
        logger.info(
            'HostSecurityPosturePlugin started with poll_interval=%s',
            self.poll_interval,
        )
        while not self.stop_event.is_set():
            self.collect_once()
            self.stop_event.wait(self.poll_interval)

collect_once(timestamp=None)

Collect and enqueue one posture snapshot or its component diffs.

Source code in agent/plugins/host_security_posture_plugin.py

def collect_once(self, timestamp=None):
    """Collect and enqueue one posture snapshot or its component diffs."""
    timestamp = time.time() if timestamp is None else timestamp
    started = time.monotonic()
    snapshot, issues = self._collect_snapshot()
    for component, current in snapshot.items():
        if component not in self._last_snapshot:
            self._queue_event(
                'posture_snapshot',
                timestamp,
                component,
                current,
            )
        elif current != self._last_snapshot[component]:
            self._queue_event(
                'posture_changed',
                timestamp,
                component,
                current,
                previous=self._last_snapshot[component],
            )
        self._last_snapshot[component] = current

    self._emit_status(
        timestamp,
        {
            'state': 'partial' if issues else 'ok',
            'components_collected': sorted(snapshot),
            'components_failed': sorted(
                issue['component'] for issue in issues
            ),
            'issues': issues,
            'collection_duration_ms': round(
                (time.monotonic() - started) * 1000,
                3,
            ),
        },
    )

LinuxSchedulersPlugin

Emit diffs for cron and systemd timers/services. Config: poll_interval (default 60).

Source code in agent/plugins/linux_schedulers_plugin.py

class LinuxSchedulersPlugin:
    """
    Emit diffs for cron and systemd timers/services.
    Config: poll_interval (default 60).
    """
    def __init__(self, config, event_queue, stop_event):
        if platform.system() != 'Linux':
            raise RuntimeError('LinuxSchedulersPlugin only runs on Linux')
        self.event_queue = event_queue
        self.stop_event = stop_event
        self.poll_interval = float(config.get('poll_interval', 60))
        self.index=config.get('index','default'); self.host=config.get('host','localhost')
        self.source=config.get('source','linux_schedulers'); self.sourcetype=config.get('sourcetype','json')
        self.db_alias = config.get('db_alias')
        self._last = {'cron':{}, 'systemd':{}}

    def _cron_state(self):
        items = {}
        # system crontab + cron.d
        for path in ['/etc/crontab'] + [os.path.join('/etc/cron.d', f) for f in os.listdir('/etc/cron.d') if os.path.isfile(os.path.join('/etc/cron.d', f))]:
            try:
                with open(path, 'r', encoding='utf-8', errors='replace') as f:
                    items[path] = f.read()
            except Exception: pass
        # per-user crontab
        for udir in ['/var/spool/cron', '/var/spool/cron/crontabs']:
            if os.path.isdir(udir):
                for name in os.listdir(udir):
                    p = os.path.join(udir, name)
                    try:
                        with open(p, 'r', encoding='utf-8', errors='replace') as f:
                            items[f'user:{name}'] = f.read()
                    except Exception: pass
        return items

    def _systemd_state(self):
        def run(cmd): 
            try: return subprocess.check_output(cmd, text=True, errors='replace')
            except Exception: return ''
        timers = run(['systemctl','list-timers','--all','--no-pager','--no-legend'])
        services = run(['systemctl','list-unit-files','--type=service','--no-pager','--no-legend'])
        return {'timers': timers, 'services': services}

    def run(self):
        while not self.stop_event.is_set():
            now = time.time()
            cron = self._cron_state()
            sysd = self._systemd_state()
            cur = {'cron': cron, 'systemd': sysd}
            for key in ['cron','systemd']:
                prev = self._last.get(key,{})
                if cur[key] != prev:
                    event = {'type':'linux_schedulers','event_scope':key,'event_type':'changed',
                             'timestamp':now,'current':cur[key],'previous':prev,
                             'index':self.index,'host':self.host,'source':self.source,'sourcetype':self.sourcetype}
                    if self.db_alias: event['db_alias']=self.db_alias
                    self.event_queue.put(event)
            self._last = cur
            time.sleep(self.poll_interval)

MacOSLaunchdPlugin

Track LaunchAgents/LaunchDaemons; emit adds/removes/changes. Config: poll_interval (default 60).

Source code in agent/plugins/macos_launchd_plugin.py

class MacOSLaunchdPlugin:
    """
    Track LaunchAgents/LaunchDaemons; emit adds/removes/changes.
    Config: poll_interval (default 60).
    """
    PATHS = [
        '/Library/LaunchAgents', '/Library/LaunchDaemons',
        os.path.expanduser('~/Library/LaunchAgents')
    ]

    def __init__(self, config, event_queue, stop_event):
        if platform.system() != 'Darwin':
            raise RuntimeError('MacOSLaunchdPlugin only runs on macOS')
        self.event_queue = event_queue
        self.stop_event = stop_event
        self.poll_interval = float(config.get('poll_interval', 60))
        self.index=config.get('index','default'); self.host=config.get('host','localhost')
        self.source=config.get('source','macos_launchd'); self.sourcetype=config.get('sourcetype','json')
        self.db_alias=config.get('db_alias')
        self._last = {}

    def _hash_plist(self, path):
        try:
            with open(path,'rb') as f: data = f.read()
            return hashlib.sha256(data).hexdigest()
        except Exception: return None

    def _snapshot(self):
        snap = {}
        for base in self.PATHS:
            for p in glob.glob(os.path.join(base, '*.plist')):
                info = {'path': p, 'hash': self._hash_plist(p)}
                try:
                    with open(p,'rb') as f:
                        pl = plistlib.load(f)
                    info.update({
                        'Label': pl.get('Label'),
                        'Program': pl.get('Program'),
                        'ProgramArguments': pl.get('ProgramArguments'),
                        'RunAtLoad': pl.get('RunAtLoad'),
                        'KeepAlive': pl.get('KeepAlive'),
                        'WatchPaths': pl.get('WatchPaths'),
                        'UserName': pl.get('UserName')
                    })
                except Exception:
                    pass
                snap[p] = info
        return snap

    def run(self):
        while not self.stop_event.is_set():
            now = time.time()
            cur = self._snapshot()
            added = {k:v for k,v in cur.items() if k not in self._last}
            removed = {k:v for k,v in self._last.items() if k not in cur}
            changed = {k:v for k,v in cur.items() if k in self._last and v.get('hash') != self._last[k].get('hash')}
            for kind, payload in (('added', added), ('removed', removed), ('changed', changed)):
                if not payload: continue
                event = {'type':'macos_launchd','event_type':kind,'timestamp':now,'items':payload,
                         'index':self.index,'host':self.host,'source':self.source,'sourcetype':self.sourcetype}
                if self.db_alias: event['db_alias']=self.db_alias
                self.event_queue.put(event)
            self._last = cur
            time.sleep(self.poll_interval)

Cross-platform network exposure and connection telemetry.

NetworkSecurityPlugin

Report changes to Internet listeners and active connections.

The collector uses psutil's system-wide connection API so the same code runs on Windows, Linux, and macOS. Process details are best effort because operating-system permissions can prevent attribution for some sockets.

Source code in agent/plugins/network_security_plugin.py

class NetworkSecurityPlugin:
    """Report changes to Internet listeners and active connections.

    The collector uses psutil's system-wide connection API so the same code
    runs on Windows, Linux, and macOS. Process details are best effort because
    operating-system permissions can prevent attribution for some sockets.
    """

    def __init__(self, config, event_queue, stop_event):
        self.event_queue = event_queue
        self.stop_event = stop_event
        self.poll_interval = float(config.get('poll_interval', 30.0))
        self.status_interval = float(config.get('status_interval', 300.0))
        if self.poll_interval <= 0:
            raise ValueError('poll_interval must be greater than zero')
        if self.status_interval <= 0:
            raise ValueError('status_interval must be greater than zero')

        self.include_cmdline = bool(config.get('include_cmdline', False))
        self.index = config.get('index', 'network_security')
        self.host = config.get('host', 'localhost')
        self.source = config.get('source', 'network_security')
        self.sourcetype = config.get('sourcetype', 'json')
        self.db_alias = config.get('db_alias')

        self._last_listeners = {}
        self._last_connections = {}
        self._last_status_emit = None
        self._last_health = None

    @staticmethod
    def _endpoint(value):
        if not value:
            return None
        address = getattr(value, 'ip', value[0])
        port = getattr(value, 'port', value[1])
        return {'address': str(address), 'port': int(port)}

    @staticmethod
    def _address_scope(address):
        if address in ('0.0.0.0', '::'):
            return 'wildcard'
        try:
            parsed = ipaddress.ip_address(address)
        except ValueError:
            return 'unknown'
        if parsed.is_loopback:
            return 'loopback'
        if parsed.is_link_local:
            return 'link_local'
        if parsed.is_multicast:
            return 'multicast'
        if parsed.is_private:
            return 'private'
        if parsed.is_global:
            return 'public'
        return 'special'

    @staticmethod
    def _family_name(family):
        if family == socket.AF_INET:
            return 'ipv4'
        if family == socket.AF_INET6:
            return 'ipv6'
        return str(getattr(family, 'name', family)).lower()

    @staticmethod
    def _protocol_name(socket_type):
        if socket_type == socket.SOCK_STREAM:
            return 'tcp'
        if socket_type == socket.SOCK_DGRAM:
            return 'udp'
        return str(getattr(socket_type, 'name', socket_type)).lower()

    def _process_details(self, pid, cache, counters):
        if pid is None:
            return {
                'process_name': None,
                'process_exe': None,
                'process_user': None,
                'process_cmdline': None,
            }
        if pid in cache:
            return cache[pid]

        details = {
            'process_name': None,
            'process_exe': None,
            'process_user': None,
            'process_cmdline': None,
        }
        inaccessible = False
        try:
            process = psutil.Process(pid)
        except (psutil.AccessDenied, psutil.NoSuchProcess, psutil.ZombieProcess):
            counters['processes_unavailable'] += 1
            cache[pid] = details
            return details

        getters = {
            'process_name': process.name,
            'process_exe': process.exe,
            'process_user': process.username,
        }
        if self.include_cmdline:
            getters['process_cmdline'] = process.cmdline

        for field, getter in getters.items():
            try:
                details[field] = getter()
            except psutil.AccessDenied:
                inaccessible = True
            except (psutil.NoSuchProcess, psutil.ZombieProcess, OSError):
                counters['processes_unavailable'] += 1
                break
        if inaccessible:
            counters['processes_access_denied'] += 1
        cache[pid] = details
        return details

    def _record(self, connection, process_cache, counters):
        local = self._endpoint(connection.laddr)
        remote = self._endpoint(connection.raddr)
        protocol = self._protocol_name(connection.type)
        status = str(connection.status or 'NONE').upper()
        record = {
            'protocol': protocol,
            'address_family': self._family_name(connection.family),
            'local_address': local['address'],
            'local_port': local['port'],
            'local_scope': self._address_scope(local['address']),
            'remote_address': remote['address'] if remote else None,
            'remote_port': remote['port'] if remote else None,
            'remote_scope': self._address_scope(remote['address']) if remote else None,
            'status': status,
            'pid': connection.pid,
        }
        record.update(self._process_details(connection.pid, process_cache, counters))
        is_listener = (
            status == str(psutil.CONN_LISTEN).upper()
            or (protocol == 'udp' and remote is None)
        )
        return record, is_listener

    @staticmethod
    def _identity(record, listener):
        values = [
            record['protocol'],
            record['address_family'],
            record['local_address'],
            record['local_port'],
            record['pid'],
        ]
        if not listener:
            values.extend((record['remote_address'], record['remote_port']))
        return tuple(values)

    def _snapshot(self):
        listeners = {}
        connections = {}
        process_cache = {}
        counters = {
            'processes_access_denied': 0,
            'processes_unavailable': 0,
        }
        for connection in psutil.net_connections(kind='inet'):
            if not connection.laddr:
                continue
            record, is_listener = self._record(
                connection,
                process_cache,
                counters,
            )
            if is_listener:
                listeners[self._identity(record, True)] = record
            elif connection.raddr:
                connections[self._identity(record, False)] = record
        return listeners, connections, counters

    def _queue_event(self, event_type, timestamp, data, previous=None):
        event = {
            'type': 'network_security',
            'event_type': event_type,
            'timestamp': timestamp,
            'data': data,
            'index': self.index,
            'host': self.host,
            'source': self.source,
            'sourcetype': self.sourcetype,
        }
        if previous is not None:
            event['previous'] = previous
        if self.db_alias:
            event['db_alias'] = self.db_alias
        self.event_queue.put(event)

    def _emit_changes(self, timestamp, current, previous, event_names):
        added_name, removed_name, changed_name = event_names
        for identity in sorted(current.keys() - previous.keys(), key=repr):
            self._queue_event(added_name, timestamp, current[identity])
        for identity in sorted(previous.keys() - current.keys(), key=repr):
            self._queue_event(removed_name, timestamp, previous[identity])
        for identity in sorted(current.keys() & previous.keys(), key=repr):
            if current[identity] != previous[identity]:
                self._queue_event(
                    changed_name,
                    timestamp,
                    current[identity],
                    previous=previous[identity],
                )

    def _emit_status(self, timestamp, status):
        health = (
            status['state'],
            status.get('error'),
            status.get('processes_access_denied', 0) > 0,
            status.get('processes_unavailable', 0) > 0,
        )
        due = (
            self._last_status_emit is None
            or timestamp - self._last_status_emit >= self.status_interval
        )
        if due or health != self._last_health:
            self._queue_event('collection_status', timestamp, status)
            self._last_status_emit = timestamp
            self._last_health = health

    def collect_once(self, timestamp=None):
        """Collect and enqueue one snapshot. Exposed for deterministic tests."""
        timestamp = time.time() if timestamp is None else timestamp
        started = time.monotonic()
        try:
            listeners, connections, counters = self._snapshot()
        except (psutil.Error, OSError) as exc:
            logger.warning('Network connection collection failed: %s', exc)
            self._emit_status(
                timestamp,
                {
                    'state': 'error',
                    'error': f'{type(exc).__name__}: {exc}',
                    'collection_duration_ms': round(
                        (time.monotonic() - started) * 1000,
                        3,
                    ),
                },
            )
            return

        self._emit_changes(
            timestamp,
            listeners,
            self._last_listeners,
            ('listener_added', 'listener_removed', 'listener_changed'),
        )
        self._emit_changes(
            timestamp,
            connections,
            self._last_connections,
            ('connection_opened', 'connection_closed', 'connection_changed'),
        )
        self._last_listeners = listeners
        self._last_connections = connections

        state = 'partial' if any(counters.values()) else 'ok'
        self._emit_status(
            timestamp,
            {
                'state': state,
                'listener_count': len(listeners),
                'connection_count': len(connections),
                'processes_access_denied': counters['processes_access_denied'],
                'processes_unavailable': counters['processes_unavailable'],
                'include_cmdline': self.include_cmdline,
                'collection_duration_ms': round(
                    (time.monotonic() - started) * 1000,
                    3,
                ),
            },
        )

    def run(self):
        logger.info(
            'NetworkSecurityPlugin started with poll_interval=%s',
            self.poll_interval,
        )
        while not self.stop_event.is_set():
            self.collect_once()
            self.stop_event.wait(self.poll_interval)

collect_once(timestamp=None)

Collect and enqueue one snapshot. Exposed for deterministic tests.

Source code in agent/plugins/network_security_plugin.py

def collect_once(self, timestamp=None):
    """Collect and enqueue one snapshot. Exposed for deterministic tests."""
    timestamp = time.time() if timestamp is None else timestamp
    started = time.monotonic()
    try:
        listeners, connections, counters = self._snapshot()
    except (psutil.Error, OSError) as exc:
        logger.warning('Network connection collection failed: %s', exc)
        self._emit_status(
            timestamp,
            {
                'state': 'error',
                'error': f'{type(exc).__name__}: {exc}',
                'collection_duration_ms': round(
                    (time.monotonic() - started) * 1000,
                    3,
                ),
            },
        )
        return

    self._emit_changes(
        timestamp,
        listeners,
        self._last_listeners,
        ('listener_added', 'listener_removed', 'listener_changed'),
    )
    self._emit_changes(
        timestamp,
        connections,
        self._last_connections,
        ('connection_opened', 'connection_closed', 'connection_changed'),
    )
    self._last_listeners = listeners
    self._last_connections = connections

    state = 'partial' if any(counters.values()) else 'ok'
    self._emit_status(
        timestamp,
        {
            'state': state,
            'listener_count': len(listeners),
            'connection_count': len(connections),
            'processes_access_denied': counters['processes_access_denied'],
            'processes_unavailable': counters['processes_unavailable'],
            'include_cmdline': self.include_cmdline,
            'collection_duration_ms': round(
                (time.monotonic() - started) * 1000,
                3,
            ),
        },
    )

Plugin process manager for agent plugins. Handles plugin lifecycle, authentication, and process management.

PluginProcessManager

Manage the lifecycle of plugin processes, including starting, stopping, and monitoring their status. Also manages the sender process for communicating with the indexer.

Source code in agent/plugins/plugin_process_manager.py

class PluginProcessManager:
    """
    Manage the lifecycle of plugin processes, including starting, stopping,
    and monitoring their status. Also manages the sender process for
    communicating with the indexer.
    """
    def __init__(self, plugin_path, config, indexer_cfg, credentials=None):
        self.plugin_path = plugin_path
        self.config = config
        self.indexer_cfg = indexer_cfg
        self.credentials = credentials
        self.restart_limit = config.get('restart', 3)
        self.child_processes = []
        self.restart_attempts = {}  # key by plugin_path
        self.event_queue = multiprocessing.Queue()
        self.sender_proc = None
        self.stop_event = multiprocessing.Event()  # add stop_event
        logger.debug(f"PluginProcessManager initialized for {plugin_path} with config {config}")

    def start(self):
        """
        Start the plugin process and the sender process.
        """
        logger.info(f"Starting plugin process for {self.plugin_path}")
        proc = multiprocessing.Process(target=run_plugin, args=(self.plugin_path, self.config, self.event_queue, self.stop_event))
        proc.start()
        logger.info(f"Started plugin process with PID {proc.pid}")
        self.child_processes.append(proc)
        self.restart_attempts[self.plugin_path] = 0  # initialize attempts by plugin_path
        # Start sender process
        if not self.sender_proc or not self.sender_proc.is_alive():
            logger.info("Starting sender process")
            self.sender_proc = multiprocessing.Process(target=sender_process, args=(self.event_queue, self.indexer_cfg, self.credentials))
            self.sender_proc.start()
            logger.info(f"Started sender process with PID {self.sender_proc.pid}")

    def check_and_restart(self):
        """
        Check the status of child processes and restart them if they are not alive.
        """
        for proc in list(self.child_processes):
            if not proc.is_alive():
                attempts = self.restart_attempts.get(self.plugin_path, 0)  # get attempts by plugin_path
                logger.warning(f"Plugin process PID {proc.pid} is not alive. Restart attempts: {attempts}")
                if attempts < self.restart_limit:
                    new_proc = multiprocessing.Process(target=run_plugin, args=(self.plugin_path, self.config, self.event_queue, self.stop_event))
                    new_proc.start()
                    logger.info(f"Restarted plugin process with new PID {new_proc.pid}")
                    self.child_processes.append(new_proc)
                    self.restart_attempts[self.plugin_path] = attempts + 1  # increment by plugin_path
                else:
                    logger.error(f"Restart limit reached for plugin process PID {proc.pid}")
                self.child_processes.remove(proc)
        # Restart sender if needed
        if self.sender_proc and not self.sender_proc.is_alive():
            logger.warning(f"Sender process PID {self.sender_proc.pid} is not alive. Attempting restart.")
            self.sender_proc = multiprocessing.Process(target=sender_process, args=(self.event_queue, self.indexer_cfg, self.credentials))
            self.sender_proc.start()
            logger.info(f"Restarted sender process with PID {self.sender_proc.pid}")

    def children_alive(self):
        """
        Check if child processes are alive.
        """
        alive = sum([p.is_alive() for p in self.child_processes])
        if self.sender_proc and self.sender_proc.is_alive():
            alive += 1
        logger.debug(f"children_alive: {alive} (plugin processes: {len(self.child_processes)}, sender alive: {self.sender_proc.is_alive() if self.sender_proc else False})")
        return alive

    def stop(self):
        """
        Stop the plugin processes gracefully.
        """
        logger.info(f"Stopping plugin processes for {self.plugin_path}")
        self.stop_event.set()
        for proc in self.child_processes:
            if proc.is_alive():
                proc.join(timeout=5)
                if proc.is_alive():
                    logger.warning(f"Plugin process {proc.pid} did not stop gracefully, terminating")
                    proc.terminate()
        if self.sender_proc and self.sender_proc.is_alive():
            self.sender_proc.terminate()  # sender might need to be terminated as it has its own loop
        logger.info(f"Stopped all processes for {self.plugin_path}")

check_and_restart()

Check the status of child processes and restart them if they are not alive.

Source code in agent/plugins/plugin_process_manager.py

def check_and_restart(self):
    """
    Check the status of child processes and restart them if they are not alive.
    """
    for proc in list(self.child_processes):
        if not proc.is_alive():
            attempts = self.restart_attempts.get(self.plugin_path, 0)  # get attempts by plugin_path
            logger.warning(f"Plugin process PID {proc.pid} is not alive. Restart attempts: {attempts}")
            if attempts < self.restart_limit:
                new_proc = multiprocessing.Process(target=run_plugin, args=(self.plugin_path, self.config, self.event_queue, self.stop_event))
                new_proc.start()
                logger.info(f"Restarted plugin process with new PID {new_proc.pid}")
                self.child_processes.append(new_proc)
                self.restart_attempts[self.plugin_path] = attempts + 1  # increment by plugin_path
            else:
                logger.error(f"Restart limit reached for plugin process PID {proc.pid}")
            self.child_processes.remove(proc)
    # Restart sender if needed
    if self.sender_proc and not self.sender_proc.is_alive():
        logger.warning(f"Sender process PID {self.sender_proc.pid} is not alive. Attempting restart.")
        self.sender_proc = multiprocessing.Process(target=sender_process, args=(self.event_queue, self.indexer_cfg, self.credentials))
        self.sender_proc.start()
        logger.info(f"Restarted sender process with PID {self.sender_proc.pid}")

children_alive()

Check if child processes are alive.

Source code in agent/plugins/plugin_process_manager.py

def children_alive(self):
    """
    Check if child processes are alive.
    """
    alive = sum([p.is_alive() for p in self.child_processes])
    if self.sender_proc and self.sender_proc.is_alive():
        alive += 1
    logger.debug(f"children_alive: {alive} (plugin processes: {len(self.child_processes)}, sender alive: {self.sender_proc.is_alive() if self.sender_proc else False})")
    return alive

start()

Start the plugin process and the sender process.

Source code in agent/plugins/plugin_process_manager.py

def start(self):
    """
    Start the plugin process and the sender process.
    """
    logger.info(f"Starting plugin process for {self.plugin_path}")
    proc = multiprocessing.Process(target=run_plugin, args=(self.plugin_path, self.config, self.event_queue, self.stop_event))
    proc.start()
    logger.info(f"Started plugin process with PID {proc.pid}")
    self.child_processes.append(proc)
    self.restart_attempts[self.plugin_path] = 0  # initialize attempts by plugin_path
    # Start sender process
    if not self.sender_proc or not self.sender_proc.is_alive():
        logger.info("Starting sender process")
        self.sender_proc = multiprocessing.Process(target=sender_process, args=(self.event_queue, self.indexer_cfg, self.credentials))
        self.sender_proc.start()
        logger.info(f"Started sender process with PID {self.sender_proc.pid}")

stop()

Stop the plugin processes gracefully.

Source code in agent/plugins/plugin_process_manager.py

def stop(self):
    """
    Stop the plugin processes gracefully.
    """
    logger.info(f"Stopping plugin processes for {self.plugin_path}")
    self.stop_event.set()
    for proc in self.child_processes:
        if proc.is_alive():
            proc.join(timeout=5)
            if proc.is_alive():
                logger.warning(f"Plugin process {proc.pid} did not stop gracefully, terminating")
                proc.terminate()
    if self.sender_proc and self.sender_proc.is_alive():
        self.sender_proc.terminate()  # sender might need to be terminated as it has its own loop
    logger.info(f"Stopped all processes for {self.plugin_path}")

get_indexer_transport(indexer_cfg)

Return HTTP/WS schemes plus verification settings for the indexer.

Source code in agent/plugins/plugin_process_manager.py

def get_indexer_transport(indexer_cfg):
    """Return HTTP/WS schemes plus verification settings for the indexer."""
    tls_enabled = bool(indexer_cfg.get('tls', False))
    ca_bundle = indexer_cfg.get('ca_bundle')

    if ca_bundle:
        ca_path = Path(ca_bundle)
        if not ca_path.is_file():
            raise ValueError(f'INDEXER_CA_BUNDLE does not exist: {ca_path}')

    websocket_ssl = None
    if tls_enabled:
        websocket_ssl = ssl.create_default_context(cafile=ca_bundle)

    return {
        'http_scheme': 'https' if tls_enabled else 'http',
        'websocket_scheme': 'wss' if tls_enabled else 'ws',
        # requests verifies against system roots when this is True.
        'requests_verify': ca_bundle or True,
        'websocket_ssl': websocket_ssl,
    }

get_session_cookie(indexer_cfg, credentials)

Authenticate with indexer and return session cookie.

Source code in agent/plugins/plugin_process_manager.py

def get_session_cookie(indexer_cfg, credentials):
    """
    Authenticate with indexer and return session cookie.
    """
    import re
    logger.info(f"get_session_cookie called with indexer_cfg={indexer_cfg}, credentials={'***' if credentials else None}")
    host = indexer_cfg.get('host', 'localhost')
    port = indexer_cfg.get('port', 8000)
    transport = get_indexer_transport(indexer_cfg)
    logger.info(
        "Authenticating to indexer at %s://%s:%s",
        transport['http_scheme'],
        host,
        port,
    )
    login_url = f"{transport['http_scheme']}://{host}:{port}/login/"
    if not credentials:
        raise ValueError("Credentials are required for authentication")
    with requests.Session() as session:
        resp = session.get(login_url, verify=transport['requests_verify'])
        text = resp.text
        csrf_token = session.cookies.get('csrftoken')
        if not csrf_token:
            match = re.search(r'name=["\']csrfmiddlewaretoken["\'] value=["\']([^"\']+)["\']', text)
            if match:
                csrf_token = match.group(1)
        if not csrf_token:
            return None
        payload = {
            'username': credentials['username'],
            'password': credentials['password'],
            'csrfmiddlewaretoken': csrf_token
        }
        login_headers = {
            'Referer': login_url
        }
        resp = session.post(
            login_url,
            data=payload,
            headers=login_headers,
            allow_redirects=False,
            verify=transport['requests_verify'],
        )
        if resp.status_code not in (200, 302):
            logger.error("Login failed with status code %d", resp.status_code)
            return None
        sessionid = session.cookies.get('sessionid')
        if not sessionid:
            return None
        return sessionid

run_plugin(plugin_path, config, event_queue, stop_event)

Run a plugin given its path and config, managing its lifecycle.

Source code in agent/plugins/plugin_process_manager.py

def run_plugin(plugin_path, config, event_queue, stop_event):
    """
    Run a plugin given its path and config, managing its lifecycle.
    """
    logger.info("run_plugin called with path=%s, config=%s", plugin_path, config)
    try:
        module_path, class_name = plugin_path.split(':')
        logger.debug("Importing module %s, class %s", module_path, class_name)
        module = importlib.import_module(module_path)
        plugin_cls = getattr(module, class_name)
        plugin = plugin_cls(config, event_queue, stop_event)  # pass stop_event
        logger.info("Instantiated plugin %s with config %s", plugin_cls, config)
        if hasattr(plugin, 'run'):
            logger.info("Running plugin %s", plugin_cls)
            plugin.run()
        else:
            logger.warning("Plugin %s has no 'run' method, entering keep-alive loop.", plugin_cls)
            while not stop_event.is_set():
                time.sleep(1)
    except Exception as e:
        logger.exception("Exception in run_plugin: %s", e)

sender_process(event_queue, indexer_cfg, credentials=None)

Send events to the indexer via WebSocket.

Source code in agent/plugins/plugin_process_manager.py

def sender_process(event_queue, indexer_cfg, credentials=None):
    """
    Send events to the indexer via WebSocket.
    """
    logger.info(f"sender_process started with indexer_cfg={indexer_cfg}")
    async def send_events():
        host = indexer_cfg.get('host', 'localhost')
        port = indexer_cfg.get('port', 8000)
        transport = get_indexer_transport(indexer_cfg)
        headers = {}
        retry_count = 0
        max_retries = 5
        while retry_count < max_retries:
            sessionid = get_session_cookie(indexer_cfg, credentials)
            if sessionid:
                headers['Cookie'] = f"sessionid={sessionid}"
                break
            else:
                wait_time = 2 ** retry_count  # exponential backoff
                logger.warning(f"Login failed, retrying in {wait_time} seconds (attempt {retry_count + 1}/{max_retries})")
                await asyncio.sleep(wait_time)
                retry_count += 1
        if not sessionid:
            logger.error("Failed to authenticate after retries, sender_process exiting")
            return
        uri = f"{transport['websocket_scheme']}://{host}:{port}/indexer/"
        logger.info(f"Connecting to WebSocket {uri} with headers {headers}")
        connect_options = {'additional_headers': headers}
        if transport['websocket_ssl'] is not None:
            connect_options['ssl'] = transport['websocket_ssl']
        while True:
            try:
                async with websockets.connect(uri, **connect_options) as websocket:
                    logger.info(f"WebSocket connection established to {uri}")
                    while True:
                        # Drain queue with soft limits
                        batch = []
                        deadline = time.time() + 0.5
                        while len(batch) < 500 and time.time() < deadline:
                            try:
                                batch.append(event_queue.get_nowait())
                            except Exception:
                                break
                        if not batch:
                            await asyncio.sleep(0.1)
                            continue
                        # Normalize type
                        for ev in batch:
                            if 'type' not in ev:
                                ev['type'] = 'event'
                        await websocket.send(json.dumps(batch))
            except Exception as e:
                logger.exception(f"Exception in sender_process WebSocket loop: {e}")
                await asyncio.sleep(2)
    try:
        asyncio.run(send_events())
    except Exception as e:
        logger.exception(f"Exception in sender_process: {e}")

SysmonPlugin: Collects and reports system metrics at intervals.

SysmonPlugin

Collects system metrics at a configurable interval and sends them to event_queue. Config keys: - poll_interval: seconds between checks (default: 5.0)

Source code in agent/plugins/sysmon_plugin.py

class SysmonPlugin:
    """
    Collects system metrics at a configurable interval and sends them to event_queue.
    Config keys:
      - poll_interval: seconds between checks (default: 5.0)
    """
    def __init__(self, config, event_queue, stop_event):
        self.poll_interval = float(config.get('poll_interval', 5.0))
        self.event_queue = event_queue
        self.stop_event = stop_event
        self.index = config.get('index', 'default')
        self.host = config.get('host', 'localhost')
        self.source = config.get('source', 'system_metrics')
        self.sourcetype = config.get('sourcetype', 'metrics')
        self.db_alias = config.get('db_alias', None)

        logger.info("SysmonPlugin initialized with poll_interval=%s", self.poll_interval)

    def run(self):
        logger.info("SysmonPlugin run started.")
        while not self.stop_event.is_set():
            metrics = {
                'cpu_percent': psutil.cpu_percent(interval=None),
                'cpu_count': psutil.cpu_count(),
                'memory': psutil.virtual_memory()._asdict(),
                'swap': psutil.swap_memory()._asdict(),
                'disk': {p.device: psutil.disk_usage(p.mountpoint)._asdict() for p in psutil.disk_partitions()},
                'net_io': psutil.net_io_counters()._asdict(),
                'boot_time': psutil.boot_time(),
                'timestamp': time.time()
            }
            logger.debug("SysmonPlugin collected metrics: %s", metrics)
            event = {
                'event_type': 'sysmon',
                'metrics': metrics,
                'timestamp': metrics['timestamp'],
                'index': self.index,
                'host': self.host,
                'source': self.source,
                'sourcetype': self.sourcetype
            }
            if self.db_alias:
                event['db_alias'] = self.db_alias
            self.event_queue.put(event)
            time.sleep(self.poll_interval)

TailPlugin: Tails files and sends new lines to event_queue, handling log rotation.

TailPlugin

Tails one or more files, following log rotation by name. Sends new lines to event_queue. Config keys: - patterns: list of glob patterns or absolute paths - poll_interval: seconds between checks (default: 1.0)

Source code in agent/plugins/tail_plugin.py

class TailPlugin:
    """
    Tails one or more files, following log rotation by name. Sends new lines to event_queue.
    Config keys:
      - patterns: list of glob patterns or absolute paths
      - poll_interval: seconds between checks (default: 1.0)
    """
    def __init__(self, config, event_queue, stop_event):
        self.patterns = config.get('patterns', [])
        if isinstance(self.patterns, str):
            self.patterns = [self.patterns]
        self.poll_interval = float(config.get('poll_interval', 1.0))
        self.event_queue = event_queue
        self.stop_event = stop_event
        self.seek_from_end = config.get('seek_from_end', True)
        self.file_positions = {}  # {str(path): offset}
        self.file_hashes = {}     # {str(path): hash}
        self.files = set()
        self.index = config.get('index', 'default')
        self.host = config.get('host', 'localhost')
        self.source = config.get('source', 'tail')
        self.sourcetype = config.get('sourcetype', 'text')
        self.db_alias = config.get('db_alias', None)
        logger.info("TailPlugin initialized with patterns=%s, poll_interval=%s", self.patterns, self.poll_interval)

    def _resolve_files(self):
        files = set()
        for pattern in self.patterns:
            for filename in glob.iglob(pattern, recursive=True):
                p = Path(filename).absolute()
                if p.is_file():
                    files.add(str(p))
        logger.debug("Resolved files: %s", files)
        return files

    def _init_file(self, filename):
        p = Path(filename)
        # Wait for file to exist and reach a minimum size
        while not p.exists() or p.stat().st_size < 1:
            time.sleep(0.1)
        with p.open('rb') as f:
            h = hashlib.sha256(f.read(256)).hexdigest()
        self.file_positions[filename] = p.stat().st_size if self.seek_from_end else 0
        self.file_hashes[filename] = h
        logger.info("Initialized file %s with hash %s, pos %s", filename, h, self.file_positions[filename])

    def run(self):
        logger.info("TailPlugin run started.")
        # Initial file discovery
        self.files = self._resolve_files()
        for filename in self.files:
            if filename not in self.file_positions:
                self._init_file(filename)
        while not self.stop_event.is_set():
            # Refresh file list (handle new files, log rotation)
            current_files = self._resolve_files()
            for filename in current_files:
                if filename not in self.file_positions:
                    self._init_file(filename)
            for filename in list(self.file_positions.keys()):
                p = Path(filename)
                try:
                    size = p.stat().st_size
                except Exception:
                    logger.warning("File missing or inaccessible: %s", filename)
                    continue
                pos = self.file_positions[filename]
                if pos > size:
                    # Log rotated, reset position and hash
                    self._init_file(filename)
                    pos = 0
                if pos == size:
                    continue
                with p.open('r', encoding='utf-8', errors='replace') as fin:
                    fin.seek(pos)
                    while True:
                        line = fin.readline()
                        if not line:
                            break
                        if not line.endswith('\n') and not line.endswith('\r'):
                            # Wait for full line
                            time.sleep(0.1)
                            continue
                        event = {
                            'src_path': filename,
                            'event_type': 'line',
                            'line': line,
                            'timestamp': time.time(),
                            'index': self.index,
                            'host': self.host,
                            'source': self.source,
                            'sourcetype': self.sourcetype
                        }
                        if self.db_alias:
                            event['db_alias'] = self.db_alias
                        self.event_queue.put(event)
                        pos = fin.tell()
                self.file_positions[filename] = pos
            time.sleep(self.poll_interval)

WatchdogPlugin: Monitors file system changes and sends events to event_queue.

WatchdogPlugin

Bases: PatternMatchingEventHandler

Monitor file system changes.

Source code in agent/plugins/watchdog_plugin.py

class WatchdogPlugin(PatternMatchingEventHandler):
    """
    Monitor file system changes.
    """
    def __init__(self, config, event_queue, stop_event):
        super(WatchdogPlugin, self).__init__(
            patterns=config.get('patterns', ['*']),
            ignore_patterns=config.get('ignore_patterns', []),
            ignore_directories=config.get('ignore_directories', True),
            case_sensitive=config.get('case_sensitive', False)
        )
        self.event_queue = event_queue
        self.stop_event = stop_event
        configured_path = Path(config.get('path_to_watch', '.')).expanduser()
        if not configured_path.is_absolute():
            configured_path = settings.BASE_DIR / configured_path
        watch_path = configured_path.resolve()
        log_path = (settings.BASE_DIR / 'logs').resolve()
        if (
            watch_path == log_path
            or watch_path in log_path.parents
            or log_path in watch_path.parents
        ):
            message = (
                f"Refusing to watch {watch_path}: recursive monitoring overlaps "
                f"SIEMatic's log directory {log_path}"
            )
            logger.error(message)
            raise ValueError(message)
        if not watch_path.is_dir():
            message = (
                f"path_to_watch '{watch_path}' does not exist. SIEMatic will not "
                "create it automatically; create the directory yourself or fix "
                "the 'path_to_watch' setting for the watchdog plugin."
            )
            logger.error(message)
            raise ValueError(message)
        self.path = str(watch_path)
        self.observer = Observer()
        self.index = config.get('index', 'default')
        self.host = config.get('host', 'localhost')
        self.source = config.get('source', 'watchdog')
        self.sourcetype = config.get('sourcetype', 'json')
        self.db_alias = config.get('db_alias', None)
        logger.info("WatchdogPlugin initialized for path=%s", self.path)

    def run(self):
        logger.info("WatchdogPlugin run started.")
        self.observer.schedule(self, self.path, recursive=True)
        self.observer.start()
        try:
            while not self.stop_event.is_set():
                time.sleep(1)
        except Exception:
            logger.exception("Exception in WatchdogPlugin run")
        finally:
            self.observer.stop()
            self.observer.join()
            logger.info("WatchdogPlugin observer stopped.")

    def on_created(self, event):
        logger.info("File created: %s", event.src_path)
        payload = {
            'src_path': event.src_path,
            'event_type': event.event_type,
            'is_directory': event.is_directory,
            'timestamp': time.time(),
            'index': self.index,
            'host': self.host,
            'source': self.source,
            'sourcetype': self.sourcetype
        }
        if self.db_alias:
            payload['db_alias'] = self.db_alias
        self.event_queue.put(payload)

    def on_modified(self, event):
        logger.info("File modified: %s", event.src_path)
        payload = {
            'src_path': event.src_path,
            'event_type': event.event_type,
            'is_directory': event.is_directory,
            'timestamp': time.time(),
            'index': self.index,
            'host': self.host,
            'source': self.source,
            'sourcetype': self.sourcetype
        }
        if self.db_alias:
            payload['db_alias'] = self.db_alias
        self.event_queue.put(payload)

    def on_deleted(self, event):
        logger.info("File deleted: %s", event.src_path)
        payload = {
            'src_path': event.src_path,
            'event_type': event.event_type,
            'is_directory': event.is_directory,
            'timestamp': time.time(),
            'index': self.index,
            'host': self.host,
            'source': self.source,
            'sourcetype': self.sourcetype
        }
        if self.db_alias:
            payload['db_alias'] = self.db_alias
        self.event_queue.put(payload)

    def on_moved(self, event):
        logger.info("File moved: %s", event.src_path)
        payload = {
            'src_path': event.src_path,
            'dest_path': event.dest_path,
            'event_type': event.event_type,
            'is_directory': event.is_directory,
            'timestamp': time.time(),
            'index': self.index,
            'host': self.host,
            'source': self.source,
            'sourcetype': self.sourcetype
        }
        if self.db_alias:
            payload['db_alias'] = self.db_alias
        self.event_queue.put(payload)

WindowsEventLogPlugin: Monitors Windows Event Log and sends new events to event_queue based on level.

WindowsEventLogPlugin

Monitors Windows Event Log for new events above a configurable level. Sends events to event_queue. Config keys: - log_type: Event log type (e.g., 'System', 'Application', 'Security') (default: 'System') - level: Minimum level to index ('ERROR', 'WARNING', 'INFORMATION', 'AUDIT_SUCCESS', 'AUDIT_FAILURE') (default: 'ERROR') - poll_interval: seconds between checks (default: 10.0)

Source code in agent/plugins/windows_event_log_plugin.py

class WindowsEventLogPlugin:
    """
    Monitors Windows Event Log for new events above a configurable level. Sends events to event_queue.
    Config keys:
      - log_type: Event log type (e.g., 'System', 'Application', 'Security') (default: 'System')
      - level: Minimum level to index ('ERROR', 'WARNING', 'INFORMATION', 'AUDIT_SUCCESS', 'AUDIT_FAILURE') (default: 'ERROR')
      - poll_interval: seconds between checks (default: 10.0)
    """
    LEVEL_MAPPING = {
        'ERROR': win32con.EVENTLOG_ERROR_TYPE if win32evtlog else 1,
        'WARNING': win32con.EVENTLOG_WARNING_TYPE if win32evtlog else 2,
        'INFORMATION': win32con.EVENTLOG_INFORMATION_TYPE if win32evtlog else 4,
        'AUDIT_SUCCESS': win32con.EVENTLOG_AUDIT_SUCCESS if win32evtlog else 8,
        'AUDIT_FAILURE': win32con.EVENTLOG_AUDIT_FAILURE if win32evtlog else 16,
    }

    def __init__(self, config, event_queue, stop_event):
        if platform.system() != 'Windows':
            raise RuntimeError("WindowsEventLogPlugin only works on Windows")
        self.log_type = config.get('log_type', 'System')
        self.level_str = config.get('level', 'ERROR').upper()
        self.min_level = self.LEVEL_MAPPING.get(self.level_str, self.LEVEL_MAPPING['ERROR'])
        self.poll_interval = float(config.get('poll_interval', 10.0))
        self.event_queue = event_queue
        self.stop_event = stop_event
        self.state_dir = os.path.expanduser('~/.siematic/state')
        os.makedirs(self.state_dir, exist_ok=True)
        self.state_file = os.path.join(self.state_dir, f'{self.log_type}.json')
        self.last_record = self._load_last_record()
        self.index = config.get('index', 'default')
        self.host = config.get('host', 'localhost')
        self.source = config.get('source', 'windows_event_log')
        self.sourcetype = config.get('sourcetype', 'json')
        self.db_alias = config.get('db_alias', None)
        logger.info("WindowsEventLogPlugin initialized with log_type=%s, level=%s, poll_interval=%s", self.log_type, self.level_str, self.poll_interval)

    def run(self):
        logger.info("WindowsEventLogPlugin run started.")
        while not self.stop_event.is_set():
            try:
                self._read_new_events()
            except Exception as e:
                logger.exception("Error reading Windows Event Log: %s", e)
            time.sleep(self.poll_interval)

    def _read_new_events(self):
        # Open the event log
        hand = win32evtlog.OpenEventLog(None, self.log_type)
        try:
            # Get total records
            total = win32evtlog.GetNumberOfEventLogRecords(hand)
            if total == 0:
                return
            # Read forwards sequentially from the beginning, filter by record number
            flags = win32evtlog.EVENTLOG_FORWARDS_READ | win32evtlog.EVENTLOG_SEQUENTIAL_READ
            events = win32evtlog.ReadEventLog(hand, flags, 0)
            for event in events:
                if event.RecordNumber > self.last_record and event.EventType >= self.min_level:
                    event_data = {
                        'record_number': event.RecordNumber,
                        'event_id': event.EventID,
                        'source_name': event.SourceName,
                        'time_generated': event.TimeGenerated.Format(),
                        'event_type': event.EventType,
                        'category': event.EventCategory,
                        'strings': event.StringInserts,
                        'data': event.Data if event.Data else None,
                        'computer_name': event.ComputerName,
                        'sid': str(event.Sid) if event.Sid else None,
                    }
                    queue_event = {
                        'src_log': self.log_type,
                        'event_type': 'windows_event',
                        'event_data': event_data,
                        'timestamp': time.time(),
                        'index': self.index,
                        'host': self.host,
                        'source': self.source,
                        'sourcetype': self.sourcetype,
                    }
                    if self.db_alias:
                        queue_event['db_alias'] = self.db_alias
                    self.event_queue.put(queue_event)
                    logger.debug("Queued Windows event: %s", event.RecordNumber)
                self.last_record = max(self.last_record, event.RecordNumber)
            self._save_last_record()
        finally:
            win32evtlog.CloseEventLog(hand)

    def _load_last_record(self):
        try:
            with open(self.state_file, 'r') as f:
                data = json.load(f)
                return data.get('last_record', 0)
        except (FileNotFoundError, json.JSONDecodeError):
            return 0

    def _save_last_record(self):
        try:
            with open(self.state_file, 'w') as f:
                json.dump({'last_record': self.last_record}, f)
        except Exception as e:
            logger.warning("Failed to save last_record: %s", e)

WindowsScheduledTasksPlugin

Enumerate Scheduled Tasks and emit a snapshot + diffs. Config: poll_interval (sec, default 60).

Source code in agent/plugins/windows_scheduled_tasks_plugin.py

class WindowsScheduledTasksPlugin:
    """
    Enumerate Scheduled Tasks and emit a snapshot + diffs.
    Config: poll_interval (sec, default 60).
    """
    def __init__(self, config, event_queue, stop_event):
        if platform.system() != 'Windows':
            raise RuntimeError('WindowsScheduledTasksPlugin only runs on Windows')
        self.event_queue = event_queue
        self.stop_event = stop_event
        self.poll_interval = float(config.get('poll_interval', 60))
        self.index  = config.get('index', 'default')
        self.host   = config.get('host', 'localhost')
        self.source = config.get('source', 'scheduled_tasks')
        self.sourcetype = config.get('sourcetype', 'json')
        self.db_alias = config.get('db_alias')

        self._last = {}

    def _list_tasks(self):
        # schtasks /Query /FO JSON requires newer builds; fall back to CSV if needed
        try:
            out = subprocess.check_output(['schtasks', '/Query', '/V', '/FO', 'LIST'], text=True, errors='replace')
        except Exception as e:
            logger.exception("schtasks failed: %s", e); return {}
        tasks, cur = {}, {}
        for line in out.splitlines():
            if not line.strip():
                if cur.get('TaskName'):
                    tasks[cur['TaskName']] = cur; cur = {}
                continue
            if ':' in line:
                k, v = line.split(':', 1)
                cur[k.strip()] = v.strip()
        if cur.get('TaskName'):
            tasks[cur['TaskName']] = cur
        return tasks

    def run(self):
        while not self.stop_event.is_set():
            now = time.time()
            tasks = self._list_tasks()
            # Diff
            added = {k:v for k,v in tasks.items() if k not in self._last}
            removed = {k:v for k,v in self._last.items() if k not in tasks}
            changed = {k:v for k,v in tasks.items() if k in self._last and v != self._last[k]}
            for kind, payload in (('added', added), ('removed', removed), ('changed', changed)):
                if not payload: continue
                event = {
                    'type': 'windows_scheduled_tasks',
                    'event_type': kind,
                    'timestamp': now,
                    'items': payload,
                    'index': self.index,
                    'host': self.host,
                    'source': self.source,
                    'sourcetype': self.sourcetype,
                }
                if self.db_alias:
                    event['db_alias'] = self.db_alias
                self.event_queue.put(event)
            self._last = tasks
            time.sleep(self.poll_interval)

Events

Models for the events app.

This module defines the Event model for storing indexed event data with metadata and extracted fields.

Event

Bases: Model

Model representing an indexed event.

Stores event data along with metadata like index, sourcetype, source, and host. Includes extracted fields for structured data access.

Source code in events/models.py

class Event(models.Model):
    """
    Model representing an indexed event.

    Stores event data along with metadata like index, sourcetype, source, and host.
    Includes extracted fields for structured data access.
    """
    id = models.BigAutoField(primary_key=True)
    index = models.CharField(max_length=255, default="default", db_index=True)
    sourcetype = models.CharField(max_length=255, default="default", db_index=True)
    source = models.CharField(max_length=255, default="default", db_index=True)
    host = models.CharField(max_length=255, default="default", db_index=True)
    data = models.TextField()
    created = models.DateTimeField(auto_now_add=True, db_index=True)
    updated = models.DateTimeField(auto_now=True, db_index=True)
    extracted_fields = models.JSONField(default=dict, blank=True, null=True)

    def save(self, *args, **kwargs):
        """Extract fields before the initial insert, keeping creation to one write."""
        if self._state.adding:
            apply_extractions(self)
        return super().save(*args, **kwargs)

    def __str__(self):
        """
        String representation of the event.

        Returns a truncated view of the event metadata and data.
        """
        return f"{self.host}: {self.index}: {self.source}: {self.sourcetype}: {self.data[:25]}...{self.data[-25:]}"

__str__()

String representation of the event.

Returns a truncated view of the event metadata and data.

Source code in events/models.py

def __str__(self):
    """
    String representation of the event.

    Returns a truncated view of the event metadata and data.
    """
    return f"{self.host}: {self.index}: {self.source}: {self.sourcetype}: {self.data[:25]}...{self.data[-25:]}"

save(*args, **kwargs)

Extract fields before the initial insert, keeping creation to one write.

Source code in events/models.py

def save(self, *args, **kwargs):
    """Extract fields before the initial insert, keeping creation to one write."""
    if self._state.adding:
        apply_extractions(self)
    return super().save(*args, **kwargs)

Event data extractors for the events app.

This module provides functions to determine sourcetypes and extract structured data from event raw data, such as JSON parsing.

apply_extractions(event)

Apply configured field extractions to an event without saving it.

The caller remains responsible for persisting the mutated event. Keeping extraction separate from persistence lets both normal saves and bulk_create perform exactly one database write.

Source code in events/extractors.py

def apply_extractions(event):
    """
    Apply configured field extractions to an event without saving it.

    The caller remains responsible for persisting the mutated event. Keeping
    extraction separate from persistence lets both normal saves and
    ``bulk_create`` perform exactly one database write.
    """
    results = {}
    extractions = getattr(settings, 'FIELD_EXTRACTIONS', {})
    logger.debug(
        "Running field extractions for event %s, %d extractors configured",
        event.id,
        len(extractions),
    )
    for predicate, extraction in extractions.items():
        if predicate(event):
            try:
                extracted = extraction(event)
                results.update(extracted)
                logger.debug(
                    "Applied extractor %s to event %s",
                    extraction.__name__,
                    event.id,
                )
            except Exception as exc:
                logger.error(
                    "Error in extractor %s for event %s: %s",
                    extraction.__name__,
                    event.id,
                    exc,
                )

    if results:
        event.extracted_fields = event.extracted_fields or {}
        event.extracted_fields.update(results)
        logger.debug(
            "Extracted %d fields for event %s", len(results), event.id
        )
    return event

extract_json(event)

Extract JSON data from the event.

Parses the event's raw data as JSON.

Parameters:

Name	Type	Description	Default
event		The event object containing JSON data.	required

Returns:

Name	Type	Description
dict		Parsed JSON data.

Raises:

Type	Description
JSONDecodeError	If the data is not valid JSON.

Source code in events/extractors.py

def extract_json(event):
    """
    Extract JSON data from the event.

    Parses the event's raw data as JSON.

    Args:
        event: The event object containing JSON data.

    Returns:
        dict: Parsed JSON data.

    Raises:
        json.JSONDecodeError: If the data is not valid JSON.
    """
    try:
        data = json.loads(event.data)
        logger.debug("Successfully extracted JSON from event data")
        return data
    except json.JSONDecodeError as e:
        logger.error(f"Failed to parse JSON from event data: {e}")
        raise

is_json_sourcetype(event)

Check if the event sourcetype is JSON.

Parameters:

Name	Type	Description	Default
event		The event object to check.	required

Returns:

Name	Type	Description
bool		True if sourcetype is 'json', False otherwise.

Source code in events/extractors.py

def is_json_sourcetype(event):
    """
    Check if the event sourcetype is JSON.

    Args:
        event: The event object to check.

    Returns:
        bool: True if sourcetype is 'json', False otherwise.
    """
    result = event.sourcetype.lower() == "json"
    logger.debug(f"Event sourcetype '{event.sourcetype}' is JSON: {result}")
    return result