merakicopyssid

#!/usr/bin/env python3

"""
merakicopyssid.py

Copy an existing SSID.  Can be used to copy within the same network or to another network.

Before running the script, it's best to put the APIKEY
into the MERAKI_DASHBOARD_API_KEY environment variable.  Otherwise
use the less security --apikey option.

export MERAKI_DASHBOARD_API_KEY="the_api_key_here"

"""

import getopt
import json
import os
import sys

import requests
from attr import attrib, attrs

import meraki

DEBUG = False

# pylint: disable=too-many-instance-attributes

@attrs
class NetworkSSID():
    network_name = attrib(default=None, type=str)
    network_id = attrib(default=None, type=str)
    SSID_number = attrib(default=None, type=int)
    SSID_name = attrib(default=None, type=str)
    devices = attrib(default=None, type=list)

    # Set to True if SSID 'enabled' should be set to False
    set_disabled = attrib(default=False)

class MerakiCopySSID:
    """
    Copy a Meraki SSID configuration
    """
    def __init__(self, verbose=False):
        self.verbose = verbose
        self.apikey = None
        self.organization = None
        self.dashboard = None
        self.networks_by_name = {}
        self.source = NetworkSSID()
        self.destination = NetworkSSID()
        self.secret = None

    def get_json(self, url: str) -> dict:
        """Perform REST call to Meraki with the provided URL.

        Arguments:
            url {str} -- URL to specific API

        Returns:
            dict -- Results from call.  May be False.
        """

        headers = {}
        headers['Content-Type'] = 'application/json'
        headers['X-Cisco-Meraki-API-Key'] = self.apikey

        try:
            if self.verbose:
                print(f'get_json: {url}')
            req = requests.get(url, headers=headers)
        except requests.exceptions.RequestException as exception:
            print("Error: %s" % exception)
            return False
        except KeyboardInterrupt:
            print("Aborted")
            sys.exit(25)

        if self.verbose:
            print("r.status_code = %s" % (req.status_code))
            print("r.text = %s" % (req.text))

        if req.status_code == 400:
            if self.verbose:
                print("400 Bad request: %s : %s" % (url, req.text))
            return False
        if req.status_code == 404:
            if self.verbose:
                print("404 Not found: %s" % (url))
            return False
        if req.status_code != 200:
            print("Error: %s - %s" % (req.status_code, req.text))
            return False

        data = json.loads(req.text)

        return data

    def put_json(self, url: str, data: dict) -> dict:
        """Perform REST call to Meraki with the provided URL.

        Arguments:
            url {str} -- URL to specific API
            data {dict} -- data to send in JSON payload

        Returns:
            dict -- Results from call.  May be False.
        """

        headers = {}
        headers['Content-Type'] = 'application/json'
        headers['X-Cisco-Meraki-API-Key'] = self.apikey

        try:
            if self.verbose:
                print(f'put_json: {url}')
            req = requests.put(url, headers=headers, json=data)
        except requests.exceptions.RequestException as exception:
            print("Error: %s" % exception)
            return False
        except KeyboardInterrupt:
            print("Aborted")
            sys.exit(25)

        if self.verbose:
            print("r.status_code = %s" % (req.status_code))
            print("r.text = %s" % (req.text))

        if req.status_code == 400:
            if self.verbose:
                print("400 Bad request: %s : %s" % (url, req.text))
            return False
        if req.status_code == 404:
            if self.verbose:
                print("404 Not found: %s" % (url))
            return False
        if req.status_code != 200:
            print("Error: %s - %s" % (req.status_code, req.text))
            return False

        data = json.loads(req.text)

        return data

    def getNetworkSsids(self, networkId: str) -> list:
        """Get the older API Network SSIDs as used by wireless on Z or MX models

        Args:
            networkId (str): [description]

        Returns:
            list: [description]
        """
        url = "https://dashboard.meraki.com/api/v0/networks/%s/ssids" % (networkId)
        return self.get_json(url)

    def getNetworkSsid(self, networkId: str, number: int) -> dict:
        """Get the older API Network SSIDs as used by wireless on Z or MX models

        Args:
            networkId (str): [description]

        Returns:
            list: [description]
        """
        url = "https://dashboard.meraki.com/api/v0/networks/%s/ssids/%d" % (networkId, number)
        return self.get_json(url)

    def putNetworkSsid(self, networkId: str, number: int, settings: dict) -> dict:
        """Put using the older API Network SSIDs as used by wireless on Z or MX models

        Args:
            networkId (str): [description]
            settings (dict): settings for the SSID

        Returns:
            list: [description]
        """
        url = "https://dashboard.meraki.com/api/v0/networks/%s/ssids/%d" % (networkId, number)
        return self.put_json(url, settings)

    def getNetworkWirelessSsids(self, networkId: str) -> dict:
        try:
            return self.dashboard.wireless.getNetworkWirelessSsids(networkId)
        except meraki.exceptions.APIError:
            return None

    def getNetworkWirelessSsid(self, networkId: str, SSID_number: int) -> dict:
        try:
            return self.dashboard.wireless.getNetworkWirelessSsid(networkId, SSID_number)
        except meraki.exceptions.APIError:
            return None

    def updateNetworkWirelessSsid(self, networkId: str, SSID_number: int, settings: dict) -> dict:
        try:
            return self.dashboard.wireless.updateNetworkWirelessSsid(networkId, str(SSID_number), **settings)
        except meraki.exceptions.APIError:
            return None

    def getNetworkDevices(self, networkId: str) -> dict:
        """Get all Meraki devices for a network.

        Args:
            networkId (str): [description]

        Returns:
            dict: [description]
        """
        try:
            return self.dashboard.networks.getNetworkDevices(networkId)
        except meraki.exceptions.APIError:
            return None

    def listSSIDs(self, network: NetworkSSID):

        if network.network_name in self.networks_by_name:
            networkId = self.networks_by_name[network.network_name]['id']
        else:
            print(f"Error: unknown network: {network.network_name}")
            return

        if self.has_mr(network):
            ssids = self.getNetworkWirelessSsids(networkId)
        else:
            ssids = self.getNetworkSsids(networkId)

        # ssids = self.getNetworkWirelessSsids(networkId)
        if ssids:
            for ssid in ssids:
                if self.verbose:
                    print("ssid: %s" % (ssid['name']))
                print('{},{},{},{},"{}"'.format(ssid['number'], ssid['name'], \
                    ssid['enabled'], ssid['authMode'], ssid['ipAssignmentMode']))

    def get_org_id(self, organization_id: str) -> str:

        organizations = self.dashboard.organizations.getOrganizations()
        if not organizations:
            print("Error: No organizations found")
            return None

        if len(organizations) > 1 and not organization_id:
            print("Error: more than one organization found.  Please specify one by id with -o option.")
            for organization in organizations:
                print("{},{}".format(organization['id'], organization['name']))
            return None

        if len(organizations) == 1:
            return organizations[0]['id']

        for organization in organizations:
            if self.verbose:
                #print("org={}".format(organizations))
                print("Organization: {} {}".format(organization['id'], organization['name']))
            if organization_id == organization['id']:
                return organization['id']

        print("Error: organization not found: {}".format(organization_id))
        return None

    def get_networks_by_name(self):
        networks = self.dashboard.organizations.getOrganizationNetworks(self.organization)
        if networks:
            for network in networks:
                if self.verbose:
                    print("Network: {}".format(network['name']))
                self.networks_by_name[network['name'].lower()] = network
        else:
            print("Error: no networks found in organization")

    def has_mr(self, network: NetworkSSID) -> bool:
        if not network.devices:
            network.devices = self.getNetworkDevices(network.network_id)
        # print(f'devices={devices}')
        for device in network.devices:
            if device['model'].startswith('MR'):
                return True
        return False

    def copy(self):

        if self.source.network_name not in self.networks_by_name:
            print("Error: {} is unknown network".format(self.source.network_name))
            return False

        if self.destination.network_name not in self.networks_by_name:
            print("Error: {} is unknown network".format(self.destination.network_name))
            return False

        if self.has_mr(self.source):
            src_ssid = self.getNetworkWirelessSsid(self.source.network_id, self.source.SSID_number)
        else:
            src_ssid = self.getNetworkSsid(self.source.network_id, self.source.SSID_number)

        if self.verbose:
            print("src_ssid={}".format(json.dumps(src_ssid, indent=2, sort_keys=True)))

        dst_ssid = {}
        #
        # Unfortunately the Meraki API is not perfect.
        # Not all keys should be copied over.
        #
        for key in src_ssid:
            if src_ssid[key] is not None:
                if key == 'number':
                    # the API doesn't like the number also in the kwargs despite what the
                    # documentation says
                    pass
                elif key == 'encryptionMode':
                    if 'authMode' in src_ssid and src_ssid['authMode'] != '8021x-radius':
                        dst_ssid[key] = src_ssid[key]
                else:
                    dst_ssid[key] = src_ssid[key]

                if key in ['radiusServers', 'radiusAccountingServers']:
                    if self.secret:
                        for server in dst_ssid[key]:
                            server['secret'] = self.secret
                    else:
                        print(f"Error: {key} detected.  You must specify a --secret")
                        return False
                    # unfortunately the API returns values set to null
                    # but you cannot send values that are null
                    new_servers = []
                    for server in src_ssid[key]:
                        new_server = {}
                        for attr in server.keys():
                            if server[attr] is not None:
                                new_server[attr] = server[attr]
                        new_servers.append(new_server)
                    dst_ssid[key] = new_servers

        if self.destination.SSID_name:
            # override the SSID name
            dst_ssid['name'] = self.destination.SSID_name

        if self.destination.set_disabled:
            dst_ssid['enabled'] = False

        if self.verbose:
            print("dst_ssid={}".format(json.dumps(dst_ssid, indent=2, sort_keys=True)))
        if self.has_mr(self.destination):
            self.updateNetworkWirelessSsid(self.destination.network_id, self.destination.SSID_number, dst_ssid)
        else:
            self.putNetworkSsid(self.destination.network_id, self.destination.SSID_number, dst_ssid)
        return True

    def check_src_dst(self) -> bool: #pylint: disable=too-many-return-statements

        if not self.source.network_name:
            print("Error: please specify a source network with --srcnet")
            return True

        self.source.network_id = self.networks_by_name[self.source.network_name]['id']

        if self.source.SSID_number is None:
            print("Error: please specify a source SSID number with --srcssid")
            self.listSSIDs(self.source)
            return True
        if not (self.source.SSID_number is not None and
                  self.source.SSID_number > -1 and self.source.SSID_number < 32):
            print("Error: source SSID number must be 0 to 32")
            return True
        if not self.destination.network_name:
            print("Error: please specify a destination network with --dstnet")
            return True

        self.destination.network_id = self.networks_by_name[self.destination.network_name]['id']

        if self.destination.SSID_number is None:
            print("Error: please specify a destination SSID number with --dstssid")
            self.listSSIDs(self.destination)
            return True
        if not (self.destination.SSID_number is not None and
                  self.destination.SSID_number > -1 and self.destination.SSID_number < 32):
            print("Error: destination SSID number must be 0 to 32")
            return True
        return False

    @staticmethod
    def usage():
        print("Usage:\n"
              "\t--apikey       - {API key}\n"
              "\t--dstdisabled  - set the destination SSID to disabled\n"
              "\t--dstnet       - name of destination network\n"
              "\t--dstssid      - number of destination SSID (0 - 31)\n"
              "\t--dstssidname  - new name for destination SSID\n"
              "\t--organization - ID of organization.  Required if API key has access to more than one.\n"
              "\t--secret       - RADIUS secret\n"
              "\t--srcnet       - name of source network\n"
              "\t--srcssid      - number of source SSID (0 - 31)\n"
              "\t--verbose      - verbose output\n"
             )

    def main(self):
        """
        Parse command line options and perform copy.
        """
        organization_id = None
        try:
            options = getopt.getopt(sys.argv[1:],
                                    'a:o:v',
                                    [
                                        'apikey=',
                                        'dstdisabled',
                                        'dstnet=',
                                        'dstssid=',
                                        'dstssidname=',
                                        'secret=',
                                        'srcnet=',
                                        'srcssid=',
                                        'organization=',
                                        'verbose',
                                    ])[0]
        except getopt.GetoptError as err:
            print(err)
            self.usage()
            sys.exit(2)

        for opt, arg in options:
            if opt in ('-a', '--apikey'):
                self.apikey = arg
            elif opt in ('-o', '--organization'):
                organization_id = arg
            elif opt == '--dstdisabled':
                self.destination.set_disabled = True
            elif opt == '--dstnet':
                self.destination.network_name = arg.lower()
            elif opt == '--dstssid':
                self.destination.SSID_number = int(arg)
            elif opt == '--dstssidname':
                self.destination.SSID_name = arg.lower()
            elif opt == '--secret':
                self.secret = arg
            elif opt == '--srcnet':
                self.source.network_name = arg.lower()
            elif opt == '--srcssid':
                self.source.SSID_number = int(arg)
            elif opt in ('-v', '--verbose'):
                self.verbose = True

        if 'MERAKI_APIKEY' in os.environ:
            self.apikey = os.environ['MERAKI_APIKEY']
        if 'MERAKI_DASHBOARD_API_KEY' in os.environ:
            self.apikey = os.environ['MERAKI_DASHBOARD_API_KEY']
        if 'MERAKI_ORGID' in os.environ:
            organization_id = os.environ['MERAKI_ORGID']

        if not self.apikey:
            print("Missing apikey")
            sys.exit(2)

        try:
            self.dashboard = meraki.DashboardAPI(api_key=self.apikey,
                                                 output_log=DEBUG,
                                                 print_console=self.verbose)
        except meraki.exceptions.APIKeyError as exception:
            print("Error: {}".format(exception))
            return

        self.organization = self.get_org_id(organization_id)
        if self.organization:
            self.get_networks_by_name()
            if self.check_src_dst():
                return
            self.copy()

if __name__ == "__main__":
    MerakiCopySSID().main()