90 lines
3.7 KiB
Python
90 lines
3.7 KiB
Python
from servicebase import ServiceBase
|
|
import ipaddress
|
|
import requests
|
|
import json
|
|
|
|
class ServiceDelegate(ServiceBase) :
|
|
|
|
current_auth_header = None
|
|
return_payload = {}
|
|
host = None
|
|
|
|
def get_arguments(cls) :
|
|
return ['-n', '--net', 'store_true', "Return network information about the subject (bluecat)"]
|
|
|
|
def startup(self) :
|
|
for requirement in ['host','username','key'] :
|
|
if requirement not in self.config or (requirement in self.config and (self.config[requirement] is '' or type(self.config[requirement]) is not str)):
|
|
self.error.append('Missing required config option ' + requirement)
|
|
return
|
|
self.host = self.config['host']
|
|
self.current_auth_header = self.get_bc_auth_header(self.config['host'],self.config['username'],self.config['key'])
|
|
self.debug("Searching BlueCat for hosts...",1)
|
|
|
|
def perform_lookup(self,host_tuple) :
|
|
self.return_payload = {}
|
|
if self.current_auth_header :
|
|
object = self.search_bc_object(host_tuple)
|
|
self.return_payload.update({'object':object})
|
|
parent = self.search_bc_parent_object(host_tuple)
|
|
self.return_payload.update({'parent':parent})
|
|
return self.return_payload
|
|
else :
|
|
return self.return_payload
|
|
|
|
def shutdown(self) :
|
|
if self.host and self.current_auth_header :
|
|
logout = requests.get("https://" + self.host + "/Services/REST/v1/logout",headers=self.current_auth_header)
|
|
if not logout.text.endswith("successfully logged out.\"") :
|
|
self.error.append('Unable to log out of BlueCat API session')
|
|
|
|
|
|
def get_bc_auth_header(self,host,username,password) :
|
|
auth_token = requests.get("https://" + host + "/Services/REST/v1/login?username=" + username + "&password=" + password).text[17:71]
|
|
if auth_token.startswith('BAMAuthToken') :
|
|
return { "Authorization" : auth_token }
|
|
else :
|
|
return None
|
|
|
|
def search_bc_object(self,host_tuple) :
|
|
"""Searches BC for the subject."""
|
|
response = requests.get("https://" + self.host + "/Services/REST/v1/searchByObjectTypes?keyword=" + host_tuple[2][0] + "&types=IP4Address,IP6Address&start=0&count=10",headers=self.current_auth_header)
|
|
return self.clean_response(response)
|
|
|
|
def search_bc_parent_object(self,host_tuple) :
|
|
"""Search for and return information about the objects parent, if any"""
|
|
if 'object' in self.return_payload and type(self.return_payload['object']) is dict and 'id' in self.return_payload['object'] :
|
|
response = requests.get("https://" + self.host + "/Services/REST/v1/getParent?entityId=" + str(self.return_payload['object']['id']), headers=self.current_auth_header)
|
|
return self.clean_response(response)
|
|
# Make a guess that it belongs to a static net that hasn't been put in bluecat
|
|
elif host_tuple[2][0] != '' :
|
|
for net in ['/27', '/26', '/25', '/24', '/23', '/22'] :
|
|
try :
|
|
ip = ipaddress.IPv4Network(host_tuple[2][0] + net,strict=False)
|
|
network = str(ip.network_address)
|
|
response = requests.get("https://" + self.host + "/Services/REST/v1/searchByObjectTypes?keyword=" + network + "&types=IP4Network,IP4Block&start=0&count=10",headers=self.current_auth_header)
|
|
if not response.text.startswith('[]') :
|
|
return self.clean_response(response)
|
|
except:
|
|
pass
|
|
|
|
|
|
def clean_response(self,response) :
|
|
parsed = {}
|
|
try :
|
|
parsed = json.loads(response.text)
|
|
except Exception :
|
|
return parsed
|
|
if type(parsed) is list and len(parsed) > 0 :
|
|
parsed = parsed[0]
|
|
if type(parsed) is dict :
|
|
if 'properties' in parsed :
|
|
parsed["properties"] = parsed["properties"][:-1]
|
|
bad_delim_props = [x for x in parsed["properties"].split("|") if x.count('=') > 1]
|
|
if len(bad_delim_props) > 0 :
|
|
parsed.update({"unknown":bad_delim_props})
|
|
attributes = dict(x.split("=") for x in parsed["properties"].split("|") if x.count('=') == 1)
|
|
parsed.pop("properties", None)
|
|
parsed.update(attributes)
|
|
return parsed
|