diff options
Diffstat (limited to 'pylogsentinel/core.py')
| -rw-r--r-- | pylogsentinel/core.py | 754 |
1 files changed, 754 insertions, 0 deletions
diff --git a/pylogsentinel/core.py b/pylogsentinel/core.py new file mode 100644 index 0000000..00ff324 --- /dev/null +++ b/pylogsentinel/core.py @@ -0,0 +1,754 @@ +""" +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 dry-run mode for testing (no commands executed). + +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 # 10 MiB +CONTEXT_RADIUS = 2 # lines before and after a match to include in CONTEXT + + +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 # may reference "default" + compiled: re.Pattern[str] + + +@dataclass +class MatchEvent: + rule: Rule + file_path: str + line_number: int + context: str + matched_text: 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, # only supported flag +} + + +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. + + Returns: + (log_paths (may contain directories), + 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") + + # Logs section + if not parser.has_section("logs"): + raise ConfigError("Missing [logs] section") + log_cmd = parser.get("logs", "cmd", fallback=None) + log_paths_raw = parser.get("logs", "paths", fallback=None) + if log_cmd and log_paths_raw: + raise ConfigError("Specify only one of [logs].cmd or [logs].paths") + if not log_cmd and not log_paths_raw: + raise ConfigError("Must specify one of [logs].cmd or [logs].paths") + + log_paths: list[str] + if log_paths_raw: + log_paths = [ + os.path.expanduser(os.path.expandvars(p)) + for p in log_paths_raw.split() + if p + ] + else: + # Run discovery command (log_cmd validated above) + assert log_cmd is not None and log_cmd.strip(), ( + "internal: log_cmd unexpectedly None/empty" + ) + try: + proc = subprocess.run( + log_cmd, # type: ignore[arg-type] # narrowed by assert for type checkers + shell=True, + check=False, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + except OSError as e: + raise ConfigError(f"Failed to run logs cmd: {e}") from e + if proc.returncode != 0: + raise ConfigError( + f"Log discovery command failed ({proc.returncode}): {log_cmd}\n{proc.stderr}" + ) + log_paths = [ + os.path.expanduser(os.path.expandvars(line.strip())) + for line in proc.stdout.splitlines() + if line.strip() + ] + + # Parse actions + 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)") + + # Parse rules + 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}" + ) + normalized, compiled = _compile_pattern(pattern_raw) + rules[rule_id] = Rule( + rule_id=rule_id, + pattern=normalized, + description=description, + action_id=action_id, + compiled=compiled, + ) + + if not rules: + raise ConfigError("No rules defined (need at least one [rule.*] section)") + + return 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], # type: ignore[arg-type] # pyright: acceptable sequence of str + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=False, + ) + if proc.returncode != 0: + return True # permissive fallback + desc = proc.stdout.lower() + return "text" in desc + except Exception: + return True # permissive fallback on unexpected errors + + for p in paths: + if not p: + continue + if os.path.isfile(p): + # Explicit file path: yield directly + 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: + # Truncated + 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: + # Do not include this partial line; rewind to before line. + 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 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 MatchEvent objects for each rule match. + """ + total_lines = len(lines) + for idx, line in enumerate(lines): + absolute_line_number = initial_line_number + idx + 1 + for rule in rules.values(): + m = rule.compiled.search(line) + if not m: + continue + # Build context + start_ctx = max(0, idx - context_radius) + end_ctx = min(total_lines, idx + context_radius + 1) + context = "".join(lines[start_ctx:end_ctx]) + yield MatchEvent( + rule=rule, + file_path=file_path, + line_number=absolute_line_number, + context=context, + matched_text=m.group(0), + ) + + +def execute_action( + action: Action, + event: MatchEvent, + dry_run: bool = False, + env_overrides: dict[str, str] | None = None, +) -> int: + """ + Execute an action shell command with appropriate environment. + + Returns the process exit code (0 on dry-run). + """ + 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, + } + ) + if env_overrides: + env.update(env_overrides) + + if dry_run: + print( + f"[dry-run] Would execute action {action.action_id!r} for rule " + f"{event.rule.rule_id!r} on {event.file_path}:{event.line_number}\n" + f"Command: {action.cmd}\n", + file=sys.stderr, + ) + return 0 + + try: + proc = subprocess.run( # type: ignore + 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 + dry_run: 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, + dry_run: 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.dry_run = dry_run + 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: + # Explicit error output before failing; re-raise original exception + 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: + # No new content OR failed to read + 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, dry_run=self.dry_run) + + 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.dry_run: + print(f"[dry-run] 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 + --dry-run + -h/--help + """ + config_path = DEFAULT_CONFIG_PATH + dry_run = False + it = iter(enumerate(argv)) + for _, arg in it: + if arg in ("-h", "--help"): + print( + "Usage: pylogsentinel [-c CONFIG] [--dry-run]\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 == "--dry-run": + dry_run = True + else: + # Unrecognized argument + raise SystemExit(f"ERROR: Unrecognized argument: {arg}") + return config_path, dry_run + + +def main(argv: Sequence[str] | None = None) -> int: + """ + CLI entry point. + """ + if argv is None: + argv = sys.argv[1:] + try: + config_path, dry_run = parse_args(list(argv)) + # If the provided (or default) config path does not exist, attempt standard locations. + if not os.path.isfile(config_path): + for cand in ( + os.path.expanduser("~/.pylogsentinel.conf"), + os.path.expanduser("~/.config/pylogsentinel.conf"), + "/etc/pylogsentinel.conf", + "/usr/local/etc/pylogsentinel.conf", + config_path, # fallback original (may still not exist) + ): + if os.path.isfile(cand): + config_path = cand + break + ( + 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, + dry_run=dry_run, + ) + if dry_run: + print(f"[dry-run] Using config: {config_path}", file=sys.stderr) + print(f"[dry-run] Log path entries: {len(log_paths)}", file=sys.stderr) + print( + f"[dry-run] Rules: {', '.join(sorted(rules.keys()))}", file=sys.stderr + ) + print( + f"[dry-run] 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()) |