path: root/pylogsentinel/core.py

"""
Core implementation for pylogsentinel.

The program is designed to be executed periodically (e.g. via cron every
minute). It maintains lightweight state files tracking how far into each
monitored log file it has already processed (byte offset and line number).
Only newly appended content is scanned on each invocation, bounded by a
configurable max_block_size to avoid unbounded processing time.

Key features:
- INI configuration (see pylogsentinel.conf.example).
- Dynamic or static log file discovery (command or explicit paths).
- Regex rule matching with per-rule actions, default fallback action.
- Lock file to prevent concurrent execution using same state directory.
- Safe state persistence per log inode (coping with rotation/truncation).
- Environment variable expansion for action shell commands.
- Optional --skip-actions mode (processes new data but suppresses action execution).

Only Python standard library modules are used.
"""

from __future__ import annotations

from dataclasses import dataclass
from collections.abc import Iterator, Sequence
from types import TracebackType

import configparser
import os
import re
import subprocess
import sys
import time


DEFAULT_CONFIG_PATH = "pylogsentinel.conf"
LOCK_FILENAME = "LOCK"
DEFAULT_MAX_BLOCK_SIZE = 10 * 1024 * 1024
CONTEXT_RADIUS = 5


class ConfigError(Exception):
    """Raised for configuration problems."""


@dataclass
class Action:
    action_id: str
    cmd: str


@dataclass
class Rule:
    rule_id: str
    pattern: str
    description: str
    action_id: str
    compiled: re.Pattern[str]
    log_set_ids: list[str]
    log_base_paths: list[str]


@dataclass
class MatchEvent:
    rule: Rule
    file_path: str
    line_number: int
    context: str
    matched_text: str
    match_count: int = 1
    matched_lines: str = ""


def parse_size(value: str) -> int:
    """
    Parse a human-friendly size string into bytes.
    Supports plain integers or integers suffixed by K, M, G (binary multiples).
    """
    v = value.strip()
    if not v:
        raise ValueError("Empty size value")
    multiplier = 1
    if v[-1] in "KkMmGg":
        unit = v[-1].upper()
        num = v[:-1]
        try:
            base = int(num)
        except ValueError as e:
            raise ValueError(f"Invalid numeric portion in size: {value!r}") from e
        if unit == "K":
            multiplier = 1024
        elif unit == "M":
            multiplier = 1024**2
        elif unit == "G":
            multiplier = 1024**3
        return base * multiplier
    try:
        return int(v)
    except ValueError as e:
        raise ValueError(f"Invalid size value: {value!r}") from e


_FLAG_MAP = {
    "i": re.IGNORECASE,
}


def _compile_pattern(raw: str) -> tuple[str, re.Pattern[str]]:
    """
    Compile a pattern of the required form: /pattern/flags

    Allowed flag: i (IGNORECASE) only. Any other flag is rejected.

    Raises:
        ConfigError if the pattern does not conform.
    """
    text = raw.strip()
    if not (len(text) >= 2 and text.startswith("/") and text.rfind("/") > 0):
        raise ConfigError(f"Pattern must be in /pattern/flags form (got {raw!r})")
    last = text.rfind("/")
    body = text[1:last]
    flags_part = text[(last + 1) :]
    if body == "":
        raise ConfigError("Empty pattern body is not allowed")
    flags = 0
    for ch in flags_part:
        if ch not in _FLAG_MAP:
            raise ConfigError(f"Unsupported regex flag {ch!r} in pattern {raw!r}")
        flags |= _FLAG_MAP[ch]
    try:
        compiled = re.compile(body, flags)
    except re.error as e:
        raise ConfigError(f"Invalid regex {body!r}: {e}") from e
    normalized = f"/{body}/{flags_part}"
    return normalized, compiled


