2
\$\begingroup\$

At work we are interfacing with the Jira cloud storage for our ticket management system. I created a base class to make the process easier on the user side, and I did my best to make it re-usable and generic. I wasn't able to figure out a generic way to gain the access token authorization since I used Selenium to manipulate a form for my specific application, so I figured it should be specified in each individual application.

I am just starting my career as a data engineer, and am looking to learn best practices and good design.

#!/usr/bin/env python
"""
This module contains two classes, one that creates an Oauth object that can be
used to do the "Oauth dance", and another that will create RSA_SHA1 signatures.
"""
import base64

import oauth2 as oauth
from tlslite.utils import keyfactory
from urllib import parse


class SigMethodRSA(oauth.SignatureMethod):

    def __init__(self, rsa_key):
        self.rsa_key = rsa_key

    name = 'RSA-SHA1'

    def signing_base(self, request, consumer, token):
        if not hasattr(
                request, 'normalized_url') or request.normalized_url is None:
            raise ValueError("Base URL for request is not set.")

        sig = (
            oauth.escape(request.method),
            oauth.escape(request.normalized_url),
            oauth.escape(request.get_normalized_parameters()),
        )
        key = '%s&' % oauth.escape(consumer.secret)
        if token:
            key += oauth.escape(token.secret)
        raw = '&'.join(sig)
        return key, raw

    def sign(self, request, consumer, token):
        """Builds the base signature string."""
        key, raw = self.signing_base(request, consumer, token)
        raw = bytes(raw, 'utf-8')
        # TODO figure out how to add
        with open(self.rsa_key, 'r') as f:
            data = f.read()
        private_key_string = data.strip()
        privatekey = keyfactory.parsePrivateKey(private_key_string)
        signature = privatekey.hashAndSign(raw)

        return base64.b64encode(signature)


class OauthBase(object):

    def __init__(self, urls, consumer, sig_method):
        """

        :param urls: A dictionary containing all of the necessary url's to do
                     the oauth dance.  Includes a base url, authorization url,
                     request token url, access token url, and a data url.  Will
                     raise error if url's are missing.
        :param consumer: A dictionary containing the public and private
                         consumer keys.  Will raise error if both keys aren't
                         included.
        :param sig_method: A signing method used to create the SHA1 key.  If
                           using the oauth2.SignatureMethod, pass this
                           parameter in like the following example:
                           sig_method=SigMethodRSA(rsa_key)
        :return:
        """
        self.base_url = urls['base_url']
        self.auth_url = urls['authorization_url']
        self.req_tok_url = urls['request_token_url']
        self.acc_tok_url = urls['access_token_url']
        self.data_url = urls['data_url']
        self.sig_method = sig_method

        self.cons_key = consumer['consumer_key']
        self.cons_secret = consumer['consumer_secret']
        self.consumer = oauth.Consumer(self.cons_key, self.cons_secret)

    def gain_authorization(self):
        """
        To be specified by each individual application based on individual
        needs.
        """
        raise NotImplementedError

    def initial_access(self):
        """Check for initial access to the data url."""
        client = self._set_client()
        response, content = client.request(self.data_url, 'GET')
        if response['status'] != '401':
            raise Exception('Do not have access to data url!')
        else:
            print('Initial access gained.')
        # return response, content

    def get_request_token(self):
        """Retrieves the private and public request tokens."""
        client = self._set_client(sig_meth=True)
        response, content = client.request(self.req_tok_url, 'POST')
        if response['status'] != '200':
            raise Exception('Invalid response {}: {}'.format(
                response['status'], content
            ))
        request_token = dict(parse.parse_qsl(content))
        r_pub = request_token[b'oauth_token']
        r_priv = request_token[b'oauth_token_secret']
        print('Request Token:')
        print('     -oauth_token => {}'.format(r_pub))
        print('     -oauth_secret_token => {}'.format(r_priv))
        self.request_token = oauth.Token(r_pub, r_priv)
        return r_pub

    def oauth_dance(self):
        """Performs the various steps to gain OAuthorization up to the request
           portion.  Subclass to perform the authorization step.  Suggested
           order is included in comments below"""
        self.initial_access()
        req_pub = self.get_request_token()
        return req_pub

    def oauth_dance2(self):
        """
        Assumes that authorization has been gained and finishes the process.
        """
        self.get_access_token()

    def get_access_token(self):
        client = self._set_client(self.request_token, sig_meth=True)
        response, content = client.request(self.acc_tok_url, 'POST')
        access_token = dict(parse.parse_qsl(content))
        acc_pub = access_token[b'oauth_token']
        acc_priv = access_token[b'oauth_token_secret']
        print('Access Token:')
        print('     -oauth_token => {}'.format(acc_pub))
        print('     -oauth_secret_token => {}'.format(acc_priv))
        self.access_token = oauth.Token(acc_pub, acc_priv)

    def _set_client(self, token=None, sig_meth=False):
        if token and sig_meth:
            client = oauth.Client(self.consumer, token)
            client.set_signature_method(self.sig_method)
            return client
        elif token and not sig_meth:
            return oauth.Client(self.consumer, token)
        elif sig_meth and not token:
            client = oauth.Client(self.consumer)
            client.set_signature_method(self.sig_method)
            return client
        else:
            return oauth.Client(self.consumer)

Here is my subclass with the method to gain authorization:

class JiraOauth(OauthBase):

    @staticmethod
    def gain_authorization(site):
        driver = webdriver.PhantomJS(
            r'/Applications/phantomjs-2.1.1-macosx/bin/phantomjs')
        # driver = webdriver.Firefox()
        driver.get(site)
        user = driver.find_element_by_id('username')
        user.send_keys('name@email.com')
        passw = driver.find_element_by_id(('password'))
        passw.send_keys('password')
        log_in = driver.find_element_by_id('login')
        log_in.click()
        allow = driver.find_element_by_id('approve')
        allow.click()

And finally, here is my sample usage that I got to work:

signature_method = SigMethodRSA(RSA_KEY_PATH)

o = JiraOauth(o_urls, o_cons, signature_method)
req_pub = o.oauth_dance()

authorization_url = authorization_url + '?oauth_token={}'.format(
    req_pub.decode('ascii'))

o.gain_authorization(authorization_url)

o.oauth_dance2()
\$\endgroup\$

Your Answer

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

Browse other questions tagged or ask your own question.