initc9300_ports.py

""" Init Switch ports to use mgmt vlan as native and allowed-vlan trunks """

import argparse                             # Parse arguments: Switch_Serial
import os                                   # Gets API and USER from environment
import inspect                              # Gets current calling file
import meraki                               # Meraki REST API


def parent_caller():
    """
    Return parent function's filename
    :return:
    """
    filename = ""
    previous_frame = inspect.currentframe().f_back.f_back
    if previous_frame:
        (
            filename,
            line_number,
            function_name,
            lines,
            index,
        ) = inspect.getframeinfo(previous_frame)
        if filename:
            filename = os.path.basename(filename)
    return filename


def new_dash(api_key, base_url='https://api.meraki.com/api/v1/'):
    """ Initiate a connection to dashboard """
    username = os.getenv('USER')
    username = f'{username} {api_key[-4:]}'
    appname = parent_caller()
    caller = f"{appname} {username}"
    dashboard = meraki.DashboardAPI(api_key,
                                    base_url=base_url,
                                    caller=caller,
                                    single_request_timeout=15,      # Default = 60 ; Preferred = 10 - 30
                                    wait_on_rate_limit=True,
                                    maximum_retries=5,              # Default = 2 ; Preferred = 3 - 5
                                    retry_4xx_error=True,
                                    suppress_logging=True,
                                    print_console=False,
                                    log_path='output/errorlogs',
                                    inherit_logging_config=False,
                                    )
    return dashboard


def get_ms_switchports(dashboard, serial):
    """
    List the switch ports for a switch

    :param dashboard:
    :param serial: Serial Number
    :return:
    """
    result = None
    try:
        result = dashboard.switch.getDeviceSwitchPorts(serial)
    except meraki.APIError as e:
        print(f'Err on getDeviceSwitchPorts {e}')
    return result


def get_nm_ports(dashboard, serial, port_count):
    """
    Get list of NMs in Cat9K switch
    :param dashboard:
    :param serial:
    :param port_count:
    :return:
    """
    nm_list = list()
    new_ports = get_ms_switchports(dashboard, serial)
    for port in new_ports[port_count+2:]:
        nm_list.append(port.get('portId'))
    return nm_list


def set_c9k_switch_uplinks(dashboard, model, serial, sw_name, trunk_vlans, native_vlan, port_count):
    """
    Configure Uplink ports 52-59s as uplinks

    :param dashboard:
    :param model: MS130 or C9300
    :param sw_name: Switch Name
    :param serial: Serial Number
    :param trunk_vlans: set of vlans to allow
    :param native_vlan: Native VLAN
    :param port_count: max Access port count (helps determine where uplinks are
    :return:
    """
    result = None
    if "MS130" in model:
        nm_ports = ['25', '26', '27', '28']
    else:
        nm_ports = get_nm_ports(dashboard, serial, port_count)
    for port_id in nm_ports:
        try:
            print(f"\rset_c9k_uplinks [{sw_name}]/{str(port_id)} / "
                  f"trunk native:{native_vlan} allowed-vlans:{trunk_vlans}", end='', flush=True)
            result = dashboard.switch.updateDeviceSwitchPort(
                serial,
                port_id,
                name=port_id,
                tags=['Uplink'],
                enabled=True,
                type='trunk',
                vlan=native_vlan,
                allowedVlans=trunk_vlans,
                rstpEnabled=True,
                stpGuard='disabled',
                udld='Alert only',
            )
            if result:
                print(f"\r[{sw_name}]/{result.get('portId')} / "
                      f"trunk native:{native_vlan} allowed-vlans:{result.get('allowedVlans')} / "
                      f"{'enabled' if result.get('enabled') else 'disabled'}", end='', flush=True)
        except meraki.APIError as e:
            print(f'[{sw_name}] Err on set_c9k_switch_uplinks for {serial}{port_id}\n{e}', flush=True)
    return result