def load_config(
    path: str,
) -> tuple[list[str], dict[str, Rule], dict[str, Action], int, str]:
    """
    Load and validate configuration supporting multiple log sets.

    Returns:
        (all_log_paths (union of all log set base paths),
         rules mapping,
         actions mapping,
         max_block_size,
         state_dir)
    """
    parser = configparser.ConfigParser(interpolation=None)
    with open(path, "r", encoding="utf-8") as fh:
        parser.read_file(fh)

    if not parser.has_section("system"):
        raise ConfigError("Missing [system] section")
    state_dir = parser.get("system", "state_dir", fallback=None)
    max_block_size_str = parser.get("system", "max_block_size", fallback="10M")
    if not state_dir:
        raise ConfigError("Missing required option: system.state_dir")
    state_dir = os.path.expanduser(os.path.expandvars(state_dir))
    try:
        max_block_size = parse_size(max_block_size_str)
    except ValueError as e:
        raise ConfigError(f"Invalid max_block_size: {e}") from e
    if max_block_size <= 0:
        raise ConfigError("max_block_size must be positive")

    log_sets: dict[str, list[str]] = {}

    for section in parser.sections():
        if section.startswith("logs."):
            log_set_id = section[len("logs.") :]
        else:
            continue

        cmd = parser.get(section, "cmd", fallback=None)
        paths_raw = parser.get(section, "paths", fallback=None)
        if cmd and paths_raw:
            raise ConfigError(f"Specify only one of cmd or paths in [{section}]")
        if not cmd and not paths_raw:
            raise ConfigError(f"Must specify cmd or paths in [{section}]")

        if paths_raw:
            paths = [
                os.path.expanduser(os.path.expandvars(p))
                for p in paths_raw.split()
                if p
            ]
        else:
            assert cmd is not None
            try:
                proc = subprocess.run(
                    cmd,
                    shell=True,
                    check=False,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                    text=True,
                )
            except OSError as e:
                raise ConfigError(
                    f"Failed to run log discovery cmd in [{section}]: {e}"
                ) from e
            if proc.returncode != 0:
                raise ConfigError(
                    f"Log discovery command failed ({proc.returncode}) in [{section}]: {cmd}\n{proc.stderr}"
                )
            paths = [
                os.path.expanduser(os.path.expandvars(line.strip()))
                for line in proc.stdout.splitlines()
                if line.strip()
            ]
        log_sets[log_set_id] = paths

    if not log_sets:
        raise ConfigError(
            "No log sets defined (need at least one [logs.<id>] section, e.g. [logs.default])"
        )

    actions: dict[str, Action] = {}
    for section in parser.sections():
        if section.startswith("action."):
            action_id = section[len("action.") :]
            cmd = parser.get(section, "cmd", fallback=None)
            if not cmd:
                raise ConfigError(f"Missing cmd in [{section}]")
            actions[action_id] = Action(action_id=action_id, cmd=cmd)

    if "default" not in actions:
        raise ConfigError("Missing [action.default] section (mandatory default action)")

    rules: dict[str, Rule] = {}
    for section in parser.sections():
        if section.startswith("rule."):
            rule_id = section[len("rule.") :]
            description = parser.get(
                section, "description", fallback="(no description)"
            )
            pattern_raw = parser.get(section, "pattern", fallback=None)
            if not pattern_raw:
                raise ConfigError(f"Missing pattern in [{section}]")
            action_id = parser.get(section, "action", fallback="default")
            if action_id not in actions:
                raise ConfigError(
                    f"Rule {rule_id!r} references unknown action {action_id!r}"
                )
            logs_field = parser.get(section, "logs", fallback=None)
            if not logs_field:
                logs_ids = ["default"]
            else:
                logs_ids = [tok for tok in logs_field.split() if tok]
            if not logs_ids:
                raise ConfigError(f"Rule {rule_id!r} has empty logs list")
            for lid in logs_ids:
                if lid not in log_sets:
                    raise ConfigError(
                        f"Rule {rule_id!r} references unknown log set {lid!r}"
                    )
            normalized, compiled = _compile_pattern(pattern_raw)
            combined_paths: list[str] = []
            _seen_paths: set[str] = set()
            for lid in logs_ids:
                for p in log_sets[lid]:
                    if p not in _seen_paths:
                        _seen_paths.add(p)
                        combined_paths.append(p)
            rules[rule_id] = Rule(
                rule_id=rule_id,
                pattern=normalized,
                description=description,
                action_id=action_id,
                compiled=compiled,
                log_set_ids=logs_ids,
                log_base_paths=combined_paths,
            )

    if not rules:
        raise ConfigError("No rules defined (need at least one [rule.*] section)")

    all_log_paths: list[str] = []
    seen: set[str] = set()
    for paths in log_sets.values():
        for p in paths:
            if p not in seen:
                seen.add(p)
                all_log_paths.append(p)

    return all_log_paths, rules, actions, max_block_size, state_dir


