1#@!#!123s

: / proc / 3 / root / proc / 2 / root / opt / dedrads /
Filename : modsec_disable.py
back
#!/usr/lib/rads/venv/bin/python3
"""Disable modsec rules for Cpanel or PlatformI users and domains."""
# Written by Quentin H

import re
import subprocess
import argparse
from pathlib import Path
import sys
import rads
from rads.color import red, green, yellow

PLATFORM_I = Path("/etc/ansible/wordpress-ultrastack")


def get_apache_version() -> str:
    """Get Apache version"""
    try:
        result = subprocess.run(
            ["apachectl", "-v"], capture_output=True, text=True, check=True
        )
    except FileNotFoundError:
        print(red("apachectl not found. Please ensure Apache is installed."))
        sys.exit(1)
    except subprocess.CalledProcessError:
        print(red("Failed to get Apache version."))
        sys.exit(1)
    if "Server version" in result.stdout:
        version_line = result.stdout.split("\n")[0]
        version = version_line.split()[2]
        return version.split("/")[1]
    print(red("Unable to parse Apache version."))
    sys.exit(1)


def update_modsec_configuration(
    modsec_file: Path,
    rule_ids: list[int],
    disable_all: bool,
    reset: bool,
    remove_rule_ids: list[int],
):
    """Update modsec.conf file based on provided arguments"""
    modsec_file.parent.mkdir(parents=True, exist_ok=True)  # Ensure the directory exists
    # Ensure the modsec.conf file exists before modifying it
    if not modsec_file.exists() or reset:
        # Create or reset the file if necessary
        modsec_file.write_text("", encoding="utf-8")

    if disable_all:
        modsec_file.write_text("SecRuleEngine Off\n")
    elif rule_ids:
        for rule_id in rule_ids:
            disable_rule(modsec_file, rule_id)
    elif remove_rule_ids:
        for rule_id in remove_rule_ids:
            remove_rule(modsec_file, rule_id)


def disable_rule(modsec_file: Path, rule_id: int) -> bool:
    """Add rules to modsec.conf, checks if the rule exists and skips it if
    it does"""
    rule_text = f"SecRuleRemoveById {rule_id}"
    matcher = re.compile(rf"{re.escape(rule_text)}\b")
    try:
        with modsec_file.open("r+", encoding="utf-8") as file:
            for line in file:
                if line.startswith("SecRuleEngine Off"):
                    print(
                        red(
                            "Modsec is disabled for all rules, so not "
                            f"disabling {rule_id} in {modsec_file}."
                        )
                    )
                    return False
                if matcher.match(line):
                    print(
                        red(
                            f"Rule {rule_id} already disabled, not "
                            f"adding to {modsec_file}."
                        )
                    )
                    return False
    except FileNotFoundError:
        pass
    with modsec_file.open("a", encoding="utf-8") as file:
        file.write(f"{rule_text}\n")
    print(green(f"Rule {rule_id} added to file: {modsec_file}"))
    return True


def remove_rule(modsec_file: Path, rule_id: int) -> bool:
    """Remove a rule from the modsec file"""
    changed = False
    rule_text = f"SecRuleRemoveById {rule_id}"
    matcher = re.compile(rf"{re.escape(rule_text)}\b")
    try:
        with modsec_file.open("r+", encoding="utf-8") as file:
            contents = file.readlines()
            file.seek(0)
            file.truncate()
            for line in contents:
                if matcher.match(line):
                    changed = True
                else:
                    file.write(line)
    except FileNotFoundError:
        pass
    if changed:
        print(green(f"Rule {rule_id} removed from file: {modsec_file}"))
    else:
        print(yellow(f"Rule {rule_id} was not in file: {modsec_file}"))
    return changed


def _reset(modsec_file: Path) -> bool:
    changed = False
    with modsec_file.open("r+", encoding="utf-8") as file:
        contents = file.readlines()
        file.seek(0)
        file.truncate()
        for line in contents:
            if line.startswith("SecRuleRemoveById") or line.startswith(
                "SecRuleEngine Off"
            ):
                changed = True
            else:
                file.write(line)
    return changed


def reset_modsec_configuration(modsec_file: Path) -> bool:
    """Remove all SecRuleRemoveById and SecRuleEngine entries from the modsec
    file"""
    try:
        changed = _reset(modsec_file)
    except FileNotFoundError:
        changed = False
    if changed:
        print(
            green(
                "All SecRuleRemoveById and SecRuleEngine entries removed "
                f"from file: {modsec_file}"
            )
        )
    return changed


def cpanel_restart_apache():
    """Rebuild and restart Apache"""
    subprocess.run(
        ["/usr/local/cpanel/bin/servers_queue", "queue", "build_apache_conf"],
        check=False,
    )
    subprocess.run(
        ["/usr/local/cpanel/bin/servers_queue", "queue", "apache_restart"],
        check=False,
    )
    print(green("Apache configuration and service reload queued."))


def platform_i_restart_apache():
    try:
        subprocess.run(["apachectl", "-t"], check=True)
    except subprocess.CalledProcessError:
        print(red("Syntax failed apache check"))
        sys.exit(1)
    try:
        subprocess.run(["systemctl", "restart", "httpd"], check=True)
        print(green("Apache restarted correctly."))
    except subprocess.CalledProcessError:
        print(red("Apache failed to restart"))
        sys.exit(1)


