CumulusCI-OAuth-Tool/auth.py at master · jlantz/CumulusCI-OAuth-Tool · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
'''
from https://github.com/neworganizing/salesforce-oauth2/blob/master/salesforce_oauth2/auth.py

Inspired by requests_oauth2
'''

import requests
from urllib import quote

import hmac
import hashlib
import base64

class SalesforceOAuth2(object):
    authorization_url = '/services/oauth2/authorize'
    token_url = '/services/oauth2/token'
    revoke_url = '/services/oauth2/revoke'

    def __init__(self, client_id, client_secret, redirect_uri, **kwargs):
        self.sandbox = kwargs.get('sandbox', False)

        if self.sandbox:
            self.auth_site = 'https://test.salesforce.com'
        else:
            self.auth_site = 'https://login.salesforce.com'

        self.client_id = client_id
        self.client_secret = client_secret
        self.redirect_uri = redirect_uri

    def _request_token(self, data):
        import requests
        url = "{site}{token_url}".format(
            site=self.auth_site, token_url=self.token_url)
        headers = {
            'Content-Type': 'application/x-www-form-urlencoded'
        }
        post_data = {'client_id': self.client_id,
                     'client_secret': self.client_secret}
        post_data.update(data)
        result = requests.post(url, data=post_data, headers=headers)
        return result, result.json()

    def authorize_url(self, **kwargs):
        from urllib import quote
        scope = kwargs.get('scope', quote('full'))
        fields = {
            'site': self.auth_site,
            'authorize_url': self.authorization_url,
            'clientid': self.client_id,
            'redirect_uri': quote(self.redirect_uri),
            'scope': scope
        }
        return "{site}{authorize_url}?response_type=code&client_id={clientid}&redirect_uri={redirect_uri}&scope={scope}".format(**fields)

    def get_token(self, code):
        from urllib import quote
        data = {
            'grant_type': 'authorization_code',
            'redirect_uri': self.redirect_uri,
            'code': code
        }
        response, response_json = self._request_token(data)
        if 'access_token' in response_json:
            self.access_token = response_json['access_token']
        if 'refresh_token' in response_json:
            self.refresh_token = response_json['refresh_token']
        return response_json

    def refresh_token(self, refresh_token):
        data = {
            'grant_type': 'refresh_token',
            'refresh_token': refresh_token
        }
        response, response_json = self._request_token(data)

        if 'access_token' in response_json:
            self.access_token = response_json['access_token']
        return response_json

    def generate_signature(self, id, issued_at):
        data = "{id}{issued}".format(id=id, issued=issued_at)
        digest = hmac.new(
            self.client_secret, data, digestmod=hashlib.sha256).digest()
        return base64.b64encode(digest).decode()

    def revoke_token(self, current_token):
        import requests
        from urllib import quote
        # Perform a GET request, because that's by far the easiest way
        url = "{site}{revoke_url}".format(
            site=self.auth_site, revoke_url=self.revoke_url)
        data = {
            'token': quote(current_token)
        }
        headers = {
            'Content-Type': 'application/x-www-form-urlencoded'
        }
        return requests.post(url, data=data, headers=headers)