Merge pull request #19 from RockefellerArchiveCenter/issue-18

Adds OAuth authorization endpoint

electronbonder/client.py

@@ -1,8 +1,10 @@
 import json
 from urllib.parse import urlparse
 
+from oauthlib.oauth2 import BackendApplicationClient
 from requests import Session
 from requests.adapters import HTTPAdapter
+from requests_oauthlib import OAuth2Session
 from six import add_metaclass
 from urllib3.util.retry import Retry
 
@@ -90,6 +92,21 @@ def authorize(self):
                 session_token)
             return session_token
 
+    def authorize_oauth(self):
+        """Authorizes the client using a configured OAuth provider."""
+        try:
+            client = BackendApplicationClient(
+                client_id=self.config["oauth_client_id"])
+            oauth = OAuth2Session(client=client)
+            token = oauth.fetch_token(
+                token_url=f"{self.config['oauth_client_baseurl']}/oauth2/token",
+                client_id=self.config["oauth_client_id"],
+                client_secret=self.config["oauth_client_secret"])
+        except Exception as e:
+            raise ElectronBondAuthError(
+                f"Failed to authorize OAuth with message: {str(e)}")
+        self.session.headers["Authorization"] = token["access_token"]
+
     def get_paged(self, url, *args, **kwargs):
         """get list of json objects from urls of paged items"""
         params = {"page": 1}

requirements.txt

@@ -1,2 +1,3 @@
 requests~=2.28
+requests-oauthlib~=1.3
 six~=1.16

setup.py

@@ -11,7 +11,7 @@
     long_description_content_type="text/markdown",
     author="Rockefeller Archive Center",
     author_email="archive@rockarch.org",
-    version="1.1",
+    version="2.0",
     license='MIT',
     packages=find_packages(),
     zip_safe=False,

tests/test_client.py

@@ -5,6 +5,9 @@
 from electronbonder.client import ElectronBond, ElectronBondAuthError
 
 BASEURL = "http://localhost:8007"
+OAUTH_BASEURL = "https://awscognito.com"
+OAUTH_CLIENT_ID = "123456789"
+OAUTH_CLIENT_SECRET = "987654321"
 PATH = "custom/path"
 
 
@@ -14,7 +17,10 @@ def setUp(self):
         self.client = ElectronBond(
             baseurl=BASEURL,
             username="username",
-            password="password")
+            password="password",
+            oauth_client_baseurl=OAUTH_BASEURL,
+            oauth_client_id=OAUTH_CLIENT_ID,
+            oauth_client_secret=OAUTH_CLIENT_SECRET)
 
     @patch("requests.Session.post")
     def test_authorize(self, mock_post):
@@ -29,6 +35,17 @@ def test_authorize(self, mock_post):
             mock_post.return_value.status_code = 404
             self.client.authorize()
 
+    @patch("requests_oauthlib.OAuth2Session.fetch_token")
+    def test_authorize_oauth(self, mock_token):
+        token = "12345"
+        mock_token.return_value = {"access_token": token}
+        self.client.authorize_oauth()
+        self.assertEqual(self.client.session.headers["Authorization"], token)
+        mock_token.assert_called_with(
+            token_url=f"{OAUTH_BASEURL}/oauth2/token",
+            client_id=OAUTH_CLIENT_ID,
+            client_secret=OAUTH_CLIENT_SECRET)
+
     @patch("requests.Session.get")
     def test_get_paged(self, mock_get):
         list(self.client.get_paged(PATH))