Admin Control Panel v2.0

✎ Editing: ips.py ← Back

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

Cancel