ssoapiclient.py

# SonicWall SSO API Client
# Jaime Escalera
# Special thanks to John-Michael Denton and Ian Puleston.
# Their help was instrumental in getting the security layer figured out.

# Import Requests module for HTTP requests
# This module is required to use this script, and is not part of the standard library.
import requests # For handling HTTP requests
import sys # Related to argument handling
import argparse # For parsing CLI arguments
import hashlib # For SHA256 auth support/hash generation
import base64 # To generate the Authenticator
#from urllib3 import disable_warnings
import os
import binascii

# Settings
# Set URL to hit the API
# Format: 'https://<FIREWALL IP/FQDN>:<HTTPS MGMT PORT IF NOT 443>/api/sso/user'
# Example: 'https://192.168.0.1:4443/api/sso/user'
# Example: 'https://192.168.0.1/api/sso/user'
req_url = 'https://192.168.0.1'
req_uri = '/api/sso/user'
req_url = req_url + req_uri

# Set the client security level. This should match SonicOS configuration.
# "medium" -- Medium uses SHA256 and supports sending the same hash multiple times.
# "high" -- High supports either SHA256 or SHA512 and is more secure than Medium.
security_level = "medium"

# Set either "sha256" or "sha512". Used only for High security level.
sha_level = "sha256"

# The Authenticator is used for Medium and High level security only.
# Set security_level above to "low" to disable the authenticator.
# Set security_level above to "medium" or "high" to enable the authenticator.
# Use the same security level in SonicOS as you do in the API client.
if security_level == "low":
	enable_authenticator = False
	print("enable_authenticator is", enable_authenticator, "and security_level is", security_level)
elif security_level == "medium" or security_level == "high":
	enable_authenticator = True
	print("enable_authenticator is", enable_authenticator, "and security_level is", security_level)

# Set shared secret for API communication
# Set the same secret in SonicOS.
shared_secret = b"secret" # Set the secret in between the quotation marks.

# Authenticator header setup (64-octet sequence)
# Flags are a 4-octet field of bit flags
# Bits 0-30 are reserved. Must be set to zero.

# If bit flag 31 is set to 1 (flags = "00000001"), the API client expects
# an authenticator in the API reply. If not expecting one, set it to "00000000".
# Set reply_authenticator to True if you want the reply authenticator.
# When requested, the authenticator in the authorization header
# of the reply consists of a base64 encoding of a 64 octet sequence (sha256 high sec).
# Formatted as 0-31 response-nonce, 32-63 hash
# sha512 high sec is 0-63 resp-nonce, 64-127 hash
# 128 octet sequence for sha512 high sec.
reply_authenticator = False

if reply_authenticator is True:
	auth_flags = binascii.unhexlify('00000001')
	print("reply_authenticator is", reply_authenticator, "so API client is expecting a reply authenticator.")
else:
	auth_flags = binascii.unhexlify('00000000')
	print("reply_authenticator is", reply_authenticator, "so API client is not requesting a reply authenticator")

# Sequence number: 4-octet sequence number used only with CSRF Prevention.
# Ignored if CSRF is disabled. CSRF is disabled in medium and low security.
# CSRF Prevention
# Used to prevent replays. For use with high level security only.
# Medium security allows sending reuse of authenticators so CSRF is self-defeating.
# 
csrf_enabled = False

if csrf_enabled is True:
	auth_seqnum = binascii.unhexlify('00000001')
	print("csrf_enabled is", csrf_enabled, "and sequence number is", auth_seqnum)
else:
	auth_seqnum = binascii.unhexlify('00000000')
	print("csrf_enabled is", csrf_enabled, "and sequence number is", auth_seqnum)

# Request Nonce: a 24-octet random number for medium sec. Its generated by the client for each request.
# 56-octet random number for high security.
if security_level == "medium":
	auth_reqnonce = binascii.hexlify(os.urandom(24)) # random nonce
	auth_reqnonce = binascii.unhexlify(auth_reqnonce)
	print("Using Medium security level with a random request nonce.")
