6
\$\begingroup\$

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()
\$\endgroup\$
  • \$\begingroup\$ @Dex'ter Haha yes, i was quite mad when i wrote that, i guess i should remove it, as that one function is not called, since it does not produce the exact result needed. \$\endgroup\$ – Marius Nov 13 '16 at 20:49
  • \$\begingroup\$ if that method does not return the expected result and it's not supposed to be reviewed (it shouldn't as here we only review working code), please remove it from the question too. The question as it is might be interpreted as off-topic because of it :) \$\endgroup\$ – Grajdeanu Alex. Nov 13 '16 at 20:51
  • \$\begingroup\$ @Dex'ter Done and done :) \$\endgroup\$ – Marius Nov 13 '16 at 20:53
5
\$\begingroup\$

Creating a session

As a library, I find it barely usable. If I want to use it with my own credentials, I have to:

  • either modify the source of your library to put them in __init__;
  • or tamper with the attributes after building an ArcsightLogger object:

    o = ArcsightLogger()
    o.login = 'spam'
    o.password = 'eggs'
    o.main()
    

Moreover, having to manually store and feed back to each method the generated token is unnecessary boilerplate.

Instead, I would try to log the user in as soon as possible and store the generated token in an attribute for easy access by each of your methods:

class ArcsightLogger(object):
    """
    Main Class to interact with Arcsight Logger REST API
    """

    TARGET = 'https://SOMETHING:9000'

    def __init__(self, username, password):
        """
        Log in the user whose credentials are provided and
        store the access token to be used with all requests
        against Arcsight
        """

        data = {
            'login': username,
            'password': password,
        }
        url = self.TARGET + '/core-service/rest/LoginService/login'
        r = self.post(url, data, False)
        r.raise_for_status()
        loginrequest = untangle.parse(r.content)
        self.token = loginrequest.ns3_loginResponse.ns3_return.cdata

Note the use of the class constant TARGET instead of an instance attribute as this is not something meant to be changed when using the API.

Posting data

First off, a comment would be more than helpfull to know why you requests.packages.urllib3.disable_warnings(InsecureRequestWarning). Second, do you really need to call this at each request? I would rather put this line right after your last import. Or not at all, since it let the user know about a potential MITM attack. An other solution would be to have a configurable behaviour that let the warnings appear by default but can be disabled by the user if they so wish. I would however use the warnings module to control that.

Next, I would change the signature of _post a bit. I would ask the caller to provide only the route part of the URL as the domain is arleady stored as a class constant. It will avoid some boilerplate as the concatenation can be done in this method. I would also take advantage of Python's syntax to turn data into a dictionnary instead of letting the caller do it. I would change the isjson parameter to accept a default value of True since this is what most method uses. And, lastly, I would name it _post as it is mainly an helper function for your methods rather than part of your public API.

Finally, you should let the exceptions bubble up rather than print them as they will most likely indicate an issue that will prevent further processing:

import warnings
from requests.packages.urllib3.exceptions import InsecureRequestWarning


class ArcsightLogger(object):
    """
    Main Class to interact with Arcsight Logger REST API
    """

    TARGET = 'https://SOMETHING:9000'

    def __init__(self, username, password, disable_insecure_warning=False):
        """
        Log in the user whose credentials are provided and
        store the access token to be used with all requests
        against Arcsight
        """

        action = 'ignore' if disable_insecure_warning else 'once'
        warnings.simplefilter(action, InsecureRequestWarning)

        r = self.post(
            '/core-service/rest/LoginService/login',
            login=username, password=password, is_json=False)
        r.raise_for_status()
        loginrequest = untangle.parse(r.content)
        self.token = loginrequest.ns3_loginResponse.ns3_return.cdata

    def post(self, route, is_json=True, **data):
        """
        Post Call towards Arcsight Logger
        :param route: API endpoint to fetch
        :param is_json: Checks if post needs to be JSON
        :param data: Request Body
        :return: HTTP Response
        """

        if not data:
            return

        url = self.TARGET + route
        if isjson:
            return requests.post(url, json=data, verify=False)
        else:
            return requests.post(url, data, verify=False)

Verfying search status

