#$@%#$%!

Ngiler SH3LL 360
: / opt / imunify360 / venv / share / imunify360 / scripts / migrate_csf /

Information Server
MySQL : OFF Perl : OFF CURL : ON WGET : OFF PKEXEC : OFF

Directive	Local Value
IP Address	63.250.38.10
System	Linux premium90.web-hosting.com 4.18.0-553.44.1.lve.el8.x86_64 #1 SMP Thu Mar 13 14:29:12 UTC 2025 x86_64
User	likhxapw
PHP Version	8.3.30
Software	LiteSpeed
Doc root	Writable

Edit File : ips.py \| Size : 24.55 KB
import subprocess import ipaddress from pathlib import Path from logger_config import get_logger, capture_exception class AdvancedFilter: """Represents a parsed CSF advanced filter rule.""" def __init__( self, protocol: str = "tcp", direction: str = "in", source_ip: str \| None = None, dest_port: int \| None = None, source_port: int \| None = None, original_line: str = "", ): self.protocol = protocol self.direction = direction self.source_ip = source_ip self.dest_port = dest_port self.source_port = source_port self.original_line = original_line def __repr__(self): return f"AdvancedFilter(protocol={self.protocol}, direction={self.direction}, source_ip={self.source_ip}, dest_port={self.dest_port}, source_port={self.source_port})" def is_valid_ipv6_network_for_imunify(ip_string: str) -> tuple[bool, str]: """ Validate IPv6 network according to Imunify360 requirements. Imunify360 only supports IPv6 /64 networks. Args: ip_string: IPv6 address or network in CIDR notation Returns: Tuple of (is_valid, error_message) """ if "/" not in ip_string: # Individual IPv6 addresses should be converted to /128 for validation # but Imunify360 doesn't support /128, so this is an error return ( False, "Individual IPv6 addresses not supported, use /64 networks", ) try: # First check prefix length by parsing manually _, prefix_part = ip_string.split("/", 1) try: prefix_len = int(prefix_part) except ValueError: return False, "Invalid prefix length" # Check if prefix is /64 (Imunify360 requirement) if prefix_len != 64: return False, "Supported only ipv6 /64 networks" # Now check if it's a valid network with proper alignment ipaddress.IPv6Network(ip_string, strict=True) return True, "" except ( ipaddress.AddressValueError, ipaddress.NetmaskValueError, ValueError, ) as e: error_msg = str(e) if "has host bits set" in error_msg: return False, "Supported only ipv6 /64 networks" return False, f"Invalid IPv6 network: {error_msg}" def is_valid_ipv4_network_for_imunify(ip_string: str) -> tuple[bool, str]: """ Validate IPv4 address or network according to Imunify360 requirements. Args: ip_string: IPv4 address or network in CIDR notation Returns: Tuple of (is_valid, error_message) """ try: if "/" in ip_string: # CIDR notation - check if it's a proper network ipaddress.IPv4Network(ip_string, strict=True) return True, "" else: # Individual IP address ipaddress.IPv4Address(ip_string) return True, "" except ipaddress.AddressValueError as e: if "has host bits set" in str(e): return False, "has host bits set" return False, "Invalid IPv4 address/network" except (ipaddress.NetmaskValueError, ValueError) as e: return False, f"Invalid IPv4 address/network: {str(e)}" def is_valid_ip_or_cidr(ip_string: str) -> bool: """Validate if a string is a valid IP address (IPv4 or IPv6) or CIDR notation.""" try: # Try to parse as network (handles CIDR notation) if "/" in ip_string: ipaddress.ip_network(ip_string, strict=False) else: # Try to parse as individual IP address ipaddress.ip_address(ip_string) return True except ( ipaddress.AddressValueError, ipaddress.NetmaskValueError, ValueError, ): return False def is_dangerous_catchall_ip(ip_string: str) -> bool: """ Check if an IP address or network represents a dangerous catch-all that could block all traffic. This includes: - IPv4: 0.0.0.0 or any CIDR starting with 0.0.0.0 - IPv6: :: or any CIDR starting with :: Args: ip_string: IP address or network string to check Returns: True if this is a dangerous catch-all IP/network """ ip_string = ip_string.strip() if ":" in ip_string: # IPv6 - check for :: (all zeros) if ip_string == "::" or ip_string.startswith("::/"): return True # Also check for explicit all-zeros representation if ip_string.startswith("0000:0000:0000:0000:0000:0000:0000:0000"): return True else: # IPv4 - check for 0.0.0.0 if ip_string == "0.0.0.0" or ip_string.startswith("0.0.0.0/"): return True return False def validate_ip_for_imunify(ip_string: str) -> tuple[bool, str]: """ Comprehensive validation for IP addresses/networks according to Imunify360 requirements. Args: ip_string: IP address or network string to validate Returns: Tuple of (is_valid, error_message) """ if not ip_string or not ip_string.strip(): return False, "Empty IP address" ip_string = ip_string.strip() if is_dangerous_catchall_ip(ip_string): return ( False, f"DANGEROUS: {ip_string} represents all addresses and would block all connections", ) # Determine if it's IPv4 or IPv6 if ":" in ip_string: # IPv6 return is_valid_ipv6_network_for_imunify(ip_string) else: # IPv4 return is_valid_ipv4_network_for_imunify(ip_string) def get_ip_from_line(line: str) -> str \| None: """Extract and validate IP from a line, returning None if invalid.""" if not is_valid_ip_or_cidr(line): return None is_valid, error_msg = validate_ip_for_imunify(line) if is_valid: return line logger = get_logger() msg = f"Invalid IP/CIDR '{line}': {error_msg}." if "dangerous" in error_msg.lower(): logger.error(msg) return None logger.warning(msg) return None def get_advanced_filter_from_line(line: str) -> AdvancedFilter \| None: """ Parse a CSF advanced filter line. Format: [protocol]\|[direction]\|[parameter=value]\|[parameter=value]... or minimal: parameter=value Examples: - udp\|d=53\|s=192.168.1.0/24 - in\|d=80\|s=8.8.8.8 - tcp\|out\|d=443\|s=10.0.0.1 - d=80 (minimal filter) Args: line: CSF advanced filter line Returns: AdvancedFilter object if successfully parsed, None otherwise """ parts = line.split("\|") # Require at least one meaningful part (not just empty strings) meaningful_parts = [p.strip() for p in parts if p.strip()] if len(meaningful_parts) < 1: return None # Must have at least one parameter with "=" or be a protocol/direction has_parameter = any("=" in part for part in meaningful_parts) has_protocol_or_direction = any( part.lower() in ["tcp", "udp", "in", "out"] for part in meaningful_parts ) if not has_parameter and not has_protocol_or_direction: return None # Initialize defaults protocol = "tcp" direction = "in" source_ip = None dest_port = None source_port = None for part in meaningful_parts: part = part.strip() if not part: continue # Check for protocol if part.lower() in ["tcp", "udp"]: protocol = part.lower() # Check for direction elif part.lower() in ["in", "out"]: direction = part.lower() # Check for parameter=value pairs elif "=" in part: param, value = part.split("=", 1) param = param.strip().lower() value = value.strip() if param == "s": # source try: source_port = int(value) except ValueError: # Treat as IP if not a valid port number if is_valid_ip_or_cidr(value): # Additional validation for Imunify360 requirements is_valid, error_msg = validate_ip_for_imunify(value) if is_valid: source_ip = value else: logger = get_logger() msg = f"Invalid source IP in advanced filter '{value}': {error_msg}." if "dangerous" in error_msg.lower(): logger.error(msg) else: logger.warning(msg) elif param == "d": # destination try: dest_port = int(value) except ValueError: # For destination, we only support port numbers in this migration pass return AdvancedFilter( protocol=protocol, direction=direction, source_ip=source_ip, dest_port=dest_port, source_port=source_port, original_line=line, ) def get_ips_and_filters_from_csf_file( file_path: str, processed_files: set[str] \| None = None ) -> tuple[list[str], list[AdvancedFilter]]: """ Read a CSF file and extract both IP addresses/CIDR ranges and advanced filters. Handles Include statements recursively. Args: file_path: Path to the CSF file processed_files: Set of already processed files to avoid infinite loops Returns: Tuple of (List of IP addresses/CIDR ranges, List of AdvancedFilter objects) """ logger = get_logger() if processed_files is None: processed_files = set() # Convert to absolute path to avoid processing same file multiple times abs_path = str(Path(file_path).resolve()) if abs_path in processed_files: return [], [] processed_files.add(abs_path) ips = [] advanced_filters = [] try: with open(file_path, "r", encoding="utf-8") as f: for line in f: original_line = line.strip() if not original_line or original_line.startswith("#"): continue # Remove inline comments original_line = original_line.partition("#")[0].strip() if not original_line: continue # Handle Include statements if original_line.lower().startswith("include "): include_path = original_line[ 8: ].strip() # Remove 'include ' prefix logger.debug(f"Processing included file: {include_path}") if Path(include_path).exists(): included_ips, included_filters = ( get_ips_and_filters_from_csf_file( include_path, processed_files ) ) ips.extend(included_ips) advanced_filters.extend(included_filters) else: logger.warning( f"Included file not found: {include_path}" ) continue if "\|" in original_line: advanced_filter = get_advanced_filter_from_line( original_line ) if advanced_filter: advanced_filters.append(advanced_filter) logger.debug( f"Found advanced filter: {advanced_filter.original_line}" ) else: ip = get_ip_from_line(original_line) if ip: ips.append(ip) logger.debug(f"Found IP/CIDR: {ip}") except FileNotFoundError: logger.error(f"File not found: {file_path}") except PermissionError: logger.error(f"Permission denied reading file: {file_path}") except Exception as e: logger.error(f"Error reading file {file_path}: {e}") return ips, advanced_filters def execute_blocked_port_command( port: int, protocol: str, source_ip: str \| None = None ) -> bool: """ Execute imunify360-agent blocked-port command. Args: port: Port number protocol: "tcp" or "udp" source_ip: Optional source IP restriction Returns: True if command succeeded, False otherwise """ logger = get_logger() cmd = [ "imunify360-agent", "blocked-port", "add", "--comment", '"migrated from csf advanced filter"', f"{port}:{protocol}", ] if source_ip: cmd.extend(["--ips", source_ip]) logger.debug(f"Executing command: {' '.join(cmd)}") try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=30 ) if result.returncode == 0: ip_info = f" (source: {source_ip})" if source_ip else "" logger.info( f"Successfully added blocked port {port}:{protocol}{ip_info}" ) return True else: logger.error( f"Failed to add blocked port {port}:{protocol}: {result.stderr}" ) return False except subprocess.TimeoutExpired: logger.error( f"Timeout executing blocked-port command for {port}:{protocol}" ) return False except Exception as e: logger.error( f"Error executing blocked-port command for {port}:{protocol}: {e}" ) return False def validate_and_filter_ips( ips: list[str], ) -> tuple[list[str], list[dict], list[str]]: """ Validate a list of IPs and filter out invalid ones. Args: ips: List of IP strings to validate Returns: Tuple of (valid_ips, validation_errors, dangerous_ips) where: - valid_ips: List of valid IPs that can be processed - validation_errors: List of dicts with 'ip', 'error' - dangerous_ips: List of dangerous catch-all IPs that require manual handling """ valid_ips = [] validation_errors = [] dangerous_ips = [] for ip in ips: if is_dangerous_catchall_ip(ip): dangerous_ips.append(ip) else: is_valid, error_msg = validate_ip_for_imunify(ip) if is_valid: valid_ips.append(ip) else: validation_errors.append({"ip": ip, "error": error_msg}) return valid_ips, validation_errors, dangerous_ips def log_dangerous_ips_warning( dangerous_ips: list[str], list_type: str, context: str = "" ): """Log a special warning for dangerous catch-all IPs that require manual handling.""" if list_type == "whitelist": return if not dangerous_ips: return logger = get_logger() logger.warning(f"\n === DANGEROUS IP ADDRESSES DETECTED {context} ===") logger.warning( f"Found {len(dangerous_ips)} catch-all IP address(es) that would block ALL traffic:" ) for i, ip in enumerate(dangerous_ips, 1): logger.warning( f" {i}. '{ip}' - represents ALL addresses (0.0.0.0 or ::)" ) logger.warning( "\n WARNING: Adding these IPs to blacklist will block ALL connections!" ) logger.warning(" This could make your server completely inaccessible!") logger.warning( "\nIf you really want to add these IPs, you must do it manually:" ) for ip in dangerous_ips: logger.warning( f" imunify360-agent ip-list local add --purpose drop '{ip}'" ) logger.warning("\n PLEASE BE EXTREMELY CAREFUL - THIS MAY LOCK YOU OUT!") logger.warning("=== End Dangerous IP Warning ===\n") def log_validation_summary(validation_errors: list[dict], context: str = ""): """Log a summary of validation errors.""" if not validation_errors: return logger = get_logger() logger.warning(f"\n=== IP Validation Issues {context} ===") logger.warning(f"Found {len(validation_errors)} invalid IP(s)/CIDR(s):") for i, error_info in enumerate(validation_errors, 1): ip = error_info["ip"] error = error_info["error"] logger.warning(f" {i}. IP: '{ip}' - {error}") logger.warning("=== End IP Validation Issues ===\n") def execute_imunify_command(ips: list[str], list_type: str) -> bool: logger = get_logger() if list_type == "whitelist": list_purpose = "white" elif list_type == "blacklist": list_purpose = "drop" if not ips: logger.info(f"No IPs to add to {list_type}") return True cmd = [ "imunify360-agent", "ip-list", "local", "add", "--purpose", list_purpose, "--comment", '"migrated from csf"', *ips, ] logger.debug(f"Executing command: {' '.join(cmd)}") try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=60 ) if result.returncode == 0: logger.info(f"Successfully added {len(ips)} IPs to {list_type}") return True else: logger.error( f"Failed to add {ips} to {list_type}: {result.stderr}" ) return False except subprocess.TimeoutExpired: logger.error(f"Timeout executing command for {ips}") return False except Exception as e: logger.error(f"Error executing command for {ips}: {e}") return False def migrate_advanced_filters( advanced_filters: list[AdvancedFilter], file_type: str ) -> bool: """ Migrate CSF advanced filters to Imunify360 blocked ports. Only supports inbound direction filters. Args: advanced_filters: List of AdvancedFilter objects file_type: "allow" or "deny" - determines migration behavior Returns: True if all supported filters migrated successfully, False otherwise """ logger = get_logger() if not advanced_filters: logger.info( f"No advanced filters found to migrate from {file_type} file" ) return True logger.info( f"Processing {len(advanced_filters)} advanced filter(s) from {file_type} file" ) success = True migrated_count = 0 warning_count = 0 for filter_obj in advanced_filters: # Check if direction is supported (only 'in') if filter_obj.direction != "in": logger.warning( f"Unsupported direction '{filter_obj.direction}' - skipping filter: {filter_obj.original_line}" ) warning_count += 1 continue # Check if we have a destination port (required for blocked-port commands) if filter_obj.dest_port is None: logger.warning( f"No destination port found - skipping filter: {filter_obj.original_line}" ) warning_count += 1 continue # Check if source port is specified (not supported) if filter_obj.source_port is not None: logger.warning( f"Source port restrictions not supported in blocked-port migration - skipping filter: {filter_obj.original_line}" ) warning_count += 1 continue # Validate port range if not (1 <= filter_obj.dest_port <= 65535): logger.warning( f"Invalid port number {filter_obj.dest_port} - skipping filter: {filter_obj.original_line}" ) warning_count += 1 continue if file_type == "deny": if execute_blocked_port_command( filter_obj.dest_port, filter_obj.protocol, filter_obj.source_ip, ): migrated_count += 1 else: success = False elif file_type == "allow": # For allow file filters, we ignore all of them and just log warnings logger.warning( f"Allow file advanced filters are not migrated - skipping: {filter_obj.original_line}" ) warning_count += 1 logger.info(f""" Advanced filter migration summary for {file_type} file: - Migrated: {migrated_count} - Warnings/Skipped: {warning_count} """) if warning_count > 0: logger.warning( f"Some advanced filters from {file_type} file could not be migrated. Please review the warnings above." ) return success def migrate_ips() -> bool: """ Main function to migrate IPs and advanced filters from CSF to Imunify360. Returns: True if migration completed successfully, False otherwise """ logger = get_logger() try: logger.info( "Starting IP and advanced filter migration from CSF to Imunify360" ) success = True csf_files = { "/etc/csf/csf.allow": "whitelist", "/etc/csf/csf.deny": "blacklist", } # Track advanced filters by file type allow_advanced_filters = [] deny_advanced_filters = [] for file_path, list_type in csf_files.items(): logger.info(f"Processing {file_path} for {list_type}") if not Path(file_path).exists(): logger.warning(f"CSF file not found: {file_path}") continue ips, advanced_filters = get_ips_and_filters_from_csf_file( file_path ) # Migrate IPs if ips: logger.info(f"Found {len(ips)} IP(s)/CIDR(s) in {file_path}") valid_ips, validation_errors, dangerous_ips = ( validate_and_filter_ips(ips) ) if dangerous_ips: log_dangerous_ips_warning( dangerous_ips, list_type, f"from {file_path}" ) if validation_errors: log_validation_summary( validation_errors, f"from {file_path}" ) if valid_ips: total_issues = len(dangerous_ips) + len(validation_errors) if total_issues > 0: logger.info( f"Processing {len(valid_ips)} valid IP(s)/CIDR(s) out of {len(ips)} total ({len(dangerous_ips)} dangerous, {len(validation_errors)} invalid)" ) else: logger.info( f"Processing {len(valid_ips)} valid IP(s)/CIDR(s)" ) if not execute_imunify_command(valid_ips, list_type): success = False else: if dangerous_ips or validation_errors: logger.warning( f"No valid IPs found in {file_path} after validation" ) else: logger.info(f"No IPs to process from {file_path}") else: logger.info(f"No IPs found in {file_path}") # Categorize advanced filters by file type if advanced_filters: logger.info( f"Found {len(advanced_filters)} advanced filter(s) in {file_path}" ) if "allow" in file_path: allow_advanced_filters.extend(advanced_filters) else: deny_advanced_filters.extend(advanced_filters) else: logger.info(f"No advanced filters found in {file_path}") # Migrate advanced filters separately by file type logger.info("--- Advanced Filter Migration ---") # Migrate allow file advanced filters if allow_advanced_filters: logger.info("Processing allow file advanced filters:") if not migrate_advanced_filters(allow_advanced_filters, "allow"): success = False # Migrate deny file advanced filters if deny_advanced_filters: logger.info("Processing deny file advanced filters:") if not migrate_advanced_filters(deny_advanced_filters, "deny"): success = False if not allow_advanced_filters and not deny_advanced_filters: logger.info("No advanced filters found to migrate") logger.info("\n--- Migration Summary ---") if success: logger.info("Migration completed successfully") else: logger.warning("Migration completed with some errors") return success except Exception as e: logger.error(f"Error during IP and advanced filter migration: {e}") capture_exception(e, {"migration_type": "ips"}) return False Back

