OwlCyberSecurity - MANAGER

Edit File: check-unexpected-systemd-services.py

#!/usr/libexec/platform-python
# -*- coding: utf-8 -*-

# nc-nrpe-check-unexpected-systemd-services

# Check for Nagios, monitoring unexpected systemd services;
# Return codes;
# 0 = OK; 1 = WARNING; 2 = CRITICAL; 3 = UNKNOWN;

import os
import sys
import glob
import argparse
import re
import fnmatch
from datetime import datetime
from argparse import RawTextHelpFormatter

# Allowed sysytemd service files(services, targets, scopes)
ALLOWED_SYSTEMD_SERVICES_FILE = ["systemd_services_whitelist", "systemd_targets_whitelist", "systemd_scopes_whitelist"]
# Paths to folders where systemd files located;
SYSTEMD_SERVICES_DIR_PATHS = {}
# Unexpected services set;
UNEXPECTED_SERVICES_SET = {}
# List of patterns to match;
SEARCH_PATTERN = []
# Path to log file;
PATH_TO_LOG = "/var/log/unexpected-systemd-services.log"

# Set patterns for file search;
def set_search_patterns(args):
    # Check if log file exists;
    check_path_exists(PATH_TO_LOG)
    search_patterns = args.pattern
    services_directory=args.dirs_file

# Check if file which contains systemd directories exists;
    if check_path_exists(services_directory):
        # Read strings from file;
        SYSTEMD_SERVICES_DIR_PATHS = read_lines_services_from_file(services_directory)
    # Check and then formatting to glob patterns names of patterns;
    SEARCH_PATTERN = set_format_glob_pattrens(search_patterns)

# Get sysytemd services from files;
    allowed_services_set = set()
    for dir_path in ALLOWED_SYSTEMD_SERVICES_FILE:
        if check_path_exists(dir_path):
            allowed_services_set.update(read_lines_services_from_file(dir_path))
    if not allowed_services_set:
         print(f"CRITICAL: No lines read from files '{ALLOWED_SYSTEMD_SERVICES_FILE}' or an error occurred.")
         sys.exit(2)
    
    # Compare allowed systemd services with system sysytemd services;  
    allowed_services_from_dirs_dict = {}
    for dir_path in SYSTEMD_SERVICES_DIR_PATHS:
        allowed_services_from_dirs_dict[dir_path] = filter_files_by_pattern(dir_path, SEARCH_PATTERN)
    check_unexpected_services(allowed_services_set, allowed_services_from_dirs_dict)

# Validate unexpected services;
    if UNEXPECTED_SERVICES_SET:
        print("CRITICAL. Unexpected objects were found!\n")
        for dir, services in UNEXPECTED_SERVICES_SET.items():
            if services:
                print(f"Directory: {dir}")
                print(f"Objects list: {services}\n")
                log_critical_to_file(f"DIRECTORY - {dir} OBJECTS - {services}")
        sys.exit(2)
    else:
        print("OK, unexpected servies were not found.")
        sys.exit(0)

def set_format_glob_pattrens(search_patterns):
    # Divide patterns list and remove spaces if occurs;
    patterns_list = search_patterns.replace(' ', '').split(',')
    # Format patterns to glob;
    patterns_list = ['*.' + s for s in patterns_list]
    # Validate glob format patterns;
    for pattern in patterns_list:
        if not is_valid_glob(pattern):
           print(f"CRITICAL: Pattern: '{pattern}' is not valid!") 
           sys.exit(2)
    return patterns_list

# Glod validation function;
def is_valid_glob(pattern):
    # Validate glob patterns format;
    try:
        # Define a regular expression for valid glob patterns;
        # Match any character except null character;
        glob_regex = re.compile(r'^[\w\*\.\-\?\[\]]+$')
        
        # Check if the pattern matches the regular expression;
        if not glob_regex.match(pattern):
            print(pattern)
            return False
        
        # Use fnmatch.translate to ensure the pattern is a valid glob pattern;
        try:
            fnmatch.translate(pattern)
        except Exception:
            return False        
        return True
    except Exception as e:
        print(f"An error occurred during validation: {e}")
        return False