class LockFile:
    """
    Simple lock file using atomic O_CREAT | O_EXCL semantics.

    Ensures only one running process per state directory.
    """

    state_dir: str
    path: str
    _fd: int | None

    def __init__(self, state_dir: str, filename: str = LOCK_FILENAME) -> None:
        self.state_dir = state_dir
        self.path = os.path.join(state_dir, filename)
        self._fd: int | None = None

    def acquire(self) -> None:
        try:
            self._fd = os.open(self.path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
            os.write(
                self._fd,
                f"pid={os.getpid()} time={int(time.time())}\n".encode("utf-8"),
            )
        except FileExistsError:
            raise SystemExit(
                f"ERROR: Lock file exists: {self.path}\n"
                "If no other process is running, delete it manually."
            )
        except OSError as e:
            raise SystemExit(f"ERROR: Unable to create lock file {self.path}: {e}")

    def release(self) -> None:
        if self._fd is not None:
            try:
                os.close(self._fd)
            except OSError:
                pass
            self._fd = None
        try:
            os.unlink(self.path)
        except FileNotFoundError:
            pass
        except OSError as e:
            print(
                f"WARNING: Failed to remove lock file {self.path}: {e}", file=sys.stderr
            )

    def __enter__(self) -> "LockFile":
        self.acquire()
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        self.release()


class StateManager:
    """
    Manage per-inode state files.

    Format per file:
        <byte_offset> <line_number>

    If file is truncated (size < stored offset) we reset to 0,0.
    """

    state_dir: str

    def __init__(self, state_dir: str) -> None:
        self.state_dir = state_dir

    def _path_for_inode(self, inode: int) -> str:
        return os.path.join(self.state_dir, str(inode))

    def load(self, inode: int) -> tuple[int, int]:
        path = self._path_for_inode(inode)
        try:
            with open(path, "r", encoding="utf-8") as fh:
                content = fh.read().strip()
        except FileNotFoundError:
            return 0, 0
        except OSError as e:
            print(f"WARNING: Failed reading state file {path}: {e}", file=sys.stderr)
            return 0, 0
        if not content:
            return 0, 0
        parts = content.split()
        if len(parts) != 2:
            print(f"WARNING: Malformed state file {path}: {content!r}", file=sys.stderr)
            return 0, 0
        try:
            byte_offset = int(parts[0])
            line_number = int(parts[1])
        except ValueError:
            print(
                f"WARNING: Non-integer state file {path}: {content!r}", file=sys.stderr
            )
            return 0, 0
        return byte_offset, line_number

    def save(self, inode: int, byte_offset: int, line_number: int) -> None:
        path = self._path_for_inode(inode)
        tmp_path = f"{path}.tmp"
        data = f"{byte_offset} {line_number}\n"
        try:
            with open(tmp_path, "w", encoding="utf-8") as fh:
                fh.write(data)
            os.replace(tmp_path, path)
        except OSError as e:
            print(
                f"WARNING: Failed writing state file {path}: {e}",
                file=sys.stderr,
            )
            try:
                if os.path.exists(tmp_path):
                    os.unlink(tmp_path)
            except OSError:
                pass


def discover_files(paths: list[str]) -> Iterator[str]:
    """
    Expand a list of paths which may include directories.
    For directories: recursively traverse and yield ONLY files that the system
    'file' command reports as some form of text (case-insensitive substring
    'text' in its output). User-specified explicit file paths are yielded
    regardless of type (user responsibility).

    Fallback behavior:
      - If invoking 'file' fails (missing command or error exit), the file is
        treated as text (permissive fallback).
      - Only regular files are yielded.

    NOTE: This introduces a subprocess per file encountered under directories;
    if performance becomes a concern, consider constraining directory breadth
    or switching to explicit file paths.
    """

    def _is_text(path: str) -> bool:
        try:
            proc = subprocess.run(
                ["file", "-b", path],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                check=False,
            )
            if proc.returncode != 0:
                return True
            desc = proc.stdout.lower()
            return "text" in desc
        except Exception:
            return True

    for p in paths:
        if not p:
            continue
        if os.path.isfile(p):
            yield p
        elif os.path.isdir(p):
            for root, _, files in os.walk(p):
                for name in files:
                    full = os.path.join(root, name)
                    if os.path.isfile(full) and _is_text(full):
                        yield full
        else:
            continue


def read_new_lines(
    file_path: str,
    start_offset: int,
    start_line_number: int,
    max_block_size: int,
) -> tuple[list[str], int, int]:
    """
    Read up to max_block_size newly appended bytes from file.

    Returns:
        (new_lines (decoded with replacement errors),
         new_byte_offset,
         new_last_line_number)

    new_last_line_number is the absolute line number of the final line read.
    """
    lines: list[str] = []
    try:
        with open(file_path, "rb") as fh:
            fh.seek(0, os.SEEK_END)
            file_size = fh.tell()
            if file_size < start_offset:
                start_offset = 0
                start_line_number = 0
            fh.seek(start_offset)
            bytes_read = 0
            while bytes_read < max_block_size:
                line_bytes = fh.readline()
                if not line_bytes:
                    break
                bytes_read += len(line_bytes)
                if bytes_read > max_block_size:
                    bytes_read -= len(line_bytes)
                    break
                try:
                    decoded = line_bytes.decode("utf-8", errors="replace")
                except Exception:
                    decoded = line_bytes.decode("latin1", errors="replace")
                lines.append(decoded)
            new_offset = start_offset + bytes_read
    except FileNotFoundError:
        return [], start_offset, start_line_number
    except PermissionError:
        print(f"WARNING: Permission denied: {file_path}", file=sys.stderr)
        return [], start_offset, start_line_number
    except OSError as e:
        print(f"WARNING: Failed reading {file_path}: {e}", file=sys.stderr)
        return [], start_offset, start_line_number

    if not lines:
        return [], start_offset, start_line_number

    new_last_line_number = start_line_number + len(lines)
    return lines, new_offset, new_last_line_number


def _rule_applies(rule: Rule, file_path: str) -> bool:
    """Check if a rule applies to the given file path."""
    for base in rule.log_base_paths:
        b = base.rstrip(os.sep)
        if file_path == b or file_path.startswith(b + os.sep):
            return True
    return False


def generate_matches(
    file_path: str,
    lines: list[str],
    initial_line_number: int,
    rules: dict[str, Rule],
    context_radius: int = CONTEXT_RADIUS,
) -> Iterator[MatchEvent]:
    """
    Iterate over lines and yield one MatchEvent per cluster of nearby matches.

    Matches for the same rule are grouped into clusters: consecutive matches
    whose gap (number of non-matching lines between them) is at most
    context_radius are merged into one cluster.  Each cluster produces a
    single MatchEvent whose context spans from context_radius lines before
    the first match to context_radius lines after the last match.
    """
    total_lines = len(lines)

    # Collect per-rule match indices in one pass.
    # Key: rule_id, Value: list of (line_index, matched_text)
    rule_matches: dict[str, list[tuple[int, str]]] = {}

    for idx, line in enumerate(lines):
        for rule in rules.values():
            if not _rule_applies(rule, file_path):
                continue
            m = rule.compiled.search(line)
            if not m:
                continue
            rule_matches.setdefault(rule.rule_id, []).append((idx, m.group(0)))

    # Build clusters per rule and yield one MatchEvent per cluster.
    for rule in rules.values():
        matches = rule_matches.get(rule.rule_id)
        if not matches:
            continue

        # matches are already in line order; group into clusters.
        clusters: list[list[tuple[int, str]]] = []
        current_cluster: list[tuple[int, str]] = [matches[0]]

        for i in range(1, len(matches)):
            prev_idx = current_cluster[-1][0]
            curr_idx = matches[i][0]
            gap = curr_idx - prev_idx - 1  # non-matching lines between
            if gap <= context_radius:
                current_cluster.append(matches[i])
            else:
                clusters.append(current_cluster)
                current_cluster = [matches[i]]
        clusters.append(current_cluster)

        for cluster in clusters:
            first_idx = cluster[0][0]
            last_idx = cluster[-1][0]

            ctx_start = max(0, first_idx - context_radius)
            ctx_end = min(total_lines - 1, last_idx + context_radius)
            context = "".join(lines[ctx_start : ctx_end + 1])

            first_line_number = initial_line_number + first_idx + 1
            matched_lines_str = ",".join(
                str(initial_line_number + idx + 1) for idx, _ in cluster
            )

            yield MatchEvent(
                rule=rule,
                file_path=file_path,
                line_number=first_line_number,
                context=context,
                matched_text=cluster[0][1],
                match_count=len(cluster),
                matched_lines=matched_lines_str,
            )


def execute_action(
    action: Action,
    event: MatchEvent,
    skip_actions: bool = False,
    env_overrides: dict[str, str] | None = None,
) -> int:
    """
    Execute an action shell command with appropriate environment.

    Returns the process exit code (0 when actions are skipped).
    """
    env = os.environ.copy()
    env.update(
        {
            "RULE_ID": event.rule.rule_id,
            "RULE_PATTERN": event.rule.pattern,
            "RULE_DESCRIPTION": event.rule.description,
            "FILE": event.file_path,
            "LINE": str(event.line_number),
            "CONTEXT": event.context,
            "MATCH_COUNT": str(event.match_count),
            "MATCHED_LINES": event.matched_lines,
        }
    )
    if env_overrides:
        env.update(env_overrides)

    if skip_actions:
        print(
            f"[skip-actions] Would execute action {action.action_id!r} for rule "
            f"{event.rule.rule_id!r} on {event.file_path}:{event.line_number}, cmd: {action.cmd}",
            file=sys.stderr,
        )
        return 0

    try:
        proc = subprocess.run(
            action.cmd,
            shell=True,
            env=env,
            text=True,
        )
        if proc.returncode != 0:
            print(
                f"WARNING: Action {action.action_id} exited with {proc.returncode}",
                file=sys.stderr,
            )
        return proc.returncode
    except OSError as e:
        print(
            f"ERROR: Failed executing action {action.action_id}: {e}",
            file=sys.stderr,
        )
        return 1


class Sentinel:
    """
    Encapsulates a configured sentinel run.
    """

    log_paths: list[str]
    rules: dict[str, Rule]
    actions: dict[str, Action]
    max_block_size: int
    state_dir: str
    skip_actions: bool
    state_manager: "StateManager"

    def __init__(
        self,
        log_paths: list[str],
        rules: dict[str, Rule],
        actions: dict[str, Action],
        max_block_size: int,
        state_dir: str,
        skip_actions: bool = False,
    ) -> None:
        self.log_paths = log_paths
        self.rules = rules
        self.actions = actions
        self.max_block_size = max_block_size
        self.state_dir = state_dir
        self.skip_actions = skip_actions
        self.state_manager = StateManager(state_dir)

    def ensure_state_dir(self) -> None:
        try:
            os.makedirs(self.state_dir, exist_ok=True)
        except OSError as e:
            print(
                f"ERROR: Failed to create state_dir '{self.state_dir}': {e}",
                file=sys.stderr,
            )
            raise

    def process_file(self, file_path: str) -> None:
        try:
            st = os.stat(file_path)
        except FileNotFoundError:
            return
        except PermissionError:
            print(f"WARNING: Permission denied: {file_path}", file=sys.stderr)
            return
        except OSError as e:
            print(f"WARNING: Cannot stat {file_path}: {e}", file=sys.stderr)
            return

        inode = st.st_ino
        prev_offset, prev_line_no = self.state_manager.load(inode)

        lines, new_offset, new_last_line_no = read_new_lines(
            file_path, prev_offset, prev_line_no, self.max_block_size
        )
        if not lines:
            return

        for event in generate_matches(
            file_path, lines, prev_line_no, self.rules, CONTEXT_RADIUS
        ):
            action = self.actions.get(event.rule.action_id) or self.actions["default"]
            execute_action(action, event, skip_actions=self.skip_actions)

        self.state_manager.save(inode, new_offset, new_last_line_no)

    def run(self) -> int:
        """
        Run a single monitoring pass.

        Returns process exit code (0 success).
        """
        self.ensure_state_dir()

        with LockFile(self.state_dir):
            files = list(discover_files(self.log_paths))
            if not files:
                print("INFO: No files to process", file=sys.stderr)
                return 0
            for path in files:
                if self.skip_actions:
                    print(f"[skip-actions] Scanning {path}", file=sys.stderr)
                self.process_file(path)
        return 0


def parse_args(argv: Sequence[str]) -> tuple[str, bool]:
    """
    Very small argument parser (avoid argparse for minimal footprint).

    Supported:
        -c/--config PATH
        --skip-actions
        -h/--help
    """
    config_path = DEFAULT_CONFIG_PATH
    skip_actions = False
    it = iter(enumerate(argv))
    for _, arg in it:
        if arg in ("-h", "--help"):
            print(
                "Usage: pylogsentinel [-c CONFIG] [--skip-actions]\n"
                "Environment variables may be referenced in action commands.\n",
                file=sys.stderr,
            )
            raise SystemExit(0)
        elif arg in ("-c", "--config"):
            try:
                _, value = next(it)
            except StopIteration:
                raise SystemExit("ERROR: -c/--config requires a value")
            config_path = value
        elif arg == "--skip-actions":
            skip_actions = True
        else:
            raise SystemExit(f"ERROR: Unrecognized argument: {arg}")
    return config_path, skip_actions


def main(argv: Sequence[str] | None = None) -> int:
    """
    CLI entry point.
    """
    if argv is None:
        argv = sys.argv[1:]
    try:
        config_path, skip_actions = parse_args(list(argv))
        candidates = [
            config_path,
            os.path.expanduser("~/.pylogsentinel.conf"),
            os.path.expanduser("~/.config/pylogsentinel.conf"),
            "/etc/pylogsentinel.conf",
            "/usr/local/etc/pylogsentinel.conf",
        ]
        if not os.path.isfile(config_path):
            for cand in candidates[1:]:
                if os.path.isfile(cand):
                    config_path = cand
                    break
        if not os.path.isfile(config_path):
            print(
                "CONFIG ERROR: No configuration file found (searched: "
                + ", ".join(candidates)
                + ")",
                file=sys.stderr,
            )
            return 2
        (
            log_paths,
            rules,
            actions,
            max_block_size,
            state_dir,
        ) = load_config(config_path)
        sentinel = Sentinel(
            log_paths=log_paths,
            rules=rules,
            actions=actions,
            max_block_size=max_block_size,
            state_dir=state_dir,
            skip_actions=skip_actions,
        )
        if skip_actions:
            print(f"[skip-actions] Using config: {config_path}", file=sys.stderr)
            print(
                f"[skip-actions] Log path entries (union): {len(log_paths)}",
                file=sys.stderr,
            )
            print(
                f"[skip-actions] Rules: {', '.join(sorted(rules.keys()))}",
                file=sys.stderr,
            )
            print(
                f"[skip-actions] Actions: {', '.join(sorted(actions.keys()))}",
                file=sys.stderr,
            )
        return sentinel.run()
    except ConfigError as e:
        print(f"CONFIG ERROR: {e}", file=sys.stderr)
        return 2
    except KeyboardInterrupt:
        print("Interrupted", file=sys.stderr)
        return 130
    except SystemExit as e:
        return int(e.code) if isinstance(e.code, int) else 1
    except Exception as e:
        print(f"UNHANDLED ERROR: {e.__class__.__name__}: {e}", file=sys.stderr)
        raise


if __name__ == "__main__":
    sys.exit(main())