Edit File : ips.py | Size : 24.55 KB

import subprocess
import ipaddress
from pathlib import Path
from logger_config import get_logger, capture_exception

class AdvancedFilter:
    """Represents a parsed CSF advanced filter rule."""

def __init__(
        self,
        protocol: str = "tcp",
        direction: str = "in",
        source_ip: str | None = None,
        dest_port: int | None = None,
        source_port: int | None = None,
        original_line: str = "",
    ):
        self.protocol = protocol
        self.direction = direction
        self.source_ip = source_ip
        self.dest_port = dest_port
        self.source_port = source_port
        self.original_line = original_line

def __repr__(self):
        return f"AdvancedFilter(protocol={self.protocol}, direction={self.direction}, source_ip={self.source_ip}, dest_port={self.dest_port}, source_port={self.source_port})"

def is_valid_ipv6_network_for_imunify(ip_string: str) -> tuple[bool, str]:
    """
    Validate IPv6 network according to Imunify360 requirements.
    Imunify360 only supports IPv6 /64 networks.

Args:
        ip_string: IPv6 address or network in CIDR notation

Returns:
        Tuple of (is_valid, error_message)
    """
    if "/" not in ip_string:
        # Individual IPv6 addresses should be converted to /128 for validation
        # but Imunify360 doesn't support /128, so this is an error
        return (
            False,
            "Individual IPv6 addresses not supported, use /64 networks",
        )

