I have been trying to generate a basic library that I can use at work, to call different REST calls towards a software vendor that we are using.
I would love someone's opinion on it, and what I could do better, hopefully with some examples or links, and some descriptions. I am by no means any good at Python, but I tried my best.
The library is meant to just be imported into an existing script..
Questions:
1. From a library point of view, can this be designed any different?
2. How could i make it any shorter / more understandable?
3. I feel that i need to learn more about args* kwargs**, is that needed here?
4. General coding style, naming conventions or breaking any standards (PEP8 im looking at you)?
#!/usr/bin/env python
"""Test SDK for Arcsight Logger"""
import time
import json
import datetime
import requests
import untangle
from requests.packages.urllib3.exceptions import InsecureRequestWarning
class ArcsightLogger(object):
"""
Main Class to interact with Arcsight Logger REST API
"""
def __init__(self):
self.target = 'https://SOMETHING:9000'
self.login = 'username'
self.password = 'password'
def post(self, url, data, isjson):
"""
Post Call towards Arcsight Logger
:param url: URL to retrieve
:param data: Request Body
:param isjson: Checks if post needs to be JSON
:return: HTTP Response
"""
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
if data:
if isjson:
try:
r = requests.post(url, json=data, verify=False)
return r
except requests.exceptions.RequestException as e:
print e
else:
try:
r = requests.post(url, data, verify=False)
return r
except requests.exceptions.RequestException as e:
print e
def arcsight_login(self):
"""
Log in the user defined in self.user
:return: User token to be used with all requests against Arcsight
"""
data = {
'login': self.login,
'password': self.password,
}
url = self.target + '/core-service/rest/LoginService/login'
r = self.post(url, data, False)
r.raise_for_status()
loginrequest = untangle.parse(r.content)
return loginrequest.ns3_loginResponse.ns3_return.cdata
def arcsight_search(self, token, query):
"""
Executes a searchquery, that is then stored and needs to be called again
to get results, using the returned searchid.
:param token: Token received from login method to authenticate
:param query: Query to be run with the search
:return: Array of the current searchid, which is needed for other functions,
and the content of HTTP response.
"""
data = {
'search_session_id': int(round(time.time() * 1000)),
'user_session_id': token,
'query': query,
}
url = self.target + '/server/search'
searchid = data['search_session_id']
r = (searchid, self.post(url, data, True))
return r
def arcsight_status(self, token, searchid):
"""
Checks the current status of a search using the searchid
:param token: Token received from login method to authenticate
:param searchid: The searchid that was generated when a new search was called
:return: The status of the search, currently this will wait for the search to complete
and then return that the search is finished.
"""
data = {
'search_session_id': searchid,
'user_session_id': token,
}
url = self.target + '/server/search/status'
r = self.post(url, data, True)
r = r.json()
while r['status'] != 'complete':
time.sleep(5)
print 'waiting'
r = self.post(url, data, True)
r = r.json()
print 'search is finished'
return r
def arcsight_events(self, token, searchid):
"""
Gathers events from a finished search
:param token: Token received from login method to authenticate
:param searchid: The searchid that was generated when a new search was called
:return: The events generated by a search. This returns the default arcsight
JSON format.
"""
data = {
'search_session_id': searchid,
'user_session_id': token,
}
url = self.target + '/server/search/events'
r = self.post(url, data, True)
self.arcsight_stop(token, searchid)
return r
def arcsight_events_custom(self, token, searchid):
"""
Gathers events from a finished search
:param token: Token received from login method to authenticate
:param searchid: The searchid that was generated when a new search was called
:return: The events generated by a search. This returns a custom JSON format
"""
data = {
'search_session_id': searchid,
'user_session_id': token,
}
url = self.target + '/server/search/events'
r = self.post(url, data, True)
d = json.dumps(r.json())
r = json.loads(d)
name = r['fields']
results = r['results']
a = []
for result in results:
a.append({f['name']: r for f, r in zip(name, result)})
r = (json.dumps(a, sort_keys=True, indent=4))
self.arcsight_stop(token, searchid)
return r
def arcsight_stop(self, token, searchid):
"""
Stops the search operation but keeps the search session so that the search results
can be narrowed down later.
:param token: Token received from login method to authenticate
:param searchid: The searchid that was generated when a new search was called
:return: A message that the search has been stopped.
"""
data = {
'search_session_id': searchid,
'user_session_id': token,
}
url = self.target + '/server/search/stop'
r = self.post(url, data, True)
print 'search stopped'
return r
def arcsight_close(self, token, searchid):
"""
Stops the execution of the search and clears the search session data from the server.
:param token: Token received from login method to authenticate
:param searchid: The searchid that was generated when a new search was called
:return: A message that the search has been stopped.
"""
data = {
'search_session_id': searchid,
'user_session_id': token,
}
url = self.target + '/server/search/close'
r = self.post(url, data, True)
print 'search is closed'
return r
def main(self):
"""
Testruns of different functions
"""
# token = self.arcsight_login()
# print token
# query = 'deviceAddress CONTAINS 192.168.2.26'
# r = self.arcsight_search(token, query)
# searchid = r[0]
# print searchid
# print r[1].content
# self.arcsight_status(token, searchid)
# r = self.arcsight_events(token, searchid)
# print r.content
if __name__ == "__main__":
o = ArcsightLogger()
o.main()
