Office 365 Data Group builder using F5 Python SDK
Problem this snippet solves: Python script which pulls O365 URL's/IP's from microsoft published XML formatted list, parses into python dictionaries formatted for F5 data group, and uses F5 Python SDK to create or update data groups on BIG-IP for each MS product. The script will check if the data group already exists, and if so update based on changes. Once the data groups are installed they can used/referenced by any F5 feature/configuration that can reference a data group. How to use this snippet: Python version: 3.6 Python Modules Required: * f5-sdk * requests * f5-icontrol-rest * xmltodict Update script variables for your environment: BIGIP_ADDRESS = ' ' BIGIP_USER = ' ' BIGIP_PASS = ' ' to execute: python3 o365_dg_builder.py Code : import requests import xmltodict from collections import OrderedDict from f5.bigip import ManagementRoot from icontrol.exceptions import iControlUnexpectedHTTPError import argparse MS_URL = 'https://support.content.office.net/en-us/static/O365IPAddresses.xml' BIGIP_ADDRESS = '192.168.15.165' BIGIP_USER = 'admin' BIGIP_PASS = 'admin' def load_bigip(f5_host, f5_user, f5_pass): bigip = ManagementRoot(f5_host, f5_user, f5_pass) return bigip def get_o365_data(): o365data = requests.get(MS_URL) records = [] pdata = xmltodict.parse(o365data.content) for product in pdata['products']['product']: prod_name = product['@name'] if isinstance(product['addresslist'],list): for addrlist in product['addresslist']: record = {} if 'address' in addrlist.keys(): record['name'] = "{}_{}_{}".format('MS',prod_name, addrlist['@type']) if '@type' in addrlist.keys(): if addrlist['@type'] == 'URL': record['type'] = 'string' elif (addrlist['@type'] == 'IPv4' or addrlist['@type'] == 'IPv6'): record['type'] = 'ip' # MS XML has duplicate entries, unify the list before adding to record new_addr_l = list(set(addrlist['address'])) record['records'] = new_addr_l records.append(record) if isinstance(product['addresslist'],OrderedDict): if len(product['addresslist']['address']) > 0: record = {} record['name'] = "{}_{}_{}".format('MS', prod_name, product['addresslist']['@type']) if product['addresslist']['@type'] == 'URL': record['type'] = 'string' elif product['addresslist']['@type'] == 'IPv4' or product['addresslist']['@type'] == 'IPv6': record['type'] = 'ip' record['records'] = list(set(product['addresslist']['address'])) records.append(record) return records def compare_dg_entries(record, old_dg_l): # let's get the values to compare, and see what updates we need to make #compare lists addrs_to_rem = [x for x in old_dg_l if x not in record] addrs_to_add = [x for x in record if x not in old_dg_l] return addrs_to_rem, addrs_to_add def create_dg(bigip, record): if not bigip.tm.ltm.data_group.internals.internal.exists(name=record['name']): bigip.tm.ltm.data_group.internals.internal.create(**record) create_status = 'SUCCESS' else: old_dg_entries = bigip.tm.ltm.data_group.internals.internal.load(name=record['name']) # pull into a list old_dg_l = [] for entry in old_dg_entries.records: old_dg_l.append(entry['name']) addrs_to_rem, addrs_to_add = compare_dg_entries(record['records'], old_dg_l) if len(addrs_to_add)==0 and len(addrs_to_rem)==0: create_status = 'SKIPPED' else: if len(addrs_to_rem) > 0: for addr in addrs_to_rem: old_dg_entries.records.remove(next(x for x in old_dg_entries.records if x['name'] == addr)) if len(addrs_to_add) > 0: for addr in addrs_to_add: old_dg_entries.records.append({'name':addr}) old_dg_entries.update() # if you get time create log of updated entries create_status = 'UPDATED' return create_status # ----- try: bigip = load_bigip(BIGIP_ADDRESS, BIGIP_USER, BIGIP_PASS) except iControlUnexpectedHTTPError as e: print(e) exit(1) if bigip: records = get_o365_data() for record in records: try: status = create_dg(bigip, record) print("BIGIP: {}, Data Group: {}, Create Status {}".format(BIGIP_ADDRESS, record['name'], status)) except iControlUnexpectedHTTPError as e: status = 'FAILED' print("BIGIP: {}, Data Group: {}, Create Status {}, Error {}".format(BIGIP_ADDRESS, record['name'], status, e)) Tested this on version: 13.0