try:
        # First check prefix length by parsing manually
        _, prefix_part = ip_string.split("/", 1)
        try:
            prefix_len = int(prefix_part)
        except ValueError:
            return False, "Invalid prefix length"

# Check if prefix is /64 (Imunify360 requirement)
        if prefix_len != 64:
            return False, "Supported only ipv6 /64 networks"

# Now check if it's a valid network with proper alignment
        ipaddress.IPv6Network(ip_string, strict=True)
        return True, ""
    except (
        ipaddress.AddressValueError,
        ipaddress.NetmaskValueError,
        ValueError,
    ) as e:
        error_msg = str(e)
        if "has host bits set" in error_msg:
            return False, "Supported only ipv6 /64 networks"
        return False, f"Invalid IPv6 network: {error_msg}"

def is_valid_ipv4_network_for_imunify(ip_string: str) -> tuple[bool, str]:
    """
    Validate IPv4 address or network according to Imunify360 requirements.

Args:
        ip_string: IPv4 address or network in CIDR notation

Returns:
        Tuple of (is_valid, error_message)
    """
    try:
        if "/" in ip_string:
            # CIDR notation - check if it's a proper network
            ipaddress.IPv4Network(ip_string, strict=True)
            return True, ""
        else:
            # Individual IP address
            ipaddress.IPv4Address(ip_string)
            return True, ""
    except ipaddress.AddressValueError as e:
        if "has host bits set" in str(e):
            return False, "has host bits set"
        return False, "Invalid IPv4 address/network"
    except (ipaddress.NetmaskValueError, ValueError) as e:
        return False, f"Invalid IPv4 address/network: {str(e)}"