elif security_level == "high":
	if sha_level == "sha256":
		auth_reqnonce = binascii.hexlify(os.urandom(24))
		auth_reqnonce = binascii.unhexlify(auth_reqnonce)
		print("Using High security level with SHA256.")
	elif sha_level == "sha512":
		auth_reqnonce = binascii.hexlify(os.urandom(56))
		auth_reqnonce = binascii.unhexlify(auth_reqnonce)
		print("Using High security level with SHA512.")
	print("Printing random number:", auth_reqnonce)
elif security_level == "low":
	auth_reqnonce = ""
	print("Using Low security level. No request nonce generated.")

# Medium security: Create SHA256 hash from the flags, seq num, req nonce, and shared secret
# High security: Same as High, but with the full request body content; or the URI for a request with no content
# and either SHA256 or SHA512.
if security_level == "medium":
	auth_shahash = hashlib.sha256(auth_flags + auth_seqnum + auth_reqnonce + shared_secret)
elif security_level == "high":
	if sha_level == "sha256":
		auth_shahash = hashlib.sha256(auth_flags + auth_seqnum + auth_reqnonce + shared_secret)
# and THE REQUEST BODY CONTENT OR URI
	elif sha_level == "sha512":
		auth_shahash = hashlib.sha512(auth_flags + auth_seqnum + auth_reqnonce + shared_secret)
# Currently high uses the same as medium. I need to add support for high security.

# Generate/calculate authenticator used for authentication to SSO API
# Authenticator is a base64-encoding of a 64-octet (medium sec)/128-octet (high sec) sequence.
# The format is flags + seq num + req nonce + sha256 hash
if security_level == "medium":
	auth_authenticator = base64.b64encode(auth_flags + auth_seqnum + auth_reqnonce + auth_shahash.digest())
	auth_authenticator = auth_authenticator.decode("utf-8")
elif security_level == "high":
	auth_authenticator = base64.b64encode(auth_flags + auth_seqnum + auth_reqnonce + auth_shahash.digest())
	auth_authenticator = auth_authenticator.decode("utf-8")
# Currently high uses the same as medium. I need to add support for high security.

# Debug while working on the authenticator and security levels
dbg_level = 0

# Toggle debug messages
dbg_on = False

if dbg_on == True:
	print("Debug is enabled.")

# SSO Host/User Info (new global variables)
# Also setting them to an empty string so theyre defined
global user_action
user_action = ' '
global user_ip
user_ip = ' '
global user_name
user_name = ' '
global user_domain
user_domain = ' '
global user_type
user_type = ' '
global logout_reason
logout_reason = ' '

# Prepare JSON data. Can be done in one line, or neatly like this...
# Global scope and variable defined for login
# Not sure its needed anymore since implementing the update_json function
# If this default data is sent to the API, something went wrong.
# I only expect the default json data below for the action NOT USED.
# Example: if sending a login, the logout json will be default.
# If sending a logout, the login json will be default.
# This is expected as only the json data to be used will be sent.
global req_json_login
req_json_login = {
#	'ip': user_ip,
#	'name': user_name,
#	'domain': user_domain,
#	'type': user_type,
#	'default_definition': "check around line 44 and 74",
	'YOU_HIT_THE_DEFAULT_DEF': "THIS IS OK AS LONG AS IT WASNT PASSED TO API"
	};

global req_json_logout
req_json_logout = {
#	'name': user_name,
#	'domain': user_domain,
#	'reason': logout_reason,
#	'default_definition': "check around line 49 and 74",
	'YOU_HIT_THE_DEFAULT_DEF': "THIS IS OK AS LONG AS IT WASNT PASSED TO API"
	};
	
