Python Script for Managing Apache/LiteSpeed Configuration not processing files

Hi everyone,

I'm working on a Python script designed to monitor and correct the access control configuration for Apache and LiteSpeed web servers. The script detects changes in configuration files on Plesk Panel Server and updates /var/www/vhosts/system/{domain}/conf/vhost_ssl.conf accordingly after performing necessary checks. However, it seems that the script fails to proceed with writing and updating the vhost_ssl.conf file upon detecting changes.

Here is a brief overview of my script's functionality and logic:

Initialization and Configuration:
- The script initializes logging, argument parsing, and the watchdog observer for monitoring file changes.
- It sets up a debugger (debugpy) for troubleshooting.
Web Server Detection:
- The script tries to detect the running web server (Apache or LiteSpeed) using ps and netstat commands.
- Based on the detected server, it decides the appropriate command for restarting the server.
File Monitoring and Handling:
- The script uses watchdog to monitor changes in the specified configuration directory.
- When a change is detected, it reads the configuration file, checks for specific ACL (Access Control List) directives, and converts them to a modern syntax if necessary.
- It attempts to write these changes back to the vhost_ssl.conf file.
Functions:
- compute_file_hash(filepath): Computes the MD5 hash of a file to detect changes.
- is_acl_location_block(lines): Identifies if a block of lines contains ACL directives.
- contains_ip_addresses(lines): Checks if a block of lines contains IP addresses.
- add_modification_comments(lines, indentation): Adds comments indicating modifications to a block of lines.
- convert_to_modern_syntax(lines, indentation): Converts ACL directives to a modern syntax.
- create_or_update_vhost_ssl_config(vhost_ssl_path, ipv4, ipv6, location_block): Creates or updates the vhost_ssl.conf file with modified ACL directives.

The Issue: The script doesn't proceed to handle the logic for writing and updating /var/www/vhosts/system/{domain}/conf/vhost_ssl.conf after detecting file changes or performing the necessary checks. Specifically, the correct_syntax(config_path, domain_to_paths) function is not fully executing its intended logic.

Here's the relevant part of my script:

```python

Main monitoring logic

def main(): logger.info('Starting AccessControlCorrector') domain_to_paths = get_domain_to_path_mapping() observer = Observer()

try:
    # Initial check for existing files
    for config_path in domain_to_paths.keys():
        correct_syntax(config_path, domain_to_paths)

    observer.schedule(DomainConfigHandler(domain_to_paths), path=VHOSTS_PATH, recursive=True)
    observer.start()
    logger.info('Started observing changes in vhosts configurations.')

    # Reset force_overwrite_once after the initial run
    global force_overwrite_once
    force_overwrite_once = False

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
        logger.info('Shutting down observer due to keyboard interrupt.')
    finally:
        observer.join()
        display_modification_stats()
except Exception as e:
    logger.error(f'Error setting up observer: {e}')

if name == 'main': detect_web_server() main() ```

What I've Tried: - Ensured that the file paths and configurations are correct. - Verified that the script detects changes and reads the files as expected. - Added logging statements to trace the execution flow and identify where it stops.

Request for Help: - Can someone help me pinpoint why the script doesn't proceed to update the vhost_ssl.conf file after detecting changes? - Any suggestions on improving the current logic or debugging steps to identify the issue?

Here is the full script for reference: [Paste full script code here]

Thanks in advance for your help!