def is_valid_ip_or_cidr(ip_string: str) -> bool:
    """Validate if a string is a valid IP address (IPv4 or IPv6) or CIDR notation."""
    try:
        # Try to parse as network (handles CIDR notation)
        if "/" in ip_string:
            ipaddress.ip_network(ip_string, strict=False)
        else:
            # Try to parse as individual IP address
            ipaddress.ip_address(ip_string)
        return True
    except (
        ipaddress.AddressValueError,
        ipaddress.NetmaskValueError,
        ValueError,
    ):
        return False

def is_dangerous_catchall_ip(ip_string: str) -> bool:
    """
    Check if an IP address or network represents a dangerous catch-all that could block all traffic.

This includes:
    - IPv4: 0.0.0.0 or any CIDR starting with 0.0.0.0
    - IPv6: :: or any CIDR starting with ::

Args:
        ip_string: IP address or network string to check

Returns:
        True if this is a dangerous catch-all IP/network
    """
    ip_string = ip_string.strip()

if ":" in ip_string:
        # IPv6 - check for :: (all zeros)
        if ip_string == "::" or ip_string.startswith("::/"):
            return True
        # Also check for explicit all-zeros representation
        if ip_string.startswith("0000:0000:0000:0000:0000:0000:0000:0000"):
            return True
    else:
        # IPv4 - check for 0.0.0.0
        if ip_string == "0.0.0.0" or ip_string.startswith("0.0.0.0/"):
            return True