def check_path_exists(path):
    # Get the directory where the script is located;
    script_directory = os.path.dirname(os.path.abspath(__file__))
    # Construct the full path to the file;
    file_path = os.path.join(script_directory, path)
    try:
        if os.path.exists(file_path):
            if os.path.isfile(file_path):
                return True
            elif os.path.isdir(file_path):
                return True
        else:
            print(f"CRITICAL: file {file_path} does not exists!")
            if path == PATH_TO_LOG:
                print("HINT: check why the log file was deleted and re-create it with the following permissions: 600, owner nrpe:nrpe (Centos/Almalinux) or nagios:nagios (Debian/Ubuntu)")
            sys.exit(2)
    except PermissionError:
        print(f"CRITICAL: Permission denied to access the path.")
    except Exception as e:
        print(f"CRITICAL: An unexpected error occurred: {e}")
    sys.exit(2)

# Read allowed systemd services from file; 
def read_lines_services_from_file(filename):
    lines = set()
    try:
        # Get the directory where the script is located
        script_directory = os.path.dirname(os.path.abspath(__file__))
        # Construct the full path to the file
        file_path = os.path.join(script_directory, filename)
        
        with open(file_path, 'r') as file:
            for line in file:
                # Strip newline characters and add to the set
                lines.add(line.strip())
    except FileNotFoundError:
        print(f"The file '{file_path}' does not exist.")
        sys.exit(2)
    except PermissionError:
        print(f"Permission denied to read the file '{file_path}'.")
        sys.exit(2)
    except Exception as e:
        print(f"An unexpected error occurred: {e}")
        sys.exit(2)
    return lines

# Filter files by pattern;
def filter_files_by_pattern(directory, patterns):
    files_set = set()
    try:
        for pattern in patterns:
            # Build the search pattern;
            search_pattern = os.path.join(directory, pattern)
            
            # Use glob to find files matching the pattern;
            matching_files = {os.path.basename(file) for file in glob.glob(search_pattern) if os.path.isfile(file)}
            # Update the set with matching files;
            files_set.update(matching_files)
        return files_set
    except FileNotFoundError:
        print(f"The directory '{directory}' does not exist.")
    except PermissionError:
        print(f"Permission denied to access the directory '{directory}'.")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")

# Log critical to file;
def log_critical_to_file(message):
    # Format the date and time for the filename
    formatted_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    try:
        with open(PATH_TO_LOG, 'a') as file:
            file.write(f"{formatted_time} - CRITICAL - {message}\n")
    except FileNotFoundError:
        print("Error: The file was not found.")
    except IOError:
        print("Error: An I/O error occurred.")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")

# Calculate unexpected sysytemd services;
def check_unexpected_services(allowed_services, services_from_dir):
    for dir, services in services_from_dir.items():
        unexpected_service = services - allowed_services
        if len(unexpected_service) > 0:
            UNEXPECTED_SERVICES_SET[dir] = unexpected_service

def parse_args():
    parser = argparse.ArgumentParser(
        description="Search for unexpected systemd files based on specified patterns and directories.\n"
                    "Script usage:  ./check-unexpected-systemd-services.py -p 'service,scope,target' -d systemd_services_folders"
                    , formatter_class=RawTextHelpFormatter)
    parser.add_argument("-p", "--pattern", required=True, help="Services pattern for search. Example: -p service,scope,target")
    parser.add_argument("-d", "--dirs_file", required=True, help="File which contains folders with systemd services for compare")
    parser.set_defaults(func=set_search_patterns)
    return parser.parse_args()

# Main;
if __name__ == '__main__':
    args = parse_args()
    try:
        args.func(args)
    except Exception as error:
        print(error)
        sys.exit(1)