meraki_utils.py

#!/usr/bin/env python3
import ipaddress
import json
import os
import re
import sys
import unicodedata

import meraki
import requests


def is_valid_ip_or_network(value):
    try:
        ipaddress.ip_network(value, strict=False)
        return True
    except ValueError:
        return False


def is_valid_fqdn(value):
    pattern = re.compile(r"^(?=.{1,253}$)(?!-)[A-Za-z0-9-]{1,63}(?<!-)(\.[A-Za-z]{2,})+$")
    return bool(pattern.match(value))


def normalize_name(value):
    value = unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii")
    value = value.replace(".", "_").replace("/", "-").replace(",", "_").replace(" ", "_")
    value = re.sub(r"[^A-Za-z0-9_\-]", "_", value)
    return value[:50]


def load_and_validate(filename):
    objects = []
    errors = []

    if not os.path.exists(filename):
        print(f"File '{filename}' does not exist.")
        sys.exit(1)

    with open(filename, "r", encoding="utf-8") as file_handle:
        for line_number, raw_line in enumerate(file_handle, start=1):
            line = raw_line.strip()
            if not line or line.startswith("#"):
                continue

            parts = [part.strip() for part in line.split(",")]
            value = parts[0]
            custom_name = parts[1] if len(parts) > 1 and parts[1] else ""

            if is_valid_ip_or_network(value):
                obj_type = "ip"
            elif is_valid_fqdn(value):
                obj_type = "fqdn"
            else:
                errors.append(f"Line {line_number}: '{value}' has an invalid format.")
                continue

            if obj_type == "fqdn":
                name = normalize_name(custom_name) if custom_name else normalize_name(value)
            else:
                try:
                    network = ipaddress.ip_network(value, strict=False)
                    if network.prefixlen == 32:
                        base_name = f"H_{str(network.network_address).replace('.', '_')}"
                    else:
                        base_name = f"N_{str(network.network_address).replace('.', '_')}-{network.prefixlen}"
                except ValueError:
                    base_name = f"H_{value.replace('.', '_')}"

                if custom_name:
                    name = f"{base_name}_{normalize_name(custom_name)}"
                else:
                    name = base_name

            objects.append(
                {
                    "name": name[:50],
                    "type": obj_type,
                    "value": value,
                }
            )

    return objects, errors


def create_network_object(dashboard, org_id, obj):
    """Create a policy object, or return an existing object ID on conflict."""
    try:
        if obj["type"] == "ip":
            payload = {
                "name": obj["name"],
                "category": "network",
                "type": "cidr",
                "cidr": obj["value"],
                "ip": obj["value"].split("/")[0],
                "groupIds": [],
            }
        else:
            payload = {
                "name": obj["name"],
                "category": "network",
                "type": "fqdn",
                "fqdn": obj["value"],
                "groupIds": [],
            }

        return dashboard.organizations.createOrganizationPolicyObject(org_id, **payload)

    except meraki.APIError as error:
        if error.status in (400, 409):
            print(f"Object '{obj['name']}' already exists. Looking up existing ID...")
            existing_objects = dashboard.organizations.getOrganizationPolicyObjects(org_id)

            for existing in existing_objects:
                if (
                    existing["name"] == obj["name"]
                    or existing.get("cidr") == obj["value"]
                    or existing.get("fqdn") == obj["value"]
                ):
                    print(f"Found existing object ID: {existing['id']}")
                    return {"id": existing["id"]}

            print(f"Could not resolve object ID for '{obj['name']}' after status {error.status}.")
            return None

        raise Exception(f"Failed to create object {obj['name']}: {error}")


def create_network_object_group(api_key, org_id, name, object_ids):
    url = f"https://api.meraki.com/api/v1/organizations/{org_id}/policyObjects/groups"
    headers = {
        "X-Cisco-Meraki-API-Key": api_key,
        "Content-Type": "application/json",
    }
    payload = {
        "name": name,
        "category": "NetworkObjectGroup",
        "objectIds": object_ids,
    }

    response = requests.post(url, headers=headers, json=payload, timeout=60)
    if response.status_code == 201:
        return response.json()
    if response.status_code == 400 and "Name already exists" in response.text:
        print(f"Group '{name}' already exists. Skipping group creation.")
        return None

    raise Exception(f"Failed to create group {name}: {response.status_code} {response.text}")


def save_results(path, created_ids, failed):
    with open(path, "w", encoding="utf-8") as file_handle:
        json.dump({"created": created_ids, "failed": failed}, file_handle, indent=2, ensure_ascii=False)