return False

def validate_ip_for_imunify(ip_string: str) -> tuple[bool, str]:
    """
    Comprehensive validation for IP addresses/networks according to Imunify360 requirements.

Args:
        ip_string: IP address or network string to validate

Returns:
        Tuple of (is_valid, error_message)
    """
    if not ip_string or not ip_string.strip():
        return False, "Empty IP address"

ip_string = ip_string.strip()
    if is_dangerous_catchall_ip(ip_string):
        return (
            False,
            f"DANGEROUS: {ip_string} represents all addresses and would block all connections",
        )

# Determine if it's IPv4 or IPv6
    if ":" in ip_string:
        # IPv6
        return is_valid_ipv6_network_for_imunify(ip_string)
    else:
        # IPv4
        return is_valid_ipv4_network_for_imunify(ip_string)

def get_ip_from_line(line: str) -> str | None:
    """Extract and validate IP from a line, returning None if invalid."""
    if not is_valid_ip_or_cidr(line):
        return None

is_valid, error_msg = validate_ip_for_imunify(line)
    if is_valid:
        return line

logger = get_logger()
    msg = f"Invalid IP/CIDR '{line}': {error_msg}."
    if "dangerous" in error_msg.lower():
        logger.error(msg)
        return None
    logger.warning(msg)

return None

def get_advanced_filter_from_line(line: str) -> AdvancedFilter | None:
    """
    Parse a CSF advanced filter line.

Format: [protocol]|[direction]|[parameter=value]|[parameter=value]...
    or minimal: parameter=value

Examples:
    - udp|d=53|s=192.168.1.0/24
    - in|d=80|s=8.8.8.8
    - tcp|out|d=443|s=10.0.0.1
    - d=80  (minimal filter)

Args:
        line: CSF advanced filter line

Returns:
        AdvancedFilter object if successfully parsed, None otherwise
    """