def cpanel_run_ngxconf(user: str):
    """Run ngxconf if it exists"""
    try:
        subprocess.run(["/usr/bin/ngxconf", "-u", user, "-rd"], check=True)
    except OSError:
        pass  # not installed; probably not an ngx server
    except subprocess.CalledProcessError:
        print(red(f"/usr/bin/ngxconf -u {user} -rd failed"))


def platform_i_modsec(args):
    """Function to handle platformI check and set the modsec file path"""
    modsec_file_path = Path("/etc/httpd/modsecurity.d/activated_rules/z-mycustom.conf")
    update_modsec_configuration(
        modsec_file=modsec_file_path,
        rule_ids=args.rule,
        disable_all=args.off,
        reset=args.reset,
        remove_rule_ids=args.drop,
    )
    if args.reset:
        reset_modsec_configuration(modsec_file_path)
    platform_i_restart_apache()


def cpanel_modsec(args):
    """Cpanel function that sets the modsec.conf file based on the flags/args
    passed to the script"""
    try:
        apache_version = get_apache_version()
        if apache_version.startswith("2.4"):
            std_dir = Path("/usr/local/apache/conf/userdata/std/2_4")
            ssl_dir = Path("/usr/local/apache/conf/userdata/ssl/2_4")
        elif apache_version.startswith("2."):
            std_dir = Path("/usr/local/apache/conf/userdata/std/2")
            ssl_dir = Path("/usr/local/apache/conf/userdata/ssl/2")
        else:
            print(red(f"Unsupported Apache version {apache_version}."))
            sys.exit(1)
        # Determine the correct modsec_file path based on the presence of the
        # domain flag
        if args.domain:
            std_conf: Path = std_dir / args.user / args.domain / "modsec.conf"
            ssl_conf: Path = ssl_dir / args.user / args.domain / "modsec.conf"
        else:
            std_conf: Path = std_dir / args.user / "modsec.conf"
            ssl_conf: Path = ssl_dir / args.user / "modsec.conf"
        # Ensure the directory and files exist
        std_conf.parent.mkdir(parents=True, exist_ok=True)
        ssl_conf.parent.mkdir(parents=True, exist_ok=True)
        std_conf.touch(exist_ok=True)
        ssl_conf.touch(exist_ok=True)
        changed = False
        for rule_id in args.rule:
            changed = disable_rule(std_conf, rule_id) or changed
            changed = disable_rule(ssl_conf, rule_id) or changed
        for rule_id in args.drop:
            changed = remove_rule(std_conf, rule_id) or changed
            changed = remove_rule(ssl_conf, rule_id) or changed
        if args.reset:
            if args.domain:
                changed = reset_modsec_configuration(std_conf) or changed
                changed = reset_modsec_configuration(ssl_conf) or changed
            else:
                changed = (
                    reset_modsec_configuration(std_dir / args.user / "modsec.conf")
                    or changed
                )
                changed = (
                    reset_modsec_configuration(ssl_dir / args.user / "modsec.conf")
                    or changed
                )
        if args.off:
            changed = True
            disable_all_rules(std_conf)
            disable_all_rules(ssl_conf)
    finally:
        if changed:
            cpanel_run_ngxconf(args.user)
            cpanel_restart_apache()
        else:
            print(yellow("Skipping reload; nothing changed"))


def disable_all_rules(modsec_file: Path):
    """Function to disable all rules in modsec.conf"""
    try:
        _reset(modsec_file)
        with modsec_file.open("a", encoding="utf-8") as file:
            file.write("SecRuleEngine Off\n")
        print(green(f"All rules disabled in file: {modsec_file}"))
    except FileNotFoundError:
        pass


def parse_args():
    """Parse sys.argv"""
    # fmt: off
    parser = argparse.ArgumentParser(description=__doc__)
    if not PLATFORM_I.exists():
        parser.add_argument(
            "-u", "--user", type=str, required=True,
            help="CPanel user this is for",
        )
        parser.add_argument(
            "-d", "--domain", type=str,
            help="The domain to modify rules for",
        )
        parser.add_argument(
            "-a", "--all", action="store_true",
            help="Apply changes to all domains for the user",
        )
    parser.add_argument(
        "-o", "--off", action="store_true", help="Disable all rules"
    )
    parser.add_argument(
        "-r", "--rule", type=int, nargs='+', default=[],
        help="Specific rule IDs to disable",
    )
    parser.add_argument(
        "-R", "--reset", action="store_true",
        help="Reset the modsec configuration",
    )
    parser.add_argument(
        "-D", "--drop", nargs='+', type=int, default=[],
        help="Drop/Remove a specific rule ID from the modsec configuration"
    )
    # fmt: on
    return parser.parse_args()


def main():
    """Main function to parse arguments and run needed functions based on
    server type"""
    args = parse_args()
    try:
        with rads.lock():
            if PLATFORM_I.exists():
                platform_i_modsec(args)
            else:
                 if not args.user:
                    print(red("The --user (-u) flag is required for CPanel operations."))
                    sys.exit(1)
                 if not rads.is_cpuser(args.user):
                    print(red(f"{args.user} is not a valid cPanel user"))
                    sys.exit(1)
                 if args.domain:
                    verify_domain = rads.whoowns(args.domain)
                    if verify_domain != args.user:
                        print(red(f"The domain passed is not valid or owned by {args.user}."))
                        sys.exit(1)
                 cpanel_modsec(args)


    except rads.LockError:
        print(red("Another instance of the script is currently running. Exiting."))
        sys.exit(1)


if __name__ == "__main__":
    main()