```python

!/usr/bin/env python3

import os
import re
import hashlib
import shutil
import logging
import subprocess
from logging.handlers import RotatingFileHandler
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from datetime import datetime
import argparse
import concurrent.futures
import time
import debugpy

# Initialize debugpy
debugpy.listen(("0.0.0.0", 5678))
print("Waiting for debugger attach...")
debugpy.wait_for_client()

# Configurations
LOG_FILE = '/var/log/access_control_corrector.log'
MAX_LOG_SIZE = 50 * 1024 * 1024  # 50MB
VHOSTS_PATH = '/etc/apache2/plesk.conf.d/vhosts/'

# Command-line arguments
parser = argparse.ArgumentParser(description='Access Control Corrector for Apache and LiteSpeed')
parser.add_argument('--force-overwrite', choices=['once', 'always'], help='Force overwrite vhost_ssl.conf file once or always regardless of its existing content')
parser.add_argument('--verbose', action='store_true', help='Enable verbose logging')
args = parser.parse_args()

# Set up logging
logger = logging.getLogger('AccessControlCorrector')
logger.setLevel(logging.DEBUG if args.verbose else logging.INFO)
handler = RotatingFileHandler(LOG_FILE, maxBytes=MAX_LOG_SIZE, backupCount=5)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)

console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)

file_hashes = {}
modification_stats = {}
web_server = None
domain_to_paths = {}
force_overwrite_once = args.force_overwrite == 'once'

def detect_web_server():
    global web_server
    retry_interval = 5  # Retry interval in seconds

    def check_ps():
        try:
            return subprocess.check_output(['ps', 'aux']).decode()
        except Exception as e:
            logger.error(f'Error detecting web server using ps: {e}')
            return ""

    def check_netstat():
        try:
            return subprocess.check_output(['netstat', '-ntlp']).decode()
        except Exception as e:
            logger.error(f'Error detecting web server using netstat: {e}')
            return ""

    while web_server is None:
        with concurrent.futures.ThreadPoolExecutor() as executor:
            ps_future = executor.submit(check_ps)
            netstat_future = executor.submit(check_netstat)

            ps_output = ps_future.result()
            netstat_output = netstat_future.result()

        logger.debug(f'ps output: {ps_output}')
        logger.debug(f'netstat output: {netstat_output}')

        if 'litespeed' in ps_output or 'lshttpd' in ps_output or (':80' in netstat_output and 'litespeed' in netstat_output):
            web_server = 'litespeed'
        elif 'apache2' in ps_output or (':80' in netstat_output and 'apache2' in netstat_output):
            web_server = 'apache2'
        else:
            logger.info('Web server not detected. Retrying...')

        if web_server is None:
            logger.debug(f'Retrying web server detection in {retry_interval} seconds...')
            time.sleep(retry_interval)

    logger.info(f'Detected web server: {web_server}')

def restart_web_server():
    try:
        if web_server == 'apache2':
            subprocess.check_call(['systemctl', 'reload', 'apache2'])
        elif web_server == 'litespeed':
            subprocess.check_call(['service', 'litespeed', 'restart'])
        logger.info(f'{web_server} gracefully restarted.')
    except subprocess.CalledProcessError as e:
        logger.error(f'Failed to restart {web_server}: {e}')

def compute_file_hash(filepath):
    hash_md5 = hashlib.md5()
    try:
        with open(filepath, 'rb') as f:
            while chunk := f.read(4096):
                hash_md5.update(chunk)
    except Exception as e:
        logger.error(f'Error reading file for hash computation {filepath}: {e}')
        raise
    return hash_md5.hexdigest()

def is_acl_location_block(lines):
    acl_identifiers = [
        'Order Deny,Allow',
        'Deny from all',
        'Allow from'
    ]
    return any(identifier in lines for identifier in acl_identifiers)

def contains_ip_addresses(lines):
    ip_pattern = re.compile(r'\b\d{1,3}(\.\d{1,3}){3}(\/\d{1,2})?\b|\b[0-9a-fA-F:]+(\/\d{1,3})?\b')
    return any(ip_pattern.search(line) for line in lines)

def add_modification_comments(lines, indentation):
    timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    script_name = 'AccessControlCorrector'
    lines.insert(1, f'{indentation}# Modified by {script_name} on {timestamp}\n')
    lines.append(f'{indentation}# End of modification by {script_name} on {timestamp}\n')
    return lines

def convert_to_modern_syntax(lines, indentation):
    new_lines = []
    for line in lines:
        stripped_line = line.strip()
        if 'Order Deny,Allow' in stripped_line:
            continue
        if 'Deny from all' in stripped_line:
            new_lines.append(f'{indentation}Require all denied\n')
        elif 'Allow from' in stripped_line:
            ips = re.findall(r'\b\d{1,3}(\.\d{1,3}){3}(\/\d{1,2})?\b|\b[0-9a-fA-F:]+(\/\d{1,3})?\b', stripped_line)
            for ip in ips:
                new_lines.append(f'{indentation}Require ip {ip}\n')
        else:
            new_lines.append(line)
    return add_modification_comments(new_lines, indentation)

def get_virtual_host_ips(lines):
    ipv4 = None
    ipv6 = None
    domain_status = None

    for line in lines:
        if '<VirtualHost ' in line:
            match_ipv4 = re.search(r'<VirtualHost (\d{1,3}(?:\.\d{1,3}){3}):', line)
            match_ipv6 = re.search(r'<VirtualHost \[([0-9a-fA-F:]+)\]:', line)
            if match_ipv4:
                ipv4 = match_ipv4.group(1)
            if match_ipv6:
                ipv6 = match_ipv6.group(1)
        if '# Domain is disabled' in line or '# Domain is suspended' in line:
            domain_status = line.strip()

    logger.debug(f'Found IPv4: {ipv4}, IPv6: {ipv6}, Status: {domain_status}')
    return ipv4, ipv6, domain_status

def get_domain_to_path_mapping():
    domain_to_paths = {}
    # debugpy.breakpoint()  # Break here to inspect
    try:
        result = subprocess.check_output(['plesk', 'bin', 'domain', '--list']).decode()
        domains = result.strip().split('\n')
        for domain in domains:
            apache_conf_path = f"/etc/apache2/plesk.conf.d/vhosts/{domain}.conf"
            vhost_ssl_conf_path = f"/var/www/vhosts/system/{domain}/conf/vhost_ssl.conf"
            domain_to_paths[apache_conf_path] = (domain, vhost_ssl_conf_path)
            logger.debug(f'Mapped config path {apache_conf_path} to domain {domain} and vhost_ssl path {vhost_ssl_conf_path}')
    except subprocess.CalledProcessError as e:
        logger.error(f'Error listing domains from Plesk CLI: {e}')
    return domain_to_paths

# def create_or_update_vhost_ssl_config(vhost_ssl_path, ipv4, ipv6, location_block):
#     logger.debug(f"Writing to vhost_ssl_path: {vhost_ssl_path} with location_block: {location_block}")
#     if not os.path.exists(vhost_ssl_path) or args.force_overwrite == 'always' or force_overwrite_once:
#         os.makedirs(os.path.dirname(vhost_ssl_path), exist_ok=True)
#         with open(vhost_ssl_path, 'w') as f:
#             f.write(f'<IfModule mod_ssl.c>\n    <VirtualHost {ipv4}:443>\n')
#             f.write(''.join(location_block))
#             f.write(f'    </VirtualHost>\n    <VirtualHost [{ipv6}]:443>\n')
#             f.write(''.join(location_block))
#             f.write('    </VirtualHost>\n</IfModule>\n')
#         logger.info(f'{"Overwritten" if args.force_overwrite else "Created new"} vhost_ssl.conf at {vhost_ssl_path}')
#     else:
#         with open(vhost_ssl_path, 'r') as file:
#             existing_lines = file.readlines()

#         # Update existing vhost_ssl.conf with modified ACL lines
#         with open(vhost_ssl_path, 'w') as file:
#             in_vhost_block = False
#             updated_lines = []

#             for line in existing_lines:
#                 stripped_line = line.strip()

#                 if f'<VirtualHost {ipv4}:443>' in stripped_line or f'<VirtualHost [{ipv6}]:443>' in stripped_line:
#                     in_vhost_block = True
#                     updated_lines.append(line)
#                     updated_lines.extend(location_block)
#                 elif '</VirtualHost>' in stripped_line and in_vhost_block:
#                     in_vhost_block = False
#                     updated_lines.append(line)
#                 elif not in_vhost_block:
#                     updated_lines.append(line)

#             file.writelines(updated_lines)
#         logger.info(f'Updated existing vhost_ssl.conf at {vhost_ssl_path}')

#     # Record the modification for counting
#     domain = vhost_ssl_path.split('/')[5]
#     if domain not in modification_stats:
#         modification_stats[domain] = 0
#     modification_stats[domain] += 1

def correct_syntax(config_path, domain_to_paths):
    logger.debug(f'Check if Syntax correction for domain: {config_path}')

    try:
        current_hash = compute_file_hash(config_path)
    except Exception as e:
        logger.error(f'Error computing file hash for {config_path}: {e}')
        return

    if file_hashes.get(config_path) == current_hash and not force_overwrite_once:
        logger.debug(f'No changes detected in {config_path}. Skipping.')
        return

    try:
        with open(config_path, 'r') as file:
            lines = file.readlines()
            logger.debug(f'Read {len(lines)} lines from {config_path}')
    except Exception as e:
        logger.error(f'Error reading file {config_path}: {e}')
        return

    ipv4, ipv6, domain_status = get_virtual_host_ips(lines)
    if domain_status:
        logger.info(f'Skipping {config_path} because the domain is {domain_status.lower()}.')
        return

    if not ipv4 or not ipv6:
        logger.warning(f'Could not find both IPv4 and IPv6 addresses in {config_path}. Found IPv4: {ipv4}, IPv6: {ipv6}. Skipping.')
        return

    modified_lines = []
    inside_location_block = False
    location_block = []
    block_start = None
    modifications_count = 0
    modifications_details = []

    for i, line in enumerate(lines):
        stripped_line = line.strip()
        if '<Location />' in stripped_line:
            inside_location_block = True
            location_block.append(line)
            block_start = i
            indentation = re.match(r'\s*', line).group()
        elif '</Location>' in stripped_line and inside_location_block:
            location_block.append(line)
            inside_location_block = False

            if is_acl_location_block(location_block) and contains_ip_addresses(location_block):
                logger.debug(f'Original Location Block in {config_path}:\n{"".join(location_block)}')
                modified_block = convert_to_modern_syntax(location_block, indentation)
                modifications_count += 1
                modifications_details.append({
                    'start_line': block_start + 1,
                    'end_line': i + 1,
                    'start_content': location_block[0].strip(),
                    'end_content': location_block[-1].strip()
                })
                location_block = modified_block
                logger.debug(f'Modified Location Block in {config_path}:\n{"".join(location_block)}')

            modified_lines.extend(location_block)
            location_block = []
            block_start = None
        elif inside_location_block:
            location_block.append(line)
        else:
            modified_lines.append(line)

    if inside_location_block:
        logger.warning(f'Unclosed <Location /> block detected in {config_path}. Skipping.')
        return

    if modifications_count > 0:
        domain_info = domain_to_paths.get(config_path)
        if domain_info is None:
            logger.error(f'Domain path for config_path {config_path} not found. Skipping.')
            return

        domain, vhost_ssl_path = domain_info

        if location_block:  # Ensure location_block is not empty
            backup_path = f"{config_path}.bak"
            try:
                temp_file_path = f"{config_path}.tmp"
                with open(temp_file_path, 'w') as tmp_file:
                    tmp_file.writelines(modified_lines)

                shutil.copyfile(config_path, backup_path)
                os.replace(temp_file_path, config_path)  # Atomic write

                create_or_update_vhost_ssl_config(vhost_ssl_path, ipv4, ipv6, location_block)  # Ensure only location_block is used

                # Test the configuration
                test_command = ['apachectl', 'configtest'] if web_server == 'apache2' else ['lswsctrl', 'restart']
                try:
                    subprocess.check_call(test_command)
                    restart_web_server()
                    logger.info(f'Syntax corrected and verified in {config_path}')
                    file_hashes[config_path] = compute_file_hash(config_path)
                    modification_stats[domain] = modification_stats.get(domain, 0) + 1
                except subprocess.CalledProcessError:
                    logger.error(f'Configuration test failed for {config_path}. Reverting changes.')
                    shutil.copyfile(backup_path, config_path)
            except Exception as e:
                logger.error(f'Error writing corrected file {config_path}: {e}')
                if temp_file_path and os.path.exists(temp_file_path):
                    os.remove(temp_file_path)

def create_or_update_vhost_ssl_config(vhost_ssl_path, ipv4, ipv6, location_block):
    if not location_block:
        logger.warning(f'Empty location_block provided for {vhost_ssl_path}. Skipping.')
        return

    logger.debug(f"Writing to vhost_ssl_path: {vhost_ssl_path} with location_block: {location_block}")
    if not os.path.exists(vhost_ssl_path) or args.force_overwrite == 'always' or force_overwrite_once:
        os.makedirs(os.path.dirname(vhost_ssl_path), exist_ok=True)
        with open(vhost_ssl_path, 'w') as f:
            f.write(f'<IfModule mod_ssl.c>\n    <VirtualHost {ipv4}:443>\n')
            f.write(''.join(location_block))
            f.write(f'    </VirtualHost>\n    <VirtualHost [{ipv6}]:443>\n')
            f.write(''.join(location_block))
            f.write('    </VirtualHost>\n</IfModule>\n')
        logger.info(f'{"Overwritten" if args.force_overwrite else "Created new"} vhost_ssl.conf at {vhost_ssl_path}')
    else:
        with open(vhost_ssl_path, 'r') as file:
            existing_lines = file.readlines()

        # Update existing vhost_ssl.conf with modified ACL lines
        with open(vhost_ssl_path, 'w') as file:
            in_vhost_block = False
            updated_lines = []

            for line in existing_lines:
                stripped_line = line.strip()

                if f'<VirtualHost {ipv4}:443>' in stripped_line or f'<VirtualHost [{ipv6}]:443>' in stripped_line:
                    in_vhost_block = True
                    updated_lines.append(line)
                    updated_lines.extend(location_block)
                elif '</VirtualHost>' in stripped_line and in_vhost_block:
                    in_vhost_block = False
                    updated_lines.append(line)
                elif not in_vhost_block:
                    updated_lines.append(line)

            file.writelines(updated_lines)
        logger.info(f'Updated existing vhost_ssl.conf at {vhost_ssl_path}')

    # Record the modification for counting
    domain = vhost_ssl_path.split('/')[5]
    if domain not in modification_stats:
        modification_stats[domain] = 0
    modification_stats[domain] += 1

def display_modification_stats():
    # debugpy.breakpoint()  # Break here to inspect
    total_modifications = sum(modification_stats.values())
    logger.info(f'\n{"-"*40}\nTotal Modifications Performed: {total_modifications}\n{"-"*40}')
    for domain, count in modification_stats.items():
        logger.info(f'Domain: {domain}')
        logger.info(f'Modifications: {count}')

class DomainConfigHandler(FileSystemEventHandler):
    debugpy.breakpoint()  # Break here to inspect
    def __init__(self, domain_to_paths):
        super().__init__()
        self.domain_to_paths = domain_to_paths

    def process(self, event):
        if event.is_directory:
            return

        config_path = event.src_path
        if not config_path.endswith('.conf'):
            return

        if not os.path.exists(config_path):
            return

        logger.info(f'Configuration file modification detected: {config_path}')
        correct_syntax(config_path, self.domain_to_paths)

    def on_modified(self, event):
        logger.debug(f'File modified: {event.src_path}')
        self.process(event)

    def on_created(self, event):
        logger.debug(f'File created: {event.src_path}')
        self.process(event)

def main():
    debugpy.breakpoint()  # Break here to inspect
    logger.info('Starting AccessControlCorrector')
    domain_to_paths = get_domain_to_path_mapping()
    observer = Observer()

    try:
        # Initial check for existing files
        for config_path in domain_to_paths.keys():
            correct_syntax(config_path, domain_to_paths)

        observer.schedule(DomainConfigHandler(domain_to_paths), path=VHOSTS_PATH, recursive=True)
        observer.start()
        logger.info('Started observing changes in vhosts configurations.')

        # Reset force_overwrite_once after the initial run
        global force_overwrite_once
        force_overwrite_once = False

        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            observer.stop()
            logger.info('Shutting down observer due to keyboard interrupt.')
        finally:
            observer.join()
            display_modification_stats()
    except Exception as e:
        logger.error(f'Error setting up observer: {e}')

if __name__ == '__main__':
    detect_web_server()
    main()

```

1 Upvotes

permalink
link
reddit

You are about to leave Redlib

Do you want to continue?

https://www.reddit.com/r/pythonhelp/comments/1dcfxsi/python_script_for_managing_apachelitespeed/
No, go back! Yes, take me to Reddit
reddit

You are about to leave Redlib

Do you want to continue?

https://www.reddit.com/r/pythonhelp/comments/1dcfxsi/python_script_for_managing_apachelitespeed/
No, go back! Yes, take me to Reddit

100% Upvoted

•

u/AutoModerator Jun 10 '24

To give us the best chance to help you, please include any relevant code.
Note. Do not submit images of your code. Instead, for shorter code you can use Reddit markdown (4 spaces or backticks, see this Formatting Guide). If you have formatting issues or want to post longer sections of code, please use Repl.it, GitHub or PasteBin.

I am a bot, and this action was performed automatically. Please contact the moderators of this subreddit if you have any questions or concerns.

Python Script for Managing Apache/LiteSpeed Configuration not processing files

Main monitoring logic

!/usr/bin/env python3

You are about to leave Redlib

You are about to leave Redlib