parts = line.split("|")

# Require at least one meaningful part (not just empty strings)
    meaningful_parts = [p.strip() for p in parts if p.strip()]
    if len(meaningful_parts) < 1:
        return None

# Must have at least one parameter with "=" or be a protocol/direction
    has_parameter = any("=" in part for part in meaningful_parts)
    has_protocol_or_direction = any(
        part.lower() in ["tcp", "udp", "in", "out"]
        for part in meaningful_parts
    )

if not has_parameter and not has_protocol_or_direction:
        return None

# Initialize defaults
    protocol = "tcp"
    direction = "in"
    source_ip = None
    dest_port = None
    source_port = None

for part in meaningful_parts:
        part = part.strip()
        if not part:
            continue

# Check for protocol
        if part.lower() in ["tcp", "udp"]:
            protocol = part.lower()
        # Check for direction
        elif part.lower() in ["in", "out"]:
            direction = part.lower()
        # Check for parameter=value pairs
        elif "=" in part:
            param, value = part.split("=", 1)
            param = param.strip().lower()
            value = value.strip()

if param == "s":  # source
                try:
                    source_port = int(value)
                except ValueError:
                    # Treat as IP if not a valid port number
                    if is_valid_ip_or_cidr(value):
                        # Additional validation for Imunify360 requirements
                        is_valid, error_msg = validate_ip_for_imunify(value)
                        if is_valid:
                            source_ip = value
                        else:
                            logger = get_logger()
                            msg = f"Invalid source IP in advanced filter '{value}': {error_msg}."
                            if "dangerous" in error_msg.lower():
                                logger.error(msg)
                            else:
                                logger.warning(msg)

elif param == "d":  # destination
                try:
                    dest_port = int(value)
                except ValueError:
                    # For destination, we only support port numbers in this migration
                    pass

return AdvancedFilter(
        protocol=protocol,
        direction=direction,
        source_ip=source_ip,
        dest_port=dest_port,
        source_port=source_port,
        original_line=line,
    )

def get_ips_and_filters_from_csf_file(
    file_path: str, processed_files: set[str] | None = None
) -> tuple[list[str], list[AdvancedFilter]]:
    """
    Read a CSF file and extract both IP addresses/CIDR ranges and advanced filters.
    Handles Include statements recursively.

Args:
        file_path: Path to the CSF file
        processed_files: Set of already processed files to avoid infinite loops

Returns:
        Tuple of (List of IP addresses/CIDR ranges, List of AdvancedFilter objects)
    """
    logger = get_logger()

if processed_files is None:
        processed_files = set()

# Convert to absolute path to avoid processing same file multiple times
    abs_path = str(Path(file_path).resolve())
    if abs_path in processed_files:
        return [], []

processed_files.add(abs_path)

ips = []
    advanced_filters = []

try:
        with open(file_path, "r", encoding="utf-8") as f:
            for line in f:
                original_line = line.strip()
                if not original_line or original_line.startswith("#"):
                    continue

# Remove inline comments
                original_line = original_line.partition("#")[0].strip()
                if not original_line:
                    continue

# Handle Include statements
                if original_line.lower().startswith("include "):
                    include_path = original_line[
                        8:
                    ].strip()  # Remove 'include ' prefix
                    logger.debug(f"Processing included file: {include_path}")

if Path(include_path).exists():
                        included_ips, included_filters = (
                            get_ips_and_filters_from_csf_file(
                                include_path, processed_files
                            )
                        )
                        ips.extend(included_ips)
                        advanced_filters.extend(included_filters)
                    else:
                        logger.warning(
                            f"Included file not found: {include_path}"
                        )
                    continue

if "|" in original_line:
                    advanced_filter = get_advanced_filter_from_line(
                        original_line
                    )
                    if advanced_filter:
                        advanced_filters.append(advanced_filter)
                        logger.debug(
                            f"Found advanced filter: {advanced_filter.original_line}"
                        )
                else:
                    ip = get_ip_from_line(original_line)
                    if ip:
                        ips.append(ip)
                        logger.debug(f"Found IP/CIDR: {ip}")

except FileNotFoundError:
        logger.error(f"File not found: {file_path}")
    except PermissionError:
        logger.error(f"Permission denied reading file: {file_path}")
    except Exception as e:
        logger.error(f"Error reading file {file_path}: {e}")

return ips, advanced_filters

def execute_blocked_port_command(
    port: int, protocol: str, source_ip: str | None = None
) -> bool:
    """
    Execute imunify360-agent blocked-port command.

Args:
        port: Port number
        protocol: "tcp" or "udp"
        source_ip: Optional source IP restriction

Returns:
        True if command succeeded, False otherwise
    """
    logger = get_logger()

cmd = [
        "imunify360-agent",
        "blocked-port",
        "add",
        "--comment",
        '"migrated from csf advanced filter"',
        f"{port}:{protocol}",
    ]

if source_ip:
        cmd.extend(["--ips", source_ip])
    logger.debug(f"Executing command: {' '.join(cmd)}")

try:
        result = subprocess.run(
            cmd, capture_output=True, text=True, timeout=30
        )

