| Server IP : 104.21.30.238 / Your IP : 216.73.216.153 Web Server : LiteSpeed System : Linux premium151.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 : tempvsty ( 647) PHP Version : 8.0.30 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : OFF | Pkexec : OFF Directory : /./proc/thread-self/root/././opt/imunify360/venv/share/imunify360/scripts/migrate_csf/ |
Upload File : |
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