feat: Added client_secret support and refresh token functionality (#90)

* Added client_secret and refresh token function

* Update oauth2_cli_auth/code_grant.py

Co-authored-by: Timo Reymann <mail@timo-reymann.de>

* Update oauth2_cli_auth/code_grant.py

Co-authored-by: Timo Reymann <mail@timo-reymann.de>

* Update oauth2_cli_auth/code_grant.py

Co-authored-by: Timo Reymann <mail@timo-reymann.de>

* Update oauth2_cli_auth/code_grant.py

Co-authored-by: Timo Reymann <mail@timo-reymann.de>

* Implement suggestions from PR review and added unit tests for token refresh

* Fixed SonarCloud warnings

---------

Co-authored-by: Robert Kentish <robert@jakarisolutions.com>
Co-authored-by: Timo Reymann <mail@timo-reymann.de>

Author: 3 people

oauth2_cli_auth/__init__.py

@@ -3,5 +3,5 @@
 """
 from oauth2_cli_auth.__version__ import __version__
 from oauth2_cli_auth.http_server import OAuthCallbackHttpServer
-from oauth2_cli_auth.code_grant import OAuth2ClientInfo, exchange_code_for_access_token, get_auth_url, open_browser, load_oidc_config
+from oauth2_cli_auth.code_grant import OAuth2ClientInfo, refresh_access_token, exchange_code_for_access_token, exchange_code_for_response, get_auth_url, open_browser, load_oidc_config
 from oauth2_cli_auth.simplified_flow import get_access_token_with_browser_open

oauth2_cli_auth/code_grant.py

@@ -26,11 +26,14 @@ class OAuth2ClientInfo:
     client_id: str
     """Id of the client to request for"""
 
+    client_secret: str | None
+    """Secret of the client to request for"""
+
     scopes: list[str]
     """List of scopes to request"""
 
     @staticmethod
-    def from_oidc_endpoint(oidc_config_endpoint: str, client_id: str, scopes: list[str]) -> "OAuth2ClientInfo":
+    def from_oidc_endpoint(oidc_config_endpoint: str, client_id: str, scopes: list[str] = [], client_secret: str | None = None) -> "OAuth2ClientInfo":
         """
         Create client information object from well known endpoint in format as specified in
         https://openid.net/specs/openid-connect-discovery-1_0.html#ProviderMetadata
@@ -45,6 +48,7 @@ def from_oidc_endpoint(oidc_config_endpoint: str, client_id: str, scopes: list[s
             authorization_url=config.get("authorization_endpoint"),
             token_url=config.get("token_endpoint"),
             client_id=client_id,
+            client_secret=client_secret,
             scopes=scopes,
         )
 
@@ -102,7 +106,7 @@ def exchange_code_for_response(
     """
     headers = {
         "Content-Type": "application/x-www-form-urlencoded",
-        "Authorization": "Basic " + base64.b64encode(f"{client_info.client_id}:".encode()).decode(),
+        "Authorization": "Basic " + base64.b64encode(f"{client_info.client_id}:{client_info.client_secret if client_info.client_secret is not None else ''}".encode()).decode(),
     }
 
     data = {
@@ -117,6 +121,32 @@ def exchange_code_for_response(
 
     return json_response
 
+def refresh_access_token(
+        client_info: OAuth2ClientInfo,
+        refresh_token: str,
+) -> dict:
+    """
+    Refresh an access token using the endpoints from client info
+
+    :param client_info: Info about oauth2 client
+    :param refresh_token: Refresh token to use
+    :return: Response from OAuth2 endpoint
+    """
+    headers = {
+        "Content-Type": "application/x-www-form-urlencoded",
+        "Authorization": "Basic " + base64.b64encode(f"{client_info.client_id}:{client_info.client_secret if client_info.client_secret is not None else ''}".encode()).decode(),
+    }
+
+    data = {
+        "refresh_token": refresh_token,
+        "grant_type": "refresh_token",
+    }
+    encoded_data = urllib.parse.urlencode(data).encode('utf-8')
+
+    request = urllib.request.Request(client_info.token_url, data=encoded_data, headers=headers)
+    json_response = _load_json(request)
+
+    return json_response
 
 def exchange_code_for_access_token(
         client_info: OAuth2ClientInfo,

oauth2_cli_auth/code_grant_test.py

@@ -3,27 +3,46 @@
 from unittest.mock import patch
 
 from oauth2_cli_auth import OAuth2ClientInfo, get_auth_url, exchange_code_for_access_token, load_oidc_config
+from oauth2_cli_auth.code_grant import exchange_code_for_response, refresh_access_token
+
+REDIRECT_URI = "http://localhost:123"
 
 client_info = OAuth2ClientInfo(
     client_id="dummy",
+    client_secret=None,
     authorization_url="https://auth.com/oauth/authorize",
     token_url="https://auth.com/oauth/token",
-    scopes=["openid", "profile"]
+    scopes=["openid", "profile"],
 )
 
 
 def test_get_auth_url():
-    auth_url = get_auth_url(client_info, "http://localhost:123")
+    auth_url = get_auth_url(client_info, REDIRECT_URI)
     assert auth_url == (
-        'https://auth.com/oauth/authorize?client_id=dummy&redirect_uri=http://localhost:123&scope=openid+'
+        f'https://auth.com/oauth/authorize?client_id=dummy&redirect_uri={REDIRECT_URI}&scope=openid+'
         'profile&response_type=code'
     )
 
+def test_exchange_code_for_response(create_urlopen_mock):
+    with create_urlopen_mock(io.BytesIO(b'{"access_token":"the_token","token_type":"Bearer","expires_in":3600,"refresh_token":"the_refresh_token","scope":"create"}')):
+        response = exchange_code_for_response(client_info, REDIRECT_URI, "code")
+        assert "the_token" == response.get("access_token")
+        assert "Bearer" == response.get("token_type")
+        assert 3600 == response.get("expires_in")
+        assert "the_refresh_token" == response.get("refresh_token")
+        assert "create" == response.get("scope")
+
+def test_refresh_access_token(create_urlopen_mock):
+    with create_urlopen_mock(io.BytesIO(b'{"access_token":"the_token","token_type":"Bearer","expires_in":3600,"refresh_token":"the_refresh_token"}')):
+        response = refresh_access_token(client_info, "the_refresh_token")
+        assert "the_token" == response.get("access_token")
+        assert "Bearer" == response.get("token_type")
+        assert 3600 == response.get("expires_in")
+        assert "the_refresh_token" == response.get("refresh_token")
 
 def test_exchange_code_for_access_token(create_urlopen_mock):
     with create_urlopen_mock(io.BytesIO(b'{"access_token": "the_token"}')):
-        assert "the_token" == exchange_code_for_access_token(client_info, "http://localhost:123", "code")
-
+        assert "the_token" == exchange_code_for_access_token(client_info, REDIRECT_URI, "code")
 
 def test_load_oidc_config(create_urlopen_mock):
     with create_urlopen_mock(io.BytesIO(b'{"token_endpoint": "https://gitlab.com/oauth/token","authorization_endpoint": "https://gitlab.com/oauth/authorize"}')):

oauth2_cli_auth/http_server_test.py

@@ -1,3 +1,4 @@
+import random
 import threading
 import urllib
 from urllib.error import HTTPError
@@ -6,21 +7,22 @@
 
 from oauth2_cli_auth import OAuthCallbackHttpServer
 
+PORT = 49152 + random.randrange(15000)
 
 def test_http_server_ok():
-    server = OAuthCallbackHttpServer(5000)
+    server = OAuthCallbackHttpServer(PORT)
 
     threading.Thread(target=server.handle_request).start()
-    with urllib.request.urlopen("http://localhost:5000?code=foo") as response:
+    with urllib.request.urlopen(f"http://localhost:{PORT}?code=foo") as response:
         content = response.read().decode("utf-8")
         assert content is not None
 
 
 def test_http_server_bad_request():
-    server = OAuthCallbackHttpServer(5000)
+    server = OAuthCallbackHttpServer(PORT)
 
     threading.Thread(target=server.handle_request).start()
     with pytest.raises(HTTPError, match="HTTP Error 400: Bad Request") as e:
-        with urllib.request.urlopen("http://localhost:5000") as response:
+        with urllib.request.urlopen(f"http://localhost:{PORT}") as response:
             content = response.read().decode("utf-8")
             assert "Oh snap" in content

oauth2_cli_auth/simplified_flow_test.py

@@ -1,12 +1,16 @@
 import io
+import random
 import urllib
 import webbrowser
 from unittest.mock import patch
 
 from oauth2_cli_auth import get_access_token_with_browser_open, OAuth2ClientInfo
 
+PORT = 49152 + random.randrange(15000)
+
 client_info = OAuth2ClientInfo(
     client_id="dummy",
+    client_secret=None,
     authorization_url="http://auth.com/oauth/authorize",
     token_url="http://auth.com/oauth/token",
     scopes=["openid", "profile"]
@@ -19,7 +23,7 @@
 def test_get_access_token_with_browser_open(webbrowser_open, get_code, handle_request):
     with patch.object(urllib.request, 'urlopen', return_value=io.BytesIO(b'{"access_token": "the_token"}')):
         get_code.return_value = "code"
-        assert "the_token" == get_access_token_with_browser_open(client_info, 8080)
+        assert "the_token" == get_access_token_with_browser_open(client_info, PORT)
 
     assert get_code.call_count == 2
     assert handle_request.call_count == 1