if result.returncode == 0:
            ip_info = f" (source: {source_ip})" if source_ip else ""
            logger.info(
                f"Successfully added blocked port {port}:{protocol}{ip_info}"
            )
            return True
        else:
            logger.error(
                f"Failed to add blocked port {port}:{protocol}: {result.stderr}"
            )
            return False

except subprocess.TimeoutExpired:
        logger.error(
            f"Timeout executing blocked-port command for {port}:{protocol}"
        )
        return False
    except Exception as e:
        logger.error(
            f"Error executing blocked-port command for {port}:{protocol}: {e}"
        )
        return False

def validate_and_filter_ips(
    ips: list[str],
) -> tuple[list[str], list[dict], list[str]]:
    """
    Validate a list of IPs and filter out invalid ones.

Args:
        ips: List of IP strings to validate

Returns:
        Tuple of (valid_ips, validation_errors, dangerous_ips)
        where:
        - valid_ips: List of valid IPs that can be processed
        - validation_errors: List of dicts with 'ip', 'error'
        - dangerous_ips: List of dangerous catch-all IPs that require manual handling
    """
    valid_ips = []
    validation_errors = []
    dangerous_ips = []

for ip in ips:
        if is_dangerous_catchall_ip(ip):
            dangerous_ips.append(ip)
        else:
            is_valid, error_msg = validate_ip_for_imunify(ip)
            if is_valid:
                valid_ips.append(ip)
            else:
                validation_errors.append({"ip": ip, "error": error_msg})

return valid_ips, validation_errors, dangerous_ips

def log_dangerous_ips_warning(
    dangerous_ips: list[str], list_type: str, context: str = ""
):
    """Log a special warning for dangerous catch-all IPs that require manual handling."""
    if list_type == "whitelist":
        return

if not dangerous_ips:
        return

logger = get_logger()
    logger.warning(f"\n === DANGEROUS IP ADDRESSES DETECTED {context} ===")
    logger.warning(
        f"Found {len(dangerous_ips)} catch-all IP address(es) that would block ALL traffic:"
    )

for i, ip in enumerate(dangerous_ips, 1):
        logger.warning(
            f"  {i}. '{ip}' - represents ALL addresses (0.0.0.0 or ::)"
        )

logger.warning(
        "\n WARNING: Adding these IPs to blacklist will block ALL connections!"
    )
    logger.warning(" This could make your server completely inaccessible!")
    logger.warning(
        "\nIf you really want to add these IPs, you must do it manually:"
    )

for ip in dangerous_ips:
        logger.warning(
            f"  imunify360-agent ip-list local add --purpose drop '{ip}'"
        )

logger.warning("\n PLEASE BE EXTREMELY CAREFUL - THIS MAY LOCK YOU OUT!")
    logger.warning("=== End Dangerous IP Warning ===\n")

def log_validation_summary(validation_errors: list[dict], context: str = ""):
    """Log a summary of validation errors."""
    if not validation_errors:
        return

logger = get_logger()
    logger.warning(f"\n=== IP Validation Issues {context} ===")
    logger.warning(f"Found {len(validation_errors)} invalid IP(s)/CIDR(s):")

for i, error_info in enumerate(validation_errors, 1):
        ip = error_info["ip"]
        error = error_info["error"]

logger.warning(f"  {i}. IP: '{ip}' - {error}")

logger.warning("=== End IP Validation Issues ===\n")

def execute_imunify_command(ips: list[str], list_type: str) -> bool:
    logger = get_logger()
    if list_type == "whitelist":
        list_purpose = "white"
    elif list_type == "blacklist":
        list_purpose = "drop"

if not ips:
        logger.info(f"No IPs to add to {list_type}")
        return True

cmd = [
        "imunify360-agent",
        "ip-list",
        "local",
        "add",
        "--purpose",
        list_purpose,
        "--comment",
        '"migrated from csf"',
        *ips,
    ]
    logger.debug(f"Executing command: {' '.join(cmd)}")
    try:
        result = subprocess.run(
            cmd, capture_output=True, text=True, timeout=60
        )

if result.returncode == 0:
            logger.info(f"Successfully added {len(ips)} IPs to {list_type}")
            return True
        else:
            logger.error(
                f"Failed to add {ips} to {list_type}: {result.stderr}"
            )
            return False

except subprocess.TimeoutExpired:
        logger.error(f"Timeout executing command for {ips}")
        return False
    except Exception as e:
        logger.error(f"Error executing command for {ips}: {e}")
        return False

def migrate_advanced_filters(
    advanced_filters: list[AdvancedFilter], file_type: str
) -> bool:
    """
    Migrate CSF advanced filters to Imunify360 blocked ports.
    Only supports inbound direction filters.

Args:
        advanced_filters: List of AdvancedFilter objects
        file_type: "allow" or "deny" - determines migration behavior

Returns:
        True if all supported filters migrated successfully, False otherwise
    """
    logger = get_logger()

if not advanced_filters:
        logger.info(
            f"No advanced filters found to migrate from {file_type} file"
        )
        return True

logger.info(
        f"Processing {len(advanced_filters)} advanced filter(s) from {file_type} file"
    )

success = True
    migrated_count = 0
    warning_count = 0