Your arcsight_status method is not optimal. As a user, knowing that an operation can take a huge amount of time, I prefer have the ability to not perform a blocking call so I can perform other operations in the meantime. You could split the functionnality by providing an arcsight_search_complete method returning a boolean and build arcsight_status on top of it:

def arcsight_search_complete(self, search_id):
    """
    Checks the current status of a search using the search_id
    :param search_id: The search_id that was generated when a new search was called
    :return: Whether or not the search finished already.
    """

    response = self.post(
        '/server/search/status',
        search_session_id=search_id,
        user_session_id=self.token)
    return response.json().get('status') == 'complete'

def arcsight_wait_for_search(self, search_id):
    """
    Blocks until the search represented by search_id completes
    :param search_id: The search_id that was generated when a new search was called
    :return: The status of the search.
    """

    while not self.arcsight_search_complete(search_id):
        time.sleep(5)
    return self.post(
        '/server/search/status',
        search_session_id=search_id,
        user_session_id=self.token).json()

Generating events reports

I don't understand why you return an HTTP response in arcsight_events and try extra hard to return a string in arcsight_events_custom. For one, these two methods seek to offer the same kind of data (only filtered differently) so they should return the same type of data. For two, these data types are barely usable on their own. Why not directly return a dictionary that the user could manipulate or format the way they wish?

arcsight_events_custom would return a list, but at least they are both collections that can be directly manipulated.

Lastly, since they perform essentially the same task, I would have one rely on the other to avoid code duplication:

def arcsight_events(self, search_id):
    """
    Gathers events from a finished search
    :param search_id: The search_id that was generated when a new search was called
    :return: The events generated by a search. This returns the default arcsight
    JSON format.
    """

    response = self.post(
        '/server/search/events',
        search_session_id=search_id,
        user_session_id=self.token)
    self.arcsight_stop(token, searchid)
    return response.json()

def arcsight_events_custom(self, search_id):
    """
    Gathers events from a finished search
    :param search_id: The search_id that was generated when a new search was called
    :return: The events generated by a search. This returns a custom JSON format
    """

    events = self.arcsight_events(search_id)
    return [{
        field['name']: result
        for field, result in zip(events['fields'], results)
    } for results in events['results']]

Generic remarks

  • Since the class is called ArcsightLogger, I don't understand the decision to prefix each method with arcsight_; it doesn't really add any value.
  • The main method has nothing to do in this class. Such code should either be in the if __name__ == '__main__': clause if it is test or demo code, or in a user script.
  • You may want to reduce the amount of text per line in your docstrings. PEP 8 recommends to limit such lines to 72 characters.
  • I don't really have any insights of how it could be built, but you could provide a context manager around a search query so that the user don't have to bother storing the search ID. Something that could work along the lines of:

    arcsight = ArcsightLogger('me', 'mypass', True)
    with arcsight.search('query') as search:
        search.wait()
        data = search.events(custom=True)
    # auto close at the end of the with block
    

Putting all that together your code can become:

#!/usr/bin/env python2

"""Test SDK for Arcsight Logger"""

import time
import warnings

import untangle
import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning


class ArcsightLogger(object):
    """
    Main Class to interact with Arcsight Logger REST API
    """

    TARGET = 'https://SOMETHING:9000'

    def __init__(self, username, password, disable_insecure_warning=False):
        """
        Log in the user whose credentials are provided and
        store the access token to be used with all requests
        against Arcsight
        """

        action = 'ignore' if disable_insecure_warning else 'once'
        warnings.simplefilter(action, InsecureRequestWarning)

        r = self._post(
            '/core-service/rest/LoginService/login',
            login=username, password=password, is_json=False)
        r.raise_for_status()
        loginrequest = untangle.parse(r.content)
        self.token = loginrequest.ns3_loginResponse.ns3_return.cdata

    def _post(self, route, is_json=True, **data):
        """
        Post Call towards Arcsight Logger
        :param route: API endpoint to fetch
        :param is_json: Checks if post needs to be JSON
        :param data: Request Body
        :return: HTTP Response
        """

        if not data:
            return

        url = self.TARGET + route
        if is_json:
            return requests.post(url, json=data, verify=False)
        else:
            return requests.post(url, data, verify=False)

    def search(self, query):
        """
        Executes a searchquery, that is then stored and needs
        to be called again to get results, using the returned
        search_id.
        :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.
        """

        search_id = int(round(time.time() * 1000))
        response = self._post(
            '/server/search', query=query,
            search_session_id=search_id,
            user_session_id=self.token)
        return search_id, response.json()

    def search_complete(self, search_id):
        """
        Checks the current status of a search using the search_id
        :param search_id: The search_id that was generated
                          when a new search was called
        :return: Whether or not the search finished already.
        """

        response = self._post(
            '/server/search/status',
            search_session_id=search_id,
            user_session_id=self.token)
        return response.json().get('status') == 'complete'

    def wait(self, search_id):
        """
        Blocks until the search represented by search_id completes
        :param search_id: The search_id that was generated
                          when a new search was called
        :return: The status of the search.
        """

        while not self.search_complete(search_id):
            time.sleep(5)
        return self._post(
            '/server/search/status',
            search_session_id=search_id,
            user_session_id=self.token).json()

    def events(self, search_id, custom_format=False):
        """
        Gathers events from a finished search
        :param search_id: The search_id that was generated
                          when a new search was called
        :param custom_format: Whether to return the response from
                              ArcSight unmodified or to pre-process it.
        :return: The events generated by a search.
        """

        response = self._post(
            '/server/search/events',
            search_session_id=search_id,
            user_session_id=self.token)
        self.arcsight_stop(token, searchid)
        events = response.json()

        if not custom_format:
            return events
        return [{
            field['name']: result
            for field, result in zip(events['fields'], results)
        } for results in events['results']]

    def stop(self, search_id):
        """
        Stops the search operation but keeps the search
        session so that the search results can be narrowed
        down later.
        :param search_id: The search_id that was generated
                          when a new search was called
        :return: A message that the search has been stopped.
        """

        response = self._post(
            '/server/search/stop',
            search_session_id=search_id,
            user_session_id=self.token)
        return response.json()

    def close(self, search_id):
        """
        Stops the execution of the search and clears
        the search session data from the server.
        :param search_id: The search_id that was generated
                          when a new search was called
        :return: A message that the search has been stopped.
        """

        response = self._post(
            '/server/search/close',
            search_session_id=search_id,
            user_session_id=self.token)
        return response.json()


if __name__ == '__main__':
    arcsight = ArcsightLogger('username', 'password', True)
    print arcsight.token
    query = 'deviceAddress CONTAINS 192.168.2.26'
    search_id, r = arcsight.search(query)
    print searchid
    print r
    arcsight.wait(search_id)
    events = arcsight.events(search_id)
    print events

I also removed most of the prints as this is a bad idea to mess with the users output: they may want to have their own that they need to parse afterwards or whatever. Consider using the logging module instead as it is easy to turn it off or redirect it to an other stream.

\$\endgroup\$
  • \$\begingroup\$ that's an outstanding answer. Congrats :) I really enjoyed reading it. \$\endgroup\$ – Grajdeanu Alex. Nov 14 '16 at 7:08
  • \$\begingroup\$ Thanks again @Mathias , i have a few comments that i would like to know. I like the way you handled the SSL errors, this is due to this is always shipped, and mostly implemented with a self signed certificate. I do have one question, the "custom_format". What i currently have with the default format is this: link What i converted it to was this: link When i printed out the custom format now, it does not make any sense anymore, am i missing something? \$\endgroup\$ – Marius Nov 14 '16 at 15:24
  • \$\begingroup\$ @Marius Oh, sorry for that, I misread the code at first and didn't understood that the results field was a list of lists. I reused your code but changed it to a list-comprehension, see update. \$\endgroup\$ – 409_Conflict Nov 14 '16 at 15:44
  • \$\begingroup\$ @MathiasEttinger That solved it! This would now work in theory, as a library correct? Also, when i made this, and tried to look up standards, i did not find anything on "standards to make a sdk/library", is there any known standards for this? \$\endgroup\$ – Marius Nov 14 '16 at 15:49
  • \$\begingroup\$ @Marius You could start with the official documentation \$\endgroup\$ – 409_Conflict Nov 14 '16 at 15:57

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.