NewOrChangeRulesToMXL3Firewall.py

"""
Copyright (c) 2021 Cisco and/or its affiliates.
This software is licensed to you under the terms of the Cisco Sample
Code License, Version 1.1 (the "License"). You may obtain a copy of the
License at
               https://developer.cisco.com/docs/licenses
All use of the material herein must be in accordance with the terms of
the License. All rights not expressly granted by the License are
reserved. Unless required by applicable law or agreed to separately in
writing, software distributed under the License is distributed on an "AS
IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
or implied.
"""
# This script retrieves a Meraki MX L3 Firewall rule from the file NewRuleToAdd.txt and adds it to all
# networks in the Org. The rule inserted only specifies the Destination IPs (destCidr field) and uses
# the comment specified in the first line of the input file for the "comment" field of the rule
# all other parameters for the rule are specified in templateRuleDict below for all insertions.

import meraki
import time
import sys
import config
import requests
from NewOrChangedRuleConfig import RULE_DATA, RULE_ACTION, DupRuleAct


dashboard = meraki.DashboardAPI(api_key=config.meraki_api_key)

# Return configured Syslog Servers for a network
# https://dashboard.meraki.com/api_docs#list-the-syslog-servers-for-a-network
def getsyslogservers(apikey, networkid, suppressprint=False):
    calltype = 'Syslog servers'
    geturl = '{0}/networks/{1}/syslogServers'.format(
        str(meraki.base_url), str(networkid))
    headers = {
        'x-cisco-meraki-api-key': format(str(apikey)),
        'Content-Type': 'application/json'
    }
    dashboard = requests.get(geturl, headers=headers)
    result = meraki.__returnhandler(
        dashboard.status_code, dashboard.text, calltype, suppressprint)
    return result


# sample in case you want to obtain the orgs programmatically
#myOrgs = meraki.myorgaccess(config.meraki_api_key, True)
#print(myOrgs)

networks_skip=[]
# read a list of networks to skip
try:
    with open('networks_skip.txt') as my_file:
        networks_skip = my_file.read().splitlines()
except IOError as e:
    print("Error trying to read networks_skip.txt, skipping...")
except:
    print("Unexpected error: ", sys.exc_info()[0])


#obtain all networks in the Org specified by the config variable
#myNetworks = meraki.getnetworklist(config.meraki_api_key, config.meraki_org_id, None, True)
myNetworks = dashboard.organizations.getOrganizationNetworks(config.meraki_org_id, total_pages='all')

#stop the script if the operator does not agree with the operation being previewed
print("About start processing the insertion/update or append of the following Rule: ")
print(RULE_DATA)
print("into the following networks (pending confirmation and exclusion/inclusion rules checking):")
for theNetwork in myNetworks:
    theNetworkid = theNetwork["id"]
    theNetworkname = theNetwork["name"]
    print(theNetworkid, "  ",theNetworkname, end = '')
    if theNetwork["id"] in networks_skip:
        print(" ...... skipping because it is listed in the networks_skip.txt file." )
    else:
        print("")

if not input("Procced? (y/n): ").lower().strip()[:1] == "y": sys.exit(1)


