Python script to retrieve and filter DHCP reservations from a Windows server based on MAC prefixes and output the results in CSV format.

pi_collector.py

import subprocess
import json
import csv
import io

# List of MAC address prefixes to match (in lowercase for easier comparison)
mac_prefixes = [
    "28-CD-C1",
    "B8-27-EB",
    "D8-3A-DD",
    "DC-A6-32",
    "E4-5F-01"
]

# Option to include header in the CSV output
include_header = False

# Function to execute a PowerShell command using subprocess
def run_powershell_command(command):
    completed = subprocess.run(["powershell.exe", "-ExecutionPolicy", "Bypass", "-Command", command], capture_output=True, text=True)
    if completed.returncode != 0:
        raise Exception(f"Command failed with return code {completed.returncode}: {completed.stderr}")
    return completed.stdout.strip()

# PowerShell command to get all DHCP reservations for all scopes
ps_command = """
$scopes = Get-DhcpServerv4Scope;
$allReservations = foreach ($scope in $scopes) {
    Get-DhcpServerv4Lease -ScopeId $scope.ScopeId | Select-Object HostName, ClientId, IPAddress
};
$allReservations | ConvertTo-Json -Compress
"""

# Execute the PowerShell command and get the output
try:
    result = run_powershell_command(ps_command)
except Exception as e:
    print(f"Error executing PowerShell command: {e}")
    exit(1)

# Parse the JSON output from PowerShell
try:
    dhcp_reservations = json.loads(result)
except json.JSONDecodeError as e:
    print(f"Error parsing the PowerShell output: {e}")
    exit(1)

# Filter DHCP reservations based on the provided MAC address prefixes
filtered_reservations = []
for reservation in dhcp_reservations:
    mac_address = reservation.get('ClientId', '').lower()
    # Check if the MAC address starts with any of the prefixes
    if any(mac_address.startswith(prefix.lower()) for prefix in mac_prefixes):
        filtered_reservations.append({
            "HostName": reservation.get('HostName') or '',
            "MacAddress": reservation.get('ClientId') or '',
            "IPAddress": reservation.get('IPAddress', {}).get('IPAddressToString', 'Unknown')
        })

# Prepare the CSV output to print to the console
csv_output = io.StringIO()
fieldnames = ['HostName', 'MacAddress', 'IPAddress']
writer = csv.DictWriter(csv_output, fieldnames=fieldnames)

# Write header if include_header is True
if include_header:
    writer.writeheader()

# Write each reservation to the CSV output
for res in filtered_reservations:
    writer.writerow(res)

# Print the CSV output to the console
print(csv_output.getvalue())