for filter_obj in advanced_filters:
        # Check if direction is supported (only 'in')
        if filter_obj.direction != "in":
            logger.warning(
                f"Unsupported direction '{filter_obj.direction}' - skipping filter: {filter_obj.original_line}"
            )
            warning_count += 1
            continue

# Check if we have a destination port (required for blocked-port commands)
        if filter_obj.dest_port is None:
            logger.warning(
                f"No destination port found - skipping filter: {filter_obj.original_line}"
            )
            warning_count += 1
            continue

# Check if source port is specified (not supported)
        if filter_obj.source_port is not None:
            logger.warning(
                f"Source port restrictions not supported in blocked-port migration - skipping filter: {filter_obj.original_line}"
            )
            warning_count += 1
            continue

# Validate port range
        if not (1 <= filter_obj.dest_port <= 65535):
            logger.warning(
                f"Invalid port number {filter_obj.dest_port} - skipping filter: {filter_obj.original_line}"
            )
            warning_count += 1
            continue

if file_type == "deny":
            if execute_blocked_port_command(
                filter_obj.dest_port,
                filter_obj.protocol,
                filter_obj.source_ip,
            ):
                migrated_count += 1
            else:
                success = False
        elif file_type == "allow":
            # For allow file filters, we ignore all of them and just log warnings
            logger.warning(
                f"Allow file advanced filters are not migrated - skipping: {filter_obj.original_line}"
            )
            warning_count += 1

logger.info(f"""
Advanced filter migration summary for {file_type} file:
  - Migrated: {migrated_count}
  - Warnings/Skipped: {warning_count}
""")

if warning_count > 0:
        logger.warning(
            f"Some advanced filters from {file_type} file could not be migrated. Please review the warnings above."
        )

return success

def migrate_ips() -> bool:
    """
    Main function to migrate IPs and advanced filters from CSF to Imunify360.

Returns:
        True if migration completed successfully, False otherwise
    """
    logger = get_logger()
    try:
        logger.info(
            "Starting IP and advanced filter migration from CSF to Imunify360"
        )

success = True

csf_files = {
            "/etc/csf/csf.allow": "whitelist",
            "/etc/csf/csf.deny": "blacklist",
        }

# Track advanced filters by file type
        allow_advanced_filters = []
        deny_advanced_filters = []

for file_path, list_type in csf_files.items():
            logger.info(f"Processing {file_path} for {list_type}")

if not Path(file_path).exists():
                logger.warning(f"CSF file not found: {file_path}")
                continue

ips, advanced_filters = get_ips_and_filters_from_csf_file(
                file_path
            )

# Migrate IPs
            if ips:
                logger.info(f"Found {len(ips)} IP(s)/CIDR(s) in {file_path}")
                valid_ips, validation_errors, dangerous_ips = (
                    validate_and_filter_ips(ips)
                )

if dangerous_ips:
                    log_dangerous_ips_warning(
                        dangerous_ips, list_type, f"from {file_path}"
                    )

if validation_errors:
                    log_validation_summary(
                        validation_errors, f"from {file_path}"
                    )

if valid_ips:
                    total_issues = len(dangerous_ips) + len(validation_errors)
                    if total_issues > 0:
                        logger.info(
                            f"Processing {len(valid_ips)} valid IP(s)/CIDR(s) out of {len(ips)} total ({len(dangerous_ips)} dangerous, {len(validation_errors)} invalid)"
                        )
                    else:
                        logger.info(
                            f"Processing {len(valid_ips)} valid IP(s)/CIDR(s)"
                        )

if not execute_imunify_command(valid_ips, list_type):
                        success = False
                else:
                    if dangerous_ips or validation_errors:
                        logger.warning(
                            f"No valid IPs found in {file_path} after validation"
                        )
                    else:
                        logger.info(f"No IPs to process from {file_path}")
            else:
                logger.info(f"No IPs found in {file_path}")

# Categorize advanced filters by file type
            if advanced_filters:
                logger.info(
                    f"Found {len(advanced_filters)} advanced filter(s) in {file_path}"
                )
                if "allow" in file_path:
                    allow_advanced_filters.extend(advanced_filters)
                else:
                    deny_advanced_filters.extend(advanced_filters)
            else:
                logger.info(f"No advanced filters found in {file_path}")

# Migrate advanced filters separately by file type
        logger.info("--- Advanced Filter Migration ---")

# Migrate allow file advanced filters
        if allow_advanced_filters:
            logger.info("Processing allow file advanced filters:")
            if not migrate_advanced_filters(allow_advanced_filters, "allow"):
                success = False

# Migrate deny file advanced filters
        if deny_advanced_filters:
            logger.info("Processing deny file advanced filters:")
            if not migrate_advanced_filters(deny_advanced_filters, "deny"):
                success = False

if not allow_advanced_filters and not deny_advanced_filters:
            logger.info("No advanced filters found to migrate")

logger.info("\n--- Migration Summary ---")
        if success:
            logger.info("Migration completed successfully")
        else:
            logger.warning("Migration completed with some errors")

return success
    except Exception as e:
        logger.error(f"Error during IP and advanced filter migration: {e}")
        capture_exception(e, {"migration_type": "ips"})
        return False

Back