for theNetwork in myNetworks:
    theNetworkid = theNetwork["id"]
    #un-comment the 3 lines below and modify accordingly if you want to filter out networks whose name matches those conditions
    #if theNetwork["name"].startswith('z') or theNetwork["name"].endswith('switch-wifi') or theNetwork["name"].endswith('camera') or theNetwork["name"].endswith('systems manager'):
    #   print("Skipping network named: ", theNetwork["name"], " with id ", theNetwork["id"]," due to hard-coded name filtering rules.")
    #   continue

    # here we skip any networks listed in the networks_skip.txt file
    if theNetwork["id"] in networks_skip:
        print("Skipping network named: ",theNetwork["name"]," with id ",theNetwork["id"], " because it is listed in the networks_skip.txt file." )
        continue

    print("Updating rules for Network ID: "+theNetworkid+" named: ",theNetwork["name"],"...")
    continueAnswer="y"
    #Comment line below if you wish to skip confirmation for each Network
    continueAnswer=input("Continue? yes, no or skip(y/n/s): ").lower().strip()[:1]
    if continueAnswer=="n":
        print("Bye!")
        sys.exit(1)
    elif continueAnswer=="s":
        print("Skipping Network ID: "+theNetworkid+" named: ",theNetwork["name"],"...")
        continue

    #get the rules
    theMXL3FirewallRules=dashboard.appliance.getNetworkApplianceFirewallL3FirewallRules(theNetworkid)
    # print("THE RULES:"+str(theMXL3FirewallRules))

    #retrieving the syslog servers to know which template to use
    theSysLogServers = dashboard.networks.getNetworkSyslogServers(theNetworkid)
    #print("Syslog Servers: ", theSysLogServers)

    # If the network does not have a syslog server, remove the "syslogEnabled" key from the dict to avoid errors
    theRuleToAddDict=RULE_DATA
    if theSysLogServers["servers"] == []:
        theRuleToAddDict.pop("syslogEnabled")

    #starting with a clean set of rules to re-insert into the network configuration for L3 FW rules
    theMXL3FirewallCleanRules=[]
    #set this flag for each evaluation of a rule to mark when we have added it back in in case of replacing or appending
    processedRuleData=False

    #now iterate trough all rules extracted to put back in those that are not being appended or replaced and are not default
    for theRule in theMXL3FirewallRules["rules"]:
        if theRule["comment"]!="Default rule":     #ignoring any marked as "Default rule" to avoid duplicates
            #check to see if we allow duplicates. If not, then do not a rule if another one with the same comment exists
            if theRule["comment"]!=theRuleToAddDict["comment"]: #the new rule does not have the same name as this one, just add back in.
                theMXL3FirewallCleanRules.append(theRule)
            else: # there is already a rule with the same name, process according to what is configured in RULE_ACTION
                processedRuleData=True
                if RULE_ACTION==DupRuleAct.REPLACE:
                    theMXL3FirewallCleanRules.append(theRuleToAddDict)
                    print("Replaced rule: ", theRule["comment"])
                elif RULE_ACTION==DupRuleAct.APPEND:
                    appendedRule=theRuleToAddDict
                    #first process the srcCidr field
                    #logic is: add original and new field separated by comma if the both have values that are not 'Any'
                    #if orig has 'Any' and new value is not blank, take on the new value
                    #if orig has a valid non-'Any' value, but new value is 'Any', replace with 'Any'
                    #orig value should never be blank.
                    theSeparator=""
                    if (theRule["srcCidr"].casefold()=='Any'.casefold() and theRuleToAddDict["srcCidr"]!='') or theRuleToAddDict["srcCidr"].casefold()=='Any'.casefold():
                        theRule["srcCidr"] = '' # blank out the original srcCidr and leave separator blank as well so upon concat only the new value is added
                    elif theRule["srcCidr"]!='' and theRuleToAddDict["srcCidr"]!='':
                        theSeparator="," # now we know both orig and new have valid values, so we need to assing "," as separator when we concatenate
                    appendedRule["srcCidr"]=theRule["srcCidr"]+theSeparator+theRuleToAddDict["srcCidr"] #this should now do the right concatenation
                    #now process the destCidr field. Same logic as above for the srcCidr field regarding handling of blank/'Any' values to correctly concatenate
                    theSeparator=""
                    if (theRule["destCidr"].casefold()=='Any'.casefold() and theRuleToAddDict["destCidr"]!='') or theRuleToAddDict["destCidr"].casefold()=='Any'.casefold():
                        theRule["destCidr"] = ''
                    elif theRule["destCidr"]!='' and theRuleToAddDict["destCidr"]!='':
                        theSeparator=","
                    appendedRule["destCidr"] = theRule["destCidr"] + theSeparator + theRuleToAddDict["destCidr"]
                    # finally, with the correct "srcCidr" and "destCidr" key values that we assembled,
                    # we can add the full rule to the list of rules to updated Meraki with
                    theMXL3FirewallCleanRules.append(appendedRule)
                    print("Appended to rule: ", theRule["comment"])
                elif RULE_ACTION==DupRuleAct.ADD_DUPLICATE:
                    theMXL3FirewallCleanRules.append(theRule)
                    theMXL3FirewallCleanRules.append(theRuleToAddDict)
                    print("Added duplicate rule: ",theRule["comment"])

    #add the new rule if configured to do so and was not already added
    if not processedRuleData:
        theMXL3FirewallCleanRules.append(theRuleToAddDict)

    print("CLEAN RULES:"+str(theMXL3FirewallCleanRules))

    #update the rules with the new one added
    updateResponse = dashboard.appliance.updateNetworkApplianceFirewallL3FirewallRules(theNetworkid, rules=theMXL3FirewallCleanRules)

    #need to make sure we do not send more than 5 API calls per second for this org
    #so sleep 500ms since we are making 2 API calls per loop
    time.sleep(0.5)

print("Done!")