# The update_json function updates the json data to send to the API based on the called method
# 'add' method handles logins & handling key/value pairs to include in the json data
# 'delete' method handles logouts & handling key/value pairs to include in the json data
def update_json(user_action):
	if user_action == "add": # if logging in
		global req_json_login
		req_json_login = {
		'ip': user_ip,
		'name': user_name,
		'domain': user_domain,
		'type': user_type	
		};
	# else if action is delete/logout...
	elif user_action == "delete":
		# ...modify json data as needed
		global req_json_logout
		req_json_logout = {
		'name': user_name,
		'domain': user_domain,
		'reason': logout_reason
		};
			# SonicOS keeps reporting incorrectly formatted json when json data is sent with the delete request
			# Currently this code prepares the json data for sending, but must be included in the req.
			
			# Check push_logout() function for the commented req. The uncommented req is currently not sending JSON to avoid an error response.


# Request all params from the API
# Output is returned in XML
def req_all_params():
	global req
	req = requests.options(req_url, verify=False)
	print_response(req);

# Send the login request to the API and print the response
def push_login():
	global req
	update_json(user_action)
	
	if enable_authenticator is True:
		req_headers = { 'Authorization':'SNWL-API-Auth %s' % auth_authenticator }
		req = requests.post(req_url, headers=req_headers, json=req_json_login, verify=False)
	else:
		req = requests.post(req_url, json=req_json_login, verify=False)

	if dbg_on == True:
		print("DEBUG from push_login():", req_json_login, user_ip, user_name)
	print_response(req);

# Send the logout request to the API and print the response
def push_logout():
	global req
	update_json(user_action)

	if enable_authenticator is True:
		req_headers = { 'Authorization':'SNWL-API-Auth %s' % auth_authenticator }
		try:
			req = requests.delete(req_url + "/" + user_ip, headers=req_headers, json=req_json_logout, verify=False)
			print_response(req)
		except requests.exceptions.ConnectionError as req:
				print(req)
				print("Make sure your host IP is configured in the SonicOS 3rd Party API settings.")
	elif enable_authenticator is False:
		try:
			req = requests.delete(req_url + "/" + user_ip, json=req_json_logout, verify=False)
			print_response(req);
		except requests.exceptions.ConnectionError as req:
				print(req)
				print("Make sure your host IP is configured in the SonicOS 3rd Party API settings.")
	if dbg_on == True:
		print("DEBUG from push_logout():\n", req_json_logout)

def print_http_response():
	print("\n--------")
	print("Req Method:", req.request)
	print("URL:", req.url)
	print("HTTP", req.status_code, req.reason)
	print("Formatted Request Headers:")
	for ii in req.headers:
		print(ii + ":", req.headers[ii])
	print("--------\n")
def print_debug():
	print("\n----DEBUG START----")
	print("DEBUG:", "Req Method:", req.request)
	print("DEBUG:", req.url)
	print("DEBUG: HTTP", req.status_code, req.reason)
	print("DEBUG: Formatted Request Headers:")
	for ii in req.headers:
		print(ii + ":", req.headers[ii])
# Use below debugs for raw headers and content length, though it should be abailable above with the formatted headers
	#print("DEBUG:", req.headers)
	#print("DEBUG: Content-Length", req.headers['Content-Length'])
	#print("Extra debug info")
	#print(req.text, "<--TEXT")
	#print(req.json(), "<--JSON")
	print("")
	print("DEBUG: JSON data that would be sent to API:")
	print("If its a login...\n")
	print(req_json_login)
	print("---------------")
	print("If its a logout...\n")
	print(req_json_logout)
	print("")
	print("---------------")
	if user_action == "options":
		pass
	elif req.headers['Content-Length'] != '0':
		for i in req.json():
			print(str.capitalize(i) + ":", req.json()[i])		
	else:
		print("Content-Length is", req.headers['Content-Length'] + ".", "No response message from the API.")
	print("----DEBUG END----\n");
	
# Print results and headers
def print_response(req):
	print_http_response()
	if dbg_on == True:
		print_debug()
	if "OPTIONS" in str(req.request):
		print("\nOPTIONS request sent.")
		print("\nPrinting all parameters for use with the SSO API.")
		print("This script doesn't support _everything_ the API supports.")
		print("This output is informational and for testing communication with the API.")
		print("-------------")
		print(req.text)