def set_switch_management(dashboard, serial, mgmt_vlan):
    """
    Set Switch to specified VLAN
    :param dashboard:
    :param serial: Serial Number for new Switch
    :param mgmt_vlan: Management VLAN for new Switch
    :return:
    """
    try:
        result = dashboard.devices.updateDeviceManagementInterface(
            serial,
            wan1={'wanEnabled': 'not configured', 'usingStaticIp': False, 'vlan': int(mgmt_vlan)}
        )
        if result:
            print(f'\rset_switch_management [{serial} Mgmt VLAN: {mgmt_vlan}', end='', flush=True)
    except meraki.APIError as e:
        print(f"Error trying to reach device\n{e}")
        result = None
    return result


def init_switch_ports(dashboard, serial, mgmt_vlan):
    """
    Initialize Switch with all ports using trunks native mgmt and all-vlans-allowed
    instead of out-of-the-box native 1 allowed 1-1000
    :param dashboard:
    :param serial: Serial Number for new Switch
    :param mgmt_vlan: management vlan
    :return:
    """
    try:
        print(f'\rLooking up {serial}...', end='', flush=True)
        new_device = dashboard.devices.getDevice(serial)
    except meraki.APIError as e:
        print(f'[{serial}] init_ms_switchport could not be found \n{e}', flush=True)
        return None
    sw_name = new_device.get('name')
    new_model = new_device.get('model')
    port_count = int(new_model.split("-")[-1][0:2])
    print(f'\rInitializing {sw_name}', end='', flush=True)
    set_c9k_switch_uplinks(dashboard, new_model, serial, sw_name, mgmt_vlan, mgmt_vlan, port_count)
    set_switch_management(dashboard, serial, mgmt_vlan)
    for port_id in range(1, port_count+1):
        try:
            print(f"\rinit_switch_ports [{sw_name}]/{str(port_id)}", end='', flush=True)
            result = dashboard.switch.updateDeviceSwitchPort(
                serial,
                str(port_id),
                name='',
                enabled=True,
                type='trunk',
                vlan=mgmt_vlan,
                allowedVlans=mgmt_vlan,
                accessPolicyType='Open',
                rstpEnabled=True,
                stpGuard='loop guard',
                linkNegotiation='Auto negotiate',
                udld='Alert only',
            )
            if result:
                print(f"\rinit_switch_ports [{sw_name}]/{result.get('portId')} / "
                      f"trunk native:{result.get('vlan')} allowed-vlans:{result.get('allowedVlans')} / "
                      f"{'enabled' if result.get('enabled') else 'disabled'}", end='', flush=True)
        except meraki.APIError as e:
            print(f'[{sw_name}] init_ms_switchport could not update {serial}{port_id}\n{e}', flush=True)
    print(f'\r{" "*120}\r{sw_name} Initialized', flush=True)


if __name__ == '__main__':
    parser = argparse.ArgumentParser(
        prog='init_cs9300_ports.py',
        description='Initialize CS9300 switch ports',
        epilog='NOTE: Must be executed for all switches in a stack'
    )
    parser.add_argument('sw_serial', help='Switch Cloud Serial Number. e.g. Q5KJ-CSZF-287Y')
    parser.add_argument('sw_mgmt_vlan', help='Switch Management VLAN. e.g. 84')
    args = parser.parse_args()

    # Check for required parameters
    sw_serial = args.sw_serial
    if not sw_serial:
        print("Serial Number mandatory.  Use --help for more info")
        exit(1)
    sw_mgmt_vlan = args.sw_mgmt_vlan
    if not sw_mgmt_vlan:
        print("Management VLAN mandatory.  Use --help for more info")
        exit(1)
    meraki_api_key = os.getenv('MERAKI_DASHBOARD_API_KEY')  # Assume Global baseurl
    if not meraki_api_key:
        print("Make sure 'MERAKI_DASHBOARD_API_KEY' is set in your environment")
        exit(1)

    dash = new_dash(meraki_api_key)
    init_switch_ports(dash, sw_serial, sw_mgmt_vlan)