CSEC_PUBLIC/securityonion
1
0
Fork 0
mirror of https://github.com/Security-Onion-Solutions/securityonion.git synced 2025-12-09 02:32:46 +01:00
Code Issues Packages Projects Releases Wiki Activity

create predicatble interfaces

Browse Source
This commit is contained in:
m0duspwnens
2025-02-06 15:30:46 -05:00
parent c8e232c598
commit 7159678385
13 changed files with 792 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

378
salt/hypervisor/tools/sbin/so-qcow2-network-predictable Normal file
View File

@@ -0,0 +1,378 @@
#!/usr/bin/python3
# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
# https://securityonion.net/license; you may not use this file except in compliance with the
# Elastic License 2.0.
"""
Script for configuring network interface predictability in Security Onion VMs.
This script modifies the necessary files to ensure consistent network interface naming.
The script performs the following operations:
1. Modifies the BLS entry to set net.ifnames=1
2. Removes any existing persistent network rules
3. Updates GRUB configuration
**Usage:**
so-qcow2-network-predictable -n <domain_name> [-I <qcow2_image_path>]
**Options:**
-n, --name Domain name of the VM to configure
-I, --image (Optional) Path to the QCOW2 image. If not provided,
defaults to /nsm/libvirt/images/<domain_name>/<domain_name>.qcow2
**Examples:**
1. **Configure using domain name:**
```bash
so-qcow2-network-predictable -n sool9
```
This command will:
- Use default image path: /nsm/libvirt/images/sool9/sool9.qcow2
- Configure network interface predictability
2. **Configure using custom image path:**
```bash
so-qcow2-network-predictable -n sool9 -I /path/to/custom/image.qcow2
```
This command will:
- Use the specified image path
- Configure network interface predictability
**Notes:**
- The VM must not be running when executing this script
- Requires root privileges
- Will automatically find and modify the appropriate BLS entry
- Removes /etc/udev/rules.d/70-persistent-net.rules if it exists
- Updates GRUB configuration after changes
**Exit Codes:**
- 0: Success
- 1: General error (invalid arguments, file operations, etc.)
- 2: VM is running
- 3: Required files not found
- 4: Permission denied
**Logging:**
- Logs are written to /opt/so/log/hypervisor/so-qcow2-network-predictable.log
- Both file and console logging are enabled
- Log entries include:
- Timestamps
- Operation details
- Error messages
- Configuration changes
"""
import argparse
import guestfs
import glob
import libvirt
import logging
import os
import re
import sys
from so_logging_utils import setup_logging
# Set up logging
logger = setup_logging(
logger_name='so-qcow2-network-predictable',
log_file_path='/opt/so/log/hypervisor/so-qcow2-network-predictable.log',
log_level=logging.INFO,
format_str='%(asctime)s - %(levelname)s - %(message)s'
)
def check_domain_status(domain_name):
"""
Check if the specified domain exists and is not running.
Args:
domain_name (str): Name of the libvirt domain to check
Returns:
bool: True if domain exists and is not running, False otherwise
Raises:
RuntimeError: If domain is running or connection to libvirt fails
"""
try:
conn = libvirt.open('qemu:///system')
try:
dom = conn.lookupByName(domain_name)
is_running = dom.isActive()
if is_running:
logger.error(f"Domain '{domain_name}' is running - cannot modify configuration")
raise RuntimeError(f"Domain '{domain_name}' must not be running")
logger.info(f"Domain '{domain_name}' exists and is not running")
return True
except libvirt.libvirtError as e:
if "no domain with matching name" in str(e):
logger.error(f"Domain '{domain_name}' not found")
raise RuntimeError(f"Domain '{domain_name}' not found")
raise
finally:
conn.close()
except libvirt.libvirtError as e:
logger.error(f"Failed to connect to libvirt: {e}")
raise RuntimeError(f"Failed to connect to libvirt: {e}")
def modify_bls_entry(g):
"""
Find and modify the BLS entry to set net.ifnames=1.
Args:
g: Mounted guestfs handle
Returns:
bool: True if successful, False if no changes needed
Raises:
RuntimeError: If BLS entry cannot be found or modified
"""
bls_dir = "/boot/loader/entries"
logger.info(f"Checking BLS directory: {bls_dir}")
if g.is_dir(bls_dir):
logger.info("BLS directory exists")
else:
logger.info("Listing /boot contents:")
try:
boot_contents = g.ls("/boot")
logger.info(f"/boot contains: {boot_contents}")
if g.is_dir("/boot/loader"):
logger.info("Listing /boot/loader contents:")
loader_contents = g.ls("/boot/loader")
logger.info(f"/boot/loader contains: {loader_contents}")
except Exception as e:
logger.error(f"Error listing /boot contents: {e}")
raise RuntimeError(f"BLS directory not found: {bls_dir}")
# Find BLS entry file
entries = g.glob_expand(f"{bls_dir}/*.conf")
logger.info(f"Found BLS entries: {entries}")
if not entries:
logger.error("No BLS entry files found")
raise RuntimeError("No BLS entry files found")
# Use the first entry found
bls_file = entries[0]
logger.info(f"Found BLS entry file: {bls_file}")
try:
logger.info(f"Reading BLS file contents from: {bls_file}")
content = g.read_file(bls_file).decode('utf-8')
logger.info("Current BLS file content:")
logger.info("---BEGIN BLS CONTENT---")
logger.info(content)
logger.info("---END BLS CONTENT---")
lines = content.splitlines()
modified = False
for i, line in enumerate(lines):
if line.startswith('options '):
logger.info(f"Found options line: {line}")
# Check if net.ifnames parameter exists
if 'net.ifnames=' in line:
# Replace existing parameter
new_line = re.sub(r'net\.ifnames=[01]', 'net.ifnames=1', line)
if new_line != line:
lines[i] = new_line
modified = True
logger.info(f"Updated existing net.ifnames parameter to 1. New line: {new_line}")
else:
# Add parameter
lines[i] = f"{line} net.ifnames=1"
modified = True
logger.info(f"Added net.ifnames=1 parameter. New line: {lines[i]}")
break
if modified:
new_content = '\n'.join(lines) + '\n'
logger.info("New BLS file content:")
logger.info("---BEGIN NEW BLS CONTENT---")
logger.info(new_content)
logger.info("---END NEW BLS CONTENT---")
g.write(bls_file, new_content.encode('utf-8'))
logger.info("Successfully updated BLS entry")
return True
logger.info("No changes needed for BLS entry")
return False
except Exception as e:
logger.error(f"Failed to modify BLS entry: {e}")
raise RuntimeError(f"Failed to modify BLS entry: {e}")
def remove_persistent_net_rules(g):
"""
Remove the persistent network rules file if it exists.
Args:
g: Mounted guestfs handle
Returns:
bool: True if file was removed, False if it didn't exist
"""
rules_file = "/etc/udev/rules.d/70-persistent-net.rules"
logger.info(f"Checking for persistent network rules file: {rules_file}")
try:
if g.is_file(rules_file):
logger.info("Found persistent network rules file, removing...")
g.rm(rules_file)
logger.info(f"Successfully removed persistent network rules file: {rules_file}")
return True
logger.info("No persistent network rules file found")
return False
except Exception as e:
logger.error(f"Failed to remove persistent network rules: {e}")
raise RuntimeError(f"Failed to remove persistent network rules: {e}")
def update_grub_config(g):
"""
Update GRUB configuration.
Args:
g: Mounted guestfs handle
Raises:
RuntimeError: If GRUB update fails
"""
try:
logger.info("Updating GRUB configuration...")
output = g.command(['grub2-mkconfig', '-o', '/boot/grub2/grub.cfg'])
logger.info("GRUB update output:")
logger.info(output)
logger.info("Successfully updated GRUB configuration")
except Exception as e:
logger.error(f"Failed to update GRUB configuration: {e}")
raise RuntimeError(f"Failed to update GRUB configuration: {e}")
def configure_network_predictability(domain_name, image_path=None):
"""
Configure network interface predictability for a VM.
Args:
domain_name (str): Name of the domain to configure
image_path (str, optional): Path to the QCOW2 image
Raises:
RuntimeError: If configuration fails
"""
# Check domain status
check_domain_status(domain_name)
# Use default image path if none provided
if not image_path:
image_path = f"/nsm/libvirt/images/{domain_name}/{domain_name}.qcow2"
if not os.path.exists(image_path):
logger.error(f"Image file not found: {image_path}")
raise RuntimeError(f"Image file not found: {image_path}")
if not os.access(image_path, os.R_OK | os.W_OK):
logger.error(f"Permission denied: Cannot access image file {image_path}")
raise RuntimeError(f"Permission denied: Cannot access image file {image_path}")
logger.info(f"Configuring network predictability for domain: {domain_name}")
logger.info(f"Using image: {image_path}")
g = guestfs.GuestFS(python_return_dict=True)
try:
logger.info("Initializing guestfs...")
g.set_network(False)
g.selinux = False
g.add_drive_opts(image_path, format="qcow2")
g.launch()
logger.info("Inspecting operating system...")
roots = g.inspect_os()
if not roots:
raise RuntimeError("No operating system found in image")
root = roots[0]
logger.info(f"Found root filesystem: {root}")
logger.info(f"Operating system type: {g.inspect_get_type(root)}")
logger.info(f"Operating system distro: {g.inspect_get_distro(root)}")
logger.info(f"Operating system major version: {g.inspect_get_major_version(root)}")
logger.info(f"Operating system minor version: {g.inspect_get_minor_version(root)}")
logger.info("Getting mount points...")
mountpoints = g.inspect_get_mountpoints(root)
logger.info(f"Found mount points: {mountpoints}")
logger.info("Converting mount points to sortable list...")
# Convert dictionary to list of tuples
mountpoints = list(mountpoints.items())
logger.info(f"Converted mount points: {mountpoints}")
logger.info("Sorting mount points by path length for proper mount order...")
mountpoints.sort(key=lambda m: len(m[0]))
logger.info(f"Mount order will be: {[mp[0] for mp in mountpoints]}")
for mp_path, mp_device in mountpoints:
try:
logger.info(f"Attempting to mount {mp_device} at {mp_path}")
g.mount(mp_device, mp_path)
logger.info(f"Successfully mounted {mp_device} at {mp_path}")
except Exception as e:
logger.warning(f"Could not mount {mp_device} at {mp_path}: {str(e)}")
# Continue with other mounts
# Perform configuration steps
bls_modified = modify_bls_entry(g)
rules_removed = remove_persistent_net_rules(g)
if bls_modified or rules_removed:
update_grub_config(g)
logger.info("Network predictability configuration completed successfully")
else:
logger.info("No changes were necessary")
except Exception as e:
raise RuntimeError(f"Failed to configure network predictability: {e}")
finally:
try:
logger.info("Unmounting all filesystems...")
g.umount_all()
logger.info("Successfully unmounted all filesystems")
except Exception as e:
logger.warning(f"Error unmounting filesystems: {e}")
g.close()
def parse_arguments():
"""Parse command line arguments."""
parser = argparse.ArgumentParser(
description="Configure network interface predictability for Security Onion VMs"
)
parser.add_argument("-n", "--name", required=True,
help="Domain name of the VM to configure")
parser.add_argument("-I", "--image",
help="Path to the QCOW2 image (optional)")
return parser.parse_args()
def main():
"""Main entry point for the script."""
try:
args = parse_arguments()
configure_network_predictability(args.name, args.image)
sys.exit(0)
except RuntimeError as e:
if "must not be running" in str(e):
logger.error(str(e))
sys.exit(2)
elif "not found" in str(e):
logger.error(str(e))
sys.exit(3)
elif "Permission denied" in str(e):
logger.error(str(e))
sys.exit(4)
else:
logger.error(str(e))
sys.exit(1)
except KeyboardInterrupt:
logger.error("Operation cancelled by user")
sys.exit(1)
except Exception as e:
logger.error(f"Unexpected error: {e}")
sys.exit(1)
if __name__ == "__main__":
main()