# This else if statement will handle DELETE requests. If not a DELETE it moves on to else if below.
	elif "DELETE" in str(req.request):
		print("\nDELETE request sent.")
		global user_ip
		print("\nLogged user @", user_ip, "out of SonicOS.")
		print("-------------")
		# If there's response content iterate through the response content and capitalize the first letter and print the json response
		if req.headers['Content-Length'] != '0':
			for i in req.json():
				print(str.capitalize(i) + ":", req.json()[i])

# This code checks for successful HTTP requests and acts based on the content length/response content.
# This should fire only when the response is blank but had a successful HTTP request.
	elif req.status_code == 200 and req.text == "":
		print("\nEmpty response message, but the HTTP request was successful. \nCheck SonicOS for the new user session.")
		print("\nUser:", user_domain + "\\" + user_name, "at", user_ip)
# This iterates the json response and capitalzes the message header/key
# This should only hit when there is a response (like user was already logged in)
	elif req.headers['Content-Length'] > "0":
		for i in req.json():
			print(str.capitalize(i) + ":", req.json()[i])
			print("\nUser:", user_domain + "\\" + user_name, "at", user_ip);
			
# When we receive HTTP 401 Unauthorized response
# could be due to IP or shared secret configuration.
	elif req.status_code == 401:
                print("\nHTTP 401 Unauthorized. Verify shared secret and IP configuration.")
                print("If Medium/High level security are configured in SonicOS, it could mean your authenticator value is not the right length for the encryption selected, or was not sent to the API.")
                if req.headers['WWW-Authenticate'] == "SNWL-API-Auth":
                        print("SNWL-API-Auth header response recieved, but without supported hashes listed.")
                        if enable_authenticator == False:
                                print("Authenticator is disabled. No Authorization header was supplied in the request.")
                        else:
                                print("Authenticator is enabled.")
                        print("SonicOS expected authentication. Check SonicOS SSO API configuration.")
                        print("Enable the Authenticator in the client or set the shared secret security level to Low in SonicOS.")


# Argument parser code -- This handles the cli switches using the argparser module.
cliparser = argparse.ArgumentParser(description='SSO API Client integrates with the 3rd-Party Single-Sign-On API feature in SonicOS. That feature is not to be configured with the SonicOS API, which this client does not support.')
cliparser.add_argument('user_action', metavar='user_action', type=str, help='"add" to send login, "delete" for log out, and "options" for parameters.')
cliparser.add_argument(dest='user_ip', action="store", metavar='user_ip', type=str, help="The host IP to login")
cliparser.add_argument('-u', dest='user_name', action='store', type=str, help='User name to login. Required for logins, optional for logout.')
cliparser.add_argument('-d', dest='user_domain', action='store', type=str, help="User's FQDN or NETBIOS domain name. Optional. SonicOS will still check LDAP if enabled.")
cliparser.add_argument('-r', dest='logout_reason', action='store', type=str, help="Reason for logout. Optional. SonicOS uses this for logging only.")
cliargs = cliparser.parse_args()

# This updates the variables to the values of the passed arguments
user_action = cliargs.user_action
user_name = cliargs.user_name
user_ip = cliargs.user_ip
user_domain = cliargs.user_domain
logout_reason = cliargs.logout_reason
user_type = "domain"

if user_name is None:
		user_name = ""
if user_domain is None:
		user_domain = ""
if logout_reason is None:
		logout_reason = ""

update_json(user_action) # Update JSON data. Prep it for sending to API.

# If the action is a login, push the login.
if cliargs.user_action == "add":
	if user_name == "":
		print("User name (-u) is required for logins.")
		exit()
	push_login()
# Else if the action is a logoff, push the logoff.
elif cliargs.user_action == "delete":
	push_logout()
elif cliargs.user_action is not "add" or "delete" or "options":
	print("User action must be add, delete, or options.")
	exit()

# If user action is options, request all params from API.
if user_action == "options":
	req_all_params()
	user_action = ""