Cronocide/ifxlookup
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Removed CPython equalities (is, is not) for 3.8 compatibility

Browse Source
This commit is contained in:
Daniel Dayley 2020-07-14 15:36:46 -06:00
parent 9745175b66
commit 28bb716019
8 changed files with 32 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
ifxlookup.py
View File

@ -86,16 +86,16 @@ class IFXLookup():
def use_plugins(self,plugins,reload=False) :
"""Defines plugins that should be used in a lookup, optionally forcing them to reload."""
# Verify data
if type(plugins) is not list :
if type(plugins) != list :
raise ValueError('argument \'plugins\' should be of type list')
for plugin in plugins :
# Check if the plugin is a string or a descendent of a ServiceBase class
if type(plugin) is not str and 'ServiceBase' not in [x.__name__ for x in inspect.getmro(plugin)] :
if type(plugin) != str and 'ServiceBase' not in [x.__name__ for x in inspect.getmro(plugin)] :
raise ValueError('unkown type for plugin')
# Find plugins by name using a default path
if type(plugin) is str :
if type(plugin) == str :
available_plugins = [y for x,y in self.search_plugins().items() if x == plugin and 'ServiceBase' in [z.__name__ for z in inspect.getmro(y)]]
if len(available_plugins) is 0 :
if len(available_plugins) == 0 :
raise FileNotFoundError(plugin + '.py not found')
plugin = available_plugins[0]
if 'ServiceBase' in [x.__name__ for x in inspect.getmro(plugin)] :
@ -124,7 +124,7 @@ class IFXLookup():
def configure(self,config) :
"""Updates the configuration of the lookup with the given config."""
if type(config) is not dict :
if type(config) != dict :
raise TypeError('argument \'config\' should be of type dict')
self._config.update(config)
@ -175,12 +175,12 @@ class IFXLookup():
raise KeyError('no configuration provided for plugin ' + delegatename)
# Assert that 'subjects' is a list
if type(subjects) is not list :
if type(subjects) != list :
raise TypeError('expected list but got ' + str(type(subjects)))
# Assert that subjects are strings
for subject in subjects :
if type(subject) is not str :
if type(subject) != str :
raise TypeError('subjects must be of type \'str\'')
for delegatename,delegate in self._delegate_map.items() :
@ -195,6 +195,7 @@ class IFXLookup():
report = {}
for subject in subjects :
# Perform delegate lookup
report.update({subject:delegate.lookup(subject)})
if delegate._error != [] :
error_info.update({delegatename:delegate._error})

10
plugins/aruba.py
View File

@ -21,7 +21,7 @@ class ServiceDelegate(ServiceBase) :
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)):
if requirement not in self._config or (requirement in self._config and (self._config[requirement] == '' or type(self._config[requirement]) != str)):
self._error.append('Missing required config option ' + requirement)
return
self._host = self._config['host']
@ -54,8 +54,8 @@ class ServiceDelegate(ServiceBase) :
return entry
# By value
for entry in self._aruba_basestations :
lowercase_values = [x.lower() for x in entry.values() if type(x) is str]
all_values = lowercase_values + [x for x in entry.values() if type(x) is not str]
lowercase_values = [x.lower() for x in entry.values() if type(x) == str]
all_values = lowercase_values + [x for x in entry.values() if type(x) != str]
for value in all_values :
if subject.lower() in str(value) :
return entry
@ -68,8 +68,8 @@ class ServiceDelegate(ServiceBase) :
return entry
# By value
for entry in self._aruba_hosts :
lowercase_values = [x.lower() for x in entry.values() if type(x) is str]
all_values = lowercase_values + [x for x in entry.values() if type(x) is not str]
lowercase_values = [x.lower() for x in entry.values() if type(x) == str]
all_values = lowercase_values + [x for x in entry.values() if type(x) == str]
for value in all_values :
if subject.lower() in str(value) :
return entry

10
plugins/bluecat.py
View File

@ -14,7 +14,7 @@ class ServiceDelegate(ServiceBase) :
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)):
if requirement not in self._config or (requirement in self._config and (self._config[requirement] == '' or type(self._config[requirement]) != str)):
self._error.append('Missing required config option ' + requirement)
return
self._host = self._config['host']
@ -26,7 +26,7 @@ class ServiceDelegate(ServiceBase) :
if self._current_auth_header :
objects = self.search_bc_objects(subject)
if objects :
if type(objects) is list :
if type(objects) == list :
object_list = []
for object in objects :
result = {'object':object}
@ -70,7 +70,7 @@ class ServiceDelegate(ServiceBase) :
def search_bc_parent_object(self,object) :
"""Search for and return information about the objects parent, if any"""
if type(object) is dict and 'id' in object :
if type(object) == dict and 'id' in object :
response = requests.get('https://' + self._host + '/Services/REST/v1/getParent?entityId=' + str(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
@ -102,9 +102,9 @@ class ServiceDelegate(ServiceBase) :
parsed = json.loads(response.text)
except Exception :
return parsed
if type(parsed) is list and len(parsed) > 0 :
if type(parsed) == list and len(parsed) > 0 :
for item in parsed:
item = clean(item)
return parsed
if type(parsed) is dict :
if type(parsed) == dict :
return clean(parsed)

4
plugins/f5.py
View File

@ -17,7 +17,7 @@ class ServiceDelegate(ServiceBase) :
def startup(self) :
for requirement in ['hosts','username','key'] :
if requirement not in self._config or (requirement in self._config and self._config[requirement] is ''):
if requirement not in self._config or (requirement in self._config and self._config[requirement] == ''):
self._error.append('Missing required config option ' + requirement)
return
self._hosts = self._config['hosts']
@ -65,7 +65,7 @@ class ServiceDelegate(ServiceBase) :
# Check if address is in the pool. If it is, return the node, pool, vip, and host.
# Naw, I'll just give up.
self.debug('Unable to get VIPs from ' + host,2)
if len(self._return_payload) is 0 :
if len(self._return_payload) == 0 :
self._return_payload = None
return self._return_payload

2
plugins/guid_channels.py
View File

@ -13,7 +13,7 @@ class ServiceDelegate(ServiceBase) :
# Load sub pack from options
for requirement in ['sub_pack'] :
if requirement not in self._config or (requirement in self._config and self._config[requirement] is ''):
if requirement not in self._config or (requirement in self._config and self._config[requirement] == ''):
self._error.append('Missing required config option ' + requirement)
return
self.sub_pack = self._config['sub_pack']

2
plugins/openvpn.py
View File

@ -13,7 +13,7 @@ class ServiceDelegate(ServiceBase) :
def startup(self) :
for requirement in ['hosts','username','key'] :
if requirement not in self._config or (requirement in self._config and self._config[requirement] is ''):
if requirement not in self._config or (requirement in self._config and self._config[requirement] == ''):
self._error.append('Missing required config option ' + requirement)
rself._error
self.hosts = self._config['hosts']

6
plugins/paloalto.py
View File

@ -138,7 +138,7 @@ class ServiceDelegate(ServiceBase) :
rules = {'rules':{x:included_rules[host][x] for x in included_rules[host] if len(included_rules[host][x]) > 0}}
if len(rules['rules']) > 0 :
result[host].update(rules)
if len(result[host]) is 0 :
if len(result[host]) == 0 :
result.pop(host,None)
if len(result) > 0 :
@ -285,7 +285,7 @@ class ServiceDelegate(ServiceBase) :
try :
key_payload = requests.get('https://' + str(ip) + '/api/?type=keygen&user=' + user + '&password=' + key,verify=False,timeout=3)
except Exception as exception :
if type(exception) is requests.exceptions.ReadTimeout :
if type(exception) == requests.exceptions.ReadTimeout :
self._error.append('Unable to log into PA device ' + hostname + ': timed out.')
self.debug('Unable to log into PA device ' + hostname + ': timed out.',1)
return None
@ -318,7 +318,7 @@ class ServiceDelegate(ServiceBase) :
def config_valid(self) :
for requirement in ['hosts','username','key'] :
if requirement not in self._config or (requirement in self._config and self._config[requirement] is ''):
if requirement not in self._config or (requirement in self._config and self._config[requirement] == ''):
self._error.append('Missing required config option ' + requirement)
return False
return True

14
plugins/shodan.py
View File

@ -15,7 +15,7 @@ class ServiceDelegate(ServiceBase) :
def startup(self) :
for requirement in ['keys'] :
if requirement not in self._config or (requirement in self._config and self._config[requirement] is ''):
if requirement not in self._config or (requirement in self._config and self._config[requirement] == ''):
self._error.append('Missing required config option ' + requirement)
return
keys = self._config['keys']
@ -35,27 +35,27 @@ class ServiceDelegate(ServiceBase) :
pass
def _clean_newlines(self,structure) :
if type(structure) is dict :
if type(structure) == dict :
newstructure = {}
for key in structure :
if key :
if structure[key] and type(structure[key]) is dict or type(structure[key]) is list :
if structure[key] and type(structure[key]) == dict or type(structure[key]) == list :
newkey = key.replace('\n','')
newstructure.update({newkey:self._clean_newlines(structure[key])})
else :
newkey = key.replace('\n','')
if structure[key] and type(structure[key]) is str :
if structure[key] and type(structure[key]) == str :
newvalue = structure[key].replace('\n','')
newstructure.update({newkey:newvalue})
else :
newstructure.update({newkey:None})
return newstructure
if type(structure) is list :
if type(structure) == list :
newstructure = []
for item in structure :
if type(item) is dict or type(item) is list :
if type(item) == dict or type(item) == list :
newstructure.append(self._clean_newlines(item))
if type(item) is str :
if type(item) == str :
newstructure.append(item.replace('\n',''))
else :
newstructure.append(item)