File Upload via iControl REST
Code : def _upload(host, creds, fp): chunk_size = 512 * 1024 headers = { 'Content-Type': 'application/octet-stream' } fileobj = open(fp, 'rb') filename = os.path.basename(fp) if os.path.splitext(filename)[-1] == '.iso': uri = 'https://%s/mgmt/cm/autodeploy/software-image-uploads/%s' % (host, filename) else: uri = 'https://%s/mgmt/shared/file-transfer/uploads/%s' % (host, filename) requests.packages.urllib3.disable_warnings() size = os.path.getsize(fp) start = 0 while True: file_slice = fileobj.read(chunk_size) if not file_slice: break current_bytes = len(file_slice) if current_bytes < chunk_size: end = size else: end = start + current_bytes content_range = "%s-%s/%s" % (start, end - 1, size) headers['Content-Range'] = content_range requests.post(uri, auth=creds, data=file_slice, headers=headers, verify=False) start += current_bytes if __name__ == "__main__": import os, requests, argparse, getpass parser = argparse.ArgumentParser(description='Upload File to BIG-IP') parser.add_argument("host", help='BIG-IP IP or Hostname', ) parser.add_argument("username", help='BIG-IP Username') parser.add_argument("filepath", help='Source Filename with Absolute Path') args = vars(parser.parse_args()) hostname = args['host'] username = args['username'] filepath = args['filepath'] print "%s, enter your password: " % args['username'], password = getpass.getpass() _upload(hostname, (username, password), filepath) Tested this on version: 12.0
Suds Python Example
Problem this snippet solves: This script will pull a list of virtual servers, then fetch any default pools associated with it. Edit your particular settings as needed (url, username, password, etc.) How to use this snippet: Script Suds-sample.py Code : #!/bin/env python # Small example of how to use the Python suds library with iControl. # See comments for a description of each section. # **Note: this requires suds version 0.3.6 # Use at your own discretion; for example purposes ONLY! import sys import logging from urllib import pathname2url from suds.client import Client try: from suds.xsd.doctor import Import, ImportDoctor except: print "Requires suds 0.3.6 or higher!" sys.exit() # Setup a log to see details on what is going on. #logging.basicConfig(level=logging.INFO) #logging.getLogger('suds.client').setLevel(logging.DEBUG) # Create the URL. Note that you can point to the BigIP or the filesystem. # On windows, something like this works: # url = 'file:' + pathname2url('c:\\tmp') + '/LocalLB.VirtualServer.wsdl' # Here, I'm grabbing it from a lab box and snagging the LocalLB.VirtualServer # wsdl. url = 'https://192.168.1.245/iControl/iControlPortal.cgi?WSDL=LocalLB.VirtualServer' # Fix the missing import on our WSDLS. imp = Import('http://schemas.xmlsoap.org/soap/encoding/') doctor = ImportDoctor(imp) # Create the 'client' object. Call against this object for your methods. c = Client(url, doctor = doctor, username = 'admin', password = 'admin') # Suds sets up a file cache that stores some common content, like your wsdl # and any other basic GET style info. Uncomment to disable the cache. #c.set_options(cache=None) # Setup your endpoint, username, password, etc. against the object. Very # pythonic! Do this by setting options attributes. c.options.username = 'admin' c.options.password = 'admin' c.options.location = 'https://192.168.1.245/iControl/iControlPortal.cgi' # set the separator so we can create types. Python doesn't like using dot # notations inside of the WSDLS, which are valid according to the spec. This accomodates dot # notation. Special thanks to Jeff Ortel for adding this feature. c.factory.separator('_') # Call a basic method to get a list of pools on the box. virtuals = c.service.get_list() # Print the returned list. Your list is contained in the 'item' attribute of # the returned object. #for x in virtuals.item: # print x # Now, we'll create a type and pass it in as a keyword argument to a call. It # gets way more complex than this but we'll start slow...this part is a little painful, very java like ;). # The point is that suds isvery explicit with its typing. Overall, this is a very good thing but it # makes for some challenges with more type-heavy calls (like Virtual Server # create, for example). # Create a type of 'StringSequence'. We'll set attributes here and pass this # object into the keyword arguments. To create types, use the 'factory' object. vs_name_list = c.factory.create('Common.StringSequence') vs_name_list.item = virtuals.item # Now let's call a method that allows us to pass in the type. default_pools = c.service.get_default_pool_name(virtual_servers = vs_name_list) # Now we'll print them after we zip() them together. combined = zip(virtuals.item, default_pools.item) for x in combined: print "%s => %s" % (x[0], x[1])TACACS+ External Monitor (Python)
Problem this snippet solves: This script is an external monitor for TACACS+ that simulates a TACACS+ client authenticating a test user, and marks the status of a pool member as up if the authentication is successful. If the connection is down/times out, or the authentication fails due to invalid account settings, the script marks the pool member status as down. This is heavily inspired by the Radius External Monitor (Python) by AlanTen. How to use this snippet: Prerequisite This script uses the TACACS+ Python client by Ansible (tested on version 2.6). Create the directory /config/eav/tacacs_plus on BIG-IP Copy all contents from tacacs_plus package into /config/eav/tacacs_plus. You may also need to download six.py from https://raw.githubusercontent.com/benjaminp/six/master/six.py and place it in /config/eav/tacacs_plus. You will need to have a test account provisioned on the TACACS+ server for the script to perform authentication. Installation On BIG-IP, import the code snippet below as an External Monitor Program File. Monitor Configuration Set up an External monitor with the imported file, and configure it with the following environment variables: KEY: TACACS+serversecret USER: Usernamefortestaccount PASSWORD: Passwordfortestaccount MOD_PATH: PathtolocationofPythonpackagetacacs_plus,default:/config/eav TIMEOUT: DurationtowaitforconnectivitytoTACACSservertobeestablished,default:3 Troubleshooting SSH to BIG-IP and run the script locally $ cd /config/filestore/files_d/Common_d/external_monitor_d/ # Get name of uploaded file, e.g.: $ ls -la ... -rwxr-xr-x. 1 tomcat tomcat 1883 2021-09-17 04:05 :Common:tacacs-monitor_39568_7 # Run the script with the corresponding variables $ KEY=<my_tacacs_key> USER=<testuser> PASSWORD=<supersecure> python <external program file, e.g.:Common:tacacs-monitor_39568_7> <TACACS+ server IP> <TACACS+ server port> Code : #!/usr/bin/env python # # Filename : tacacs_plus_mon.py # Author : Leon Seng # Version : 1.2 # Date : 2021/09/21 # Python ver: 2.6+ # F5 version: 12.1+ # # ========== Installation # Import this script via GUI: # System > File Management > External Monitor Program File List > Import... # Name it however you want. # Get, modify and copy the following modules: # ========== Required modules # -- six -- # https://pypi.org/project/six/ # Copy six.py into /config/eav # # -- tacacs_plus -- # https://pypi.org/project/tacacs_plus/ | https://github.com/ansible/tacacs_plus # Copy tacacs_plus directory into /config/eav # ========== Environment Variables # NODE_IP - Supplied by F5 monitor as first argument # NODE_PORT - Supplied by F5 monitor as second argument # KEY - TACACS+ server secret # USER - Username for test account # PASSWORD - Password for test account # MOD_PATH - Path to location of Python package tacacs_plus, default: /config/eav # TIMEOUT - Duration to wait for connectivity to TACACS server to be established, default: 3 import os import socket import sys if os.environ.get('MOD_PATH'): sys.path.append(os.environ.get('MOD_PATH')) else: sys.path.append('/config/eav') # https://github.com/ansible/tacacs_plus from tacacs_plus.client import TACACSClient node_ip = sys.argv[1] node_port = int(sys.argv[2]) key = os.environ.get("KEY") user = os.environ.get("USER") password = os.environ.get("PASSWORD") timeout = int(os.environ.get("TIMEOUT", 3)) # Determine if node IP is IPv4 or IPv6 family = None try: socket.inet_pton(socket.AF_INET, node_ip) family = socket.AF_INET except socket.error: # not a valid address try: socket.inet_pton(socket.AF_INET6, node_ip) family = socket.AF_INET6 except socket.error: sys.exit(1) # Authenticate against TACACS server client = TACACSClient(node_ip, node_port, key, timeout=timeout, family=family) try: auth = client.authenticate(user, password) if auth.valid: print "up" except socket.error: # EAV script marks node as DOWN when no output is present pass Tested this on version: 12.1
Radius External Monitor (Python)
Problem this snippet solves: Note: This script works and is usable however currently the only issue is that for some unknown reason it fails authentication on valid credentials, but since it still gets a valid response from the Radius server it will still mark the pool member as up. This only occurs via a configured monitor using either variable or arguments, running the script via command line with the same arguments works. This python script is an external monitor for radius that checks for any radius response in order to mark a pool member as up. If the connection times out the script does not output any response and as such will result in the member being marked as down. The script strips of routing domain tags and the IPv6 padding of the Node IPs. How to use this snippet: Installation: Import as an External Monitor Program File and name it however you want. Get and copy python modules Six (six py) and Py-Radius (radius py) then copy them to a directory with suitable permissions such as the /config/eav directory. If you're running python 2.6, which you probably will, then the radius module will need to be modified as per Code Mods section in the code below. Monitor Configuration: Set up a monitor with at the very least a valid SECRET, if you're looking to simply test whether the Radius server is alive and responding then the user account does not need to be valid. Valid Environment Variables are as follows: DEBUG = 0, 1, 2 (Default: 0) MOD_PATH = Directory containing the Python modules (Default: /config/eav) SECRET = Radius server secret (Default: NoSecret) USER = User account (Default: radius_monitor) PASSWD = Account Password (Default: NoPassword) Notes: The configured environment variables can overridden using Arguments with TESTME as the first argument followed by the variables you want to override in the same format as above, eg. KEY=Value. This is usedful for troubleshooting and can be used to override the F5 supplied NODE_IP and NODE_PORT variables. Log file location: /var/log/monitors Debug level 2 outputs environment variables, command line arguments and radius module into the log file. Code : #!/usr/bin/env python # -*- coding: utf-8 -*- # # Filename : radius_mon.py # Author : ATennent # Version : 1.1 # Date : 2018/09/14 # Python ver: 2.6+ # F5 version: 12.0+ # ========== Installation # Import this script via GUI: # System > File Management > External Monitor Program File List > Import... # Name it however you want. # Get, modify and copy the following modules: # ========== Required modules # -- six -- # https://pypi.org/project/six/ # Copy six.py into /config/eav # # -- py-radius -- # https://pypi.org/project/py-radius/ # Copy radius.py into /config/eav # If running python2.6 modify radius.py as per "Code mods -- python.py --" section below # ========== Notes # NB: Any and all outputs from EAV scripts equate to a POSITIVE result by the F5 monitor # so to make the result of the script to be a NEGATIVE result, ie. a DOWN state, we need # to ensure that the script does not output anything when the attempt results in a time out. # ========== Environment Variables # NODE_IP - Supplied by F5 monitor # NODE_PORT - Supplied by F5 monitor # MOD_PATH - Path to location of Python modules six.py and radius.py, default: /config/eav # SECRET - Radius server secret # USER - Username for test account # PASSWD - Password for user account # DEBUG - Enable/Disable Debugging # ========== Additional Environment Variables (supplied at execution) # MON_TMPL_NAME - Monitor Name, eg. '/Common/radius_monitor' # RUN_I - Run file, eg. '/Common/radius_mon.py' # SECURITY_FIPS140_COMPLIANCE - 'true'/'false' # PATH - Colon delimited and not editable. # ARGS_I - Copy of command line Arguments # NODE_NAME - Node Name, eg, '/Common/SNGPCTXWEB01' # MON_INST_LOG_NAME - Pool member's monitor log, eg. '/var/log/monitors/Common_radius_monitor-Common_MyNode-1812.log' # SECURITY_COMMONCRITERIA - 'true'/'false' # ========== Code mods -- python.py -- # Dictionary comprehension fixes: # https://stackoverflow.com/questions/1747817/create-a-dictionary-with-list-comprehension-in-python#1747827 # LINE:193; replace with #ATTR_NAMES = dict((v.lower(), k) for (k, v) in ATTRS.items()) # # LINE:100; replace with #CODE_NAMES = dict((v.lower(), k) for (k, v) in CODES.items()) # # Logger.NullHandler fix: # https://stackoverflow.com/questions/33175763/how-to-use-logging-nullhandler-in-python-2-6#34939479 # LINE:64-65; replace with #LOGGER = logging.getLogger(__name__) #try: # Python 2.7+ # LOGGER.addHandler(logging.NullHandler()) #except AttributeError: # class NullHandler(logging.Handler): # def emit(self, record): # pass # LOGGER.addHandler(NullHandler()) # ========== Imports/Modules from sys import path from sys import argv from sys import stdout from os import environ import logging.handlers import re if environ.get('MOD_PATH'): path.append(environ.get('MOD_PATH')) else: path.append('/config/eav') import radius # ========== Dictionary Defaults opts_dict = {'NODE_IP': '', 'NODE_PORT': '', 'SECRET': 'NoSecret', 'USER': 'radius_monitor', 'PASSWD': 'NoPassword', 'DEBUG': '0', } # ========== TEST w/ command line try: if argv[3] == 'TESTME': environ['NODE_IP'] = argv[1] environ['NODE_PORT'] = argv[2] for cmd_opt in argv[3:]: if 'DEBUG' in cmd_opt: environ['DEBUG'] = cmd_opt.split('=')[1] elif 'SECRET' in cmd_opt: environ['SECRET'] = cmd_opt.split('=')[1] elif 'USER' in cmd_opt: environ['USER'] = cmd_opt.split('=')[1] elif 'PASSWD' in cmd_opt: environ['PASSWD'] = cmd_opt.split('=')[1] else: continue except: pass # ========== Logging Config try: if int(environ.get('DEBUG')) == 0: log_file = '/dev/null' elif int(environ.get('DEBUG')) <=2: log_file = environ.get('MON_INST_LOG_NAME') else: log_file = '/dev/null' except: # DEBUG not supplied as ENV variable environ['DEBUG'] = opts_dict['DEBUG'] log_file = '/dev/null' if int(environ.get('DEBUG')) == 2: logging.basicConfig( filename=log_file, level=logging.DEBUG, format='%(asctime)s %(name)s:%(levelname)s %(message)s') else: logging.basicConfig( filename=log_file, level=logging.INFO, format='%(asctime)s %(name)s:%(levelname)s %(message)s') log = logging.getLogger('radius_mon') syslog_handler = logging.handlers.SysLogHandler(address='/dev/log', facility=0) log.addHandler(syslog_handler) if int(environ.get('DEBUG')) == 0: log.setLevel(logging.INFO) opts_dict['DEBUG'] = environ.get('DEBUG') elif int(environ.get('DEBUG')) == 1: log.setLevel(logging.DEBUG) opts_dict['DEBUG'] = environ.get('DEBUG') elif int(environ.get('DEBUG')) == 2: log.setLevel(logging.DEBUG) opts_dict['DEBUG'] = environ.get('DEBUG') log.debug('VERBOSE Debug Enabled, CMD line and ENV variables will be dumped to file.') else: log.error('Bad DEBUG value!') # ========== CMD Line and ENV Dump if int(opts_dict['DEBUG']) == 2: log.debug('=========== CMD ARGS') log.debug(argv[0:]) log.debug('=========== ENV VARS') log.debug(environ) # ========== Update Dictionary if environ.get('NODE_IP'): opts_dict['NODE_IP'] = environ.get('NODE_IP') log.debug('Set NODE_IP, %s', opts_dict['NODE_IP']) else: log.error('NODE_IP not supplied as ENV variable') log.error('Exiting..') exit(0) if environ.get('NODE_PORT'): opts_dict['NODE_PORT'] = environ.get('NODE_PORT') log.debug('Set NODE_PORT, %s', opts_dict['NODE_PORT']) else: log.error('NODE_PORT not supplied as ENV variable') log.error('Exiting..') exit(0) if environ.get('SECRET'): opts_dict['SECRET'] = environ.get('SECRET') log.debug('Set SECRET, %s', opts_dict['SECRET']) else: log.debug('SECRET not supplied as ENV variable') if environ.get('USER'): opts_dict['USER'] = environ.get('USER') log.debug('Set USER, %s', opts_dict['USER']) else: log.debug('USER not supplied as ENV variable') if environ.get('PASSWD'): opts_dict['PASSWD'] = environ.get('PASSWD') log.debug('Set PASSWD, %s', opts_dict['PASSWD']) else: log.debug('PASSWD not supplied as ENV variable') # ========== Clean up NODE_IP for IPv4 if '::ffff:' in opts_dict['NODE_IP']: opts_dict['NODE_IP'] = re.sub(r'::ffff:','', opts_dict['NODE_IP']) log.debug('Stripped IPv6 notation from IPv4 address') # ========== Strip routing domain from NODE_IP if '%' in opts_dict['NODE_IP']: opts_dict['NODE_IP'] = re.sub(r'%\d{1,4}$', '', opts_dict['NODE_IP']) log.debug('Stripped routing domain from IP address') # ========== Dictionary debug if int(opts_dict['DEBUG']) == 2: log.debug('----- KEY, VALUE pairs:') for key in opts_dict.keys(): log.debug('%s, %s', key, opts_dict[key]) log.debug('-----') # ========== Main r = radius.Radius(opts_dict['SECRET'], host=opts_dict['NODE_IP'], port=int(opts_dict['NODE_PORT'])) try: if r.authenticate(opts_dict['USER'], opts_dict['PASSWD']): log.debug('Service UP, Authentication attempt Successful') stdout.write('UP') else: log.debug('Service UP, Authentication attempt Failure') stdout.write('UP') except radius.ChallengeResponse: log.debug('Service UP, Authentication attempt Challenge Response') stdout.write('UP') except (radius.NoResponse, radius.SocketError): # This includes bad SECRET keys, the radius RFC states that Invalid responses from the radius # server are to be silently dropped which, from a client perspective, will result ina timeout. # This cannot be changed without modifying radius.py to instead raise an Exception. log.debug('No Response from %s:%s', opts_dict['NODE_IP'], int(opts_dict['NODE_PORT'])) exit(0) except Exception as msg: log.error('Exception due to %s, marking as DOWN', msg) exit(0) Tested this on version: 12.0
Python BigREST ucs save and download
Problem this snippet solves: This uses BIGREST SDK API to save/download the ucs file. Adding it here since I couldn't find a similar one, How to use this snippet: Based on Python3 and BigREST (https://bigrest.readthedocs.io) Code : #Import needed libraries from bigrest.bigip import BIGIP import getpass #Replace the host name as needed host="xx.xx.xx.xx" user=input('Username') pw= getpass.getpass(prompt='Password:') #Declare the ucs filename, if needed ucsfile="test.ucs" #Connect to device device = BIGIP(host, user, pw) data = {} data["command"] = "save" data["name"] = ucsfile #You may get timeout exception below even if the file has been created task = device.task_start("/mgmt/tm/sys/ucs", data) device.task_wait(task) if device.task_completed(task): device.task_result(task) print("Backup has been completed.") else: raise Exception() #The below will download the file to the same folder as the script device.download("/mgmt/shared/file-transfer/ucs-downloads/", ucsfile) Tested this on version: 13.1Python Virtual Server and Pool Creation
Problem this snippet solves: A simple script to add and delete both Virtual IPs and Pools in: Python Code : import requests, json, time # define program-wide variables BIGIP_ADDRESS = 'test-ltm-03.element.local' BIGIP_USER = 'admin' BIGIP_PASS = 'admin' SLEEP_TIME = 20 VS_NAME = 'test-http-virtual_python' VS_ADDRESS = '1.1.1.3' VS_PORT = '80' POOL_NAME = 'test-http-pool_python' POOL_LB_METHOD = 'least-connections-member' POOL_MEMBERS = [ '10.0.0.1:80', '10.0.0.2:80', '10.0.0.3:80' ] # create/delete methods def create_pool(bigip, name, members, lb_method): payload = {} # convert member format members = [ { 'kind' : 'ltm:pool:members', 'name' : member } for member in POOL_MEMBERS ] # define test pool payload['kind'] = 'tm:ltm:pool:poolstate' payload['name'] = name payload['description'] = 'A Python REST client test pool' payload['loadBalancingMode'] = lb_method payload['monitor'] = 'http' payload['members'] = members bigip.post('%s/ltm/pool' % BIGIP_URL_BASE, data=json.dumps(payload)) def create_http_virtual(bigip, name, address, port, pool): payload = {} # define test virtual payload['kind'] = 'tm:ltm:virtual:virtualstate' payload['name'] = name payload['description'] = 'A Python REST client test virtual server' payload['destination'] = '%s:%s' % (address, port) payload['mask'] = '255.255.255.255' payload['ipProtocol'] = 'tcp' payload['sourceAddressTranslation'] = { 'type' : 'automap' } payload['profiles'] = [ { 'kind' : 'ltm:virtual:profile', 'name' : 'http' }, { 'kind' : 'ltm:virtual:profile', 'name' : 'tcp' } ] payload['pool'] = pool bigip.post('%s/ltm/virtual' % BIGIP_URL_BASE, data=json.dumps(payload)) def delete_pool(bigip, name): bigip.delete('%s/ltm/pool/%s' % (BIGIP_URL_BASE, name)) def delete_virtual(bigip, name): bigip.delete('%s/ltm/virtual/%s' % (BIGIP_URL_BASE, name)) # REST resource for BIG-IP that all other requests will use bigip = requests.session() bigip.auth = (BIGIP_USER, BIGIP_PASS) bigip.verify = False bigip.headers.update({'Content-Type' : 'application/json'}) print "created REST resource for BIG-IP at %s..." % BIGIP_ADDRESS # Requests requires a full URL to be sent as arg for every request, define base URL globally here BIGIP_URL_BASE = 'https://%s/mgmt/tm' % BIGIP_ADDRESS # create pool create_pool(bigip, POOL_NAME, POOL_MEMBERS, POOL_LB_METHOD) print "created pool \"%s\" with members %s..." % (POOL_NAME, ", ".join(POOL_MEMBERS)) # create virtual create_http_virtual(bigip, VS_NAME, VS_ADDRESS, VS_PORT, POOL_NAME) print "created virtual server \"%s\" with destination %s:%s..." % (VS_NAME, VS_ADDRESS, VS_PORT) # sleep for a little while print "sleeping for %s seconds, check for successful creation..." % SLEEP_TIME time.sleep(SLEEP_TIME) # delete virtual print "deleted virtual server \"%s\"..." % VS_NAME delete_virtual(bigip, VS_NAME) # delete pool print "deleted pool \"%s\"..." % POOL_NAME delete_pool(bigip, POOL_NAME)Getting Started with iControl Code Samples
Problem this snippet solves: This repository on github is for the full scripts highlighted in the Getting Started with iControl article series. Developed against BIG-IP TMOS version 12.0, but should work in most versions 11.6+. How to use this snippet: Choose your language and iControl portal of choice to start investigating iControl in your environment. Target audience: beginners. Code : https://github.com/f5devcentral/iControl-GettingStarted Tested this on version: 12.0BIG-IP File Transfers
Problem this snippet solves: This python script allows you to transfer files to/from BIG-IP/BIG-IQ. Note that all downloads come from /shared/images, so if you need to download files that are not .iso images, you'll need to copy them to /shared/images and then clean up afterward. How to use this snippet: # Download from BIG-IP (Files must be in /shared/images/) python f_xfer.py download bigip.test.local admin /path/you/want/on/your/system/example.txt # Upload to BIG-IP (Files placed in /var/config/rest/downloads unless extension is .iso, and then it will be placed in /shared/images/) python f_xfer.py upload bigip.test.local admin /path/file/exists/on/your/system/example.txt Code : import os, requests, argparse, getpass def _download(host, creds, fp): chunk_size = 512 * 1024 headers = { 'Content-Type': 'application/octet-stream' } filename = os.path.basename(fp) uri = 'https://%s/mgmt/cm/autodeploy/software-image-downloads/%s' % (host, filename) with open(fp, 'wb') as f: start = 0 end = chunk_size - 1 size = 0 current_bytes = 0 while True: content_range = "%s-%s/%s" % (start, end, size) headers['Content-Range'] = content_range #print headers resp = requests.get(uri, auth=creds, headers=headers, verify=False, stream=True) if resp.status_code == 200: # If the size is zero, then this is the first time through the # loop and we don't want to write data because we haven't yet # figured out the total size of the file. if size > 0: current_bytes += chunk_size for chunk in resp.iter_content(chunk_size): f.write(chunk) # Once we've downloaded the entire file, we can break out of # the loop if end == size: break crange = resp.headers['Content-Range'] # Determine the total number of bytes to read if size == 0: size = int(crange.split('/')[-1]) - 1 # If the file is smaller than the chunk size, BIG-IP will # return an HTTP 400. So adjust the chunk_size down to the # total file size... if chunk_size > size: end = size # ...and pass on the rest of the code continue start += chunk_size if (current_bytes + chunk_size) > size: end = size else: end = start + chunk_size - 1 def _upload(host, creds, fp): chunk_size = 512 * 1024 headers = { 'Content-Type': 'application/octet-stream' } fileobj = open(fp, 'rb') filename = os.path.basename(fp) if os.path.splitext(filename)[-1] == '.iso': uri = 'https://%s/mgmt/cm/autodeploy/software-image-uploads/%s' % (host, filename) else: uri = 'https://%s/mgmt/shared/file-transfer/uploads/%s' % (host, filename) size = os.path.getsize(fp) start = 0 while True: file_slice = fileobj.read(chunk_size) if not file_slice: break current_bytes = len(file_slice) if current_bytes < chunk_size: end = size else: end = start + current_bytes content_range = "%s-%s/%s" % (start, end - 1, size) headers['Content-Range'] = content_range requests.post(uri, auth=creds, data=file_slice, headers=headers, verify=False) start += current_bytes if __name__ == "__main__": import os, requests, argparse, getpass requests.packages.urllib3.disable_warnings() parser = argparse.ArgumentParser(description='Transfer files to/from BIG-IP') parser.add_argument("transfer_status", help='upload | download') parser.add_argument("host", help='BIG-IP IP or Hostname', ) parser.add_argument("username", help='BIG-IP Username') parser.add_argument("filepath", help='Appropriate Src/Dest Filename with Absolute Path') args = vars(parser.parse_args()) tstat = args['transfer_status'] hostname = args['host'] username = args['username'] filepath = args['filepath'] print('{}, enter your password: '.format(args['username'])) password = getpass.getpass() if tstat == 'upload': _upload(hostname, (username, password), filepath) elif tstat == 'download': _download(hostname, (username, password), filepath) Tested this on version: 13.0
Get All Pool Member Status Using LTM, Python, and bigsuds
Problem this snippet solves: I spent a bit of time trying to get this to work. Delivering the data in the correct format to the get_member_session_status method took me a while, so I thought I would post my solution. Code : #!/usr/bin/env python import sys import bigsuds def get_pools(obj): try: return obj.LocalLB.Pool.get_list() except Exception, e: print e def get_members(obj, pool): try: return pool, obj.LocalLB.Pool.get_member_v2(pool) except Exception, e: print e def get_status(obj, pool): try: return obj.LocalLB.Pool.get_member_session_status(pool) except Exception, e: print e try: b = bigsuds.BIGIP( hostname = “ ”, username = "admin", password = “admin”, ) except Exception, e: print e pools = get_pools(b) members = get_members(b, pools) for pool in pools: print "Pool: %s" % pool members = b.LocalLB.Pool.get_member_v2([pool]) status = b.LocalLB.Pool.get_member_session_status([pool], members) print "\tMembers:" for members, status in zip(members, status): count = 0 while count < len(members): print "\t\tHost: %s\tState: %s" % (members[count]["address"], status[count]) count += 1 Tested this on version: 11.6iRules Runtime Calculator Spreadsheet Generator
Problem this snippet solves: This script auto-generates the iRules Runtime Calculator Excel spreadsheet by connecting to BIG-IP, collecting the iRule and its stats, and then using the XlsxWriter module to populate the spreadsheet. Tested on BIG-IP 12.1, 13.1, 14.1, 15.0 with python 3.5, 3.7 from Windows 10 and MacOS Mojave Article details if interested. Screenshots: How to use this snippet: usage: runtime_calc_v2.py [-h] host username rule Tested this on version: 13.0 Download here: GitHub Gist
