OIE authorization code flow implementation with forced classic authen…

…tication.

Chained Oktas that are a mix of OIE and classic has issues, SAML2 does not return an idx cookies.

.gitignore

@@ -91,7 +91,7 @@ venv/
 ENV/
 env.bak/
 venv.bak/
-.vscode
+.vscode/
 
 # Spyder project settings
 .spyderproject

README.md

@@ -24,6 +24,13 @@ tokens into your local `~/.aws/credentials` file.
 
 ## What's new
 See [Releases](https://github.com/dowjones/tokendito/releases) for a detailed Changelog.
+### Tokendito 2.3.0
+Version 2.3.0 of Tokendito introduces the following new features:
+- Basic OIE support while forcing Classic mode.    
+- Misc bug fixes
+
+While this feature works for local OIE enabled organization, it does not for chained OIE-Classic Organizations. Future bugfixes should hopefully address that.
+
 ### Tokendito 2.2.0
 
 Version 2.2.0 of Tokendito introduces the following new features:

tests/unit/test_okta.py

@@ -470,11 +470,10 @@ def test_send_saml_request(mocker):
     )
 
     saml_request = {"relay_state": "relay_state", "request": "request", "post_url": "post_url"}
-    cookie = {"sid": "pytestcookie"}
 
     mocker.patch("tokendito.http_client.HTTP_client.get", return_value=mock_response)
 
-    assert okta.send_saml_request(saml_request, cookie) == {
+    assert okta.send_saml_request(saml_request) == {
         "response": "PD94bWwgdmVyc2lvbj0iMS4wIiBlbmNvZGluZz0iVVRGLTgiPz4=",
         "relay_state": "foobar",
         "post_url": "https://acme.okta.com/app/okta_org2org/akjlkjlksjx0xmdd/sso/saml",
@@ -484,6 +483,7 @@ def test_send_saml_request(mocker):
 def test_send_saml_response(mocker):
     """Test sending SAML response."""
     from tokendito import okta
+    from tokendito.config import Config
     from tokendito.http_client import HTTP_client
 
     mock_response = Mock()
@@ -495,12 +495,16 @@ def test_send_saml_response(mocker):
         "post_url": "https://acme.okta.com/app/okta_org2org/akjlkjlksjx0xmdd/sso/saml",
     }
 
+    mocker.patch("tokendito.okta.extract_state_token", return_value=None)
+
     mocker.patch.object(HTTP_client, "post", return_value=mock_response)
 
-    assert okta.send_saml_response(saml_response) == mock_response.cookies
+    pytest_config = Config()
+
+    assert okta.send_saml_response(pytest_config, saml_response) == mock_response.cookies
 
 
-def test_authenticate(mocker):
+def test_idp_auth(mocker):
     """Test authentication."""
     from tokendito import okta
     from tokendito.config import Config
@@ -513,23 +517,23 @@ def test_authenticate(mocker):
         }
     )
     sid = {"sid": "pytestsid"}
-    mocker.patch("tokendito.user.request_cookies", return_value=sid)
-    mocker.patch("tokendito.okta.local_auth", return_value="foobar")
-    mocker.patch("tokendito.okta.saml2_auth", return_value=sid)
+    mocker.patch("tokendito.okta.create_authn_cookies", return_value=sid)
+    mocker.patch("tokendito.okta.local_authenticate", return_value="foobar")
+    mocker.patch("tokendito.okta.saml2_authenticate", return_value=sid)
 
     mocker.patch("tokendito.okta.get_auth_properties", return_value={"type": "OKTA"})
-    assert okta.authenticate(pytest_config) == sid
+    assert okta.idp_auth(pytest_config) == sid
 
     mocker.patch("tokendito.okta.get_auth_properties", return_value={"type": "SAML2"})
-    assert okta.authenticate(pytest_config) == sid
+    assert okta.idp_auth(pytest_config) == sid
 
     mocker.patch("tokendito.okta.get_auth_properties", return_value={"type": "UNKNOWN"})
     with pytest.raises(SystemExit) as error:
-        assert okta.authenticate(pytest_config) == error
+        assert okta.idp_auth(pytest_config) == error
 
     mocker.patch("tokendito.okta.get_auth_properties", return_value={})
     with pytest.raises(SystemExit) as error:
-        assert okta.authenticate(pytest_config) == error
+        assert okta.idp_auth(pytest_config) == error
 
 
 def test_step_up_authenticate(mocker):
@@ -605,7 +609,7 @@ def test_local_auth(mocker):
     assert okta.local_auth(pytest_config) == "pytesttoken"
 
 
-def test_saml2_auth(mocker):
+def test_saml2_authenticate(mocker):
     """Test saml2 authentication."""
     from tokendito import okta
     from tokendito.config import Config
@@ -623,12 +627,12 @@ def test_saml2_auth(mocker):
         "base_url": "https://acme.okta.com",
     }
     mocker.patch("tokendito.okta.get_saml_request", return_value=saml_request)
-    mocker.patch("tokendito.okta.authenticate", return_value="pytestsessioncookie")
+    mocker.patch("tokendito.okta.idp_auth", return_value="pytestsessioncookie")
 
     saml_response = {
         "response": "pytestresponse",
     }
 
     mocker.patch("tokendito.okta.send_saml_request", return_value=saml_response)
     mocker.patch("tokendito.okta.send_saml_response", return_value="pytestsessionid")
-    assert okta.saml2_auth(pytest_config, auth_properties) == "pytestsessionid"
+    assert okta.saml2_authenticate(pytest_config, auth_properties) == "pytestsessionid"

tokendito/__init__.py

@@ -1,7 +1,7 @@
 # vim: set filetype=python ts=4 sw=4
 # -*- coding: utf-8 -*-
 """Tokendito module initialization."""
-__version__ = "2.2.0"
+__version__ = "2.3.0"
 __title__ = "tokendito"
 __description__ = "Get AWS STS tokens from Okta SSO"
 __long_description_content_type__ = "text/markdown"

tokendito/__main__.py

@@ -12,10 +12,10 @@ def main(args=None):  # needed for console script
 
         path = os.path.dirname(os.path.dirname(__file__))
         sys.path[0:0] = [path]
-    from tokendito.tool import cli
+    from tokendito.user import cmd_interface
 
     try:
-        return cli(args)
+        return cmd_interface(args)
     except KeyboardInterrupt:
         print("\nInterrupted")
         sys.exit(1)

tokendito/aws.py

@@ -47,30 +47,23 @@ def get_output_types():
     return ["json", "text", "csv", "yaml", "yaml-stream"]
 
 
-def authenticate_to_roles(config, urls, cookies=None):
+def authenticate_to_roles(config, urls, cookies):
     """Authenticate AWS user with saml.
 
     :param urls: list of tuples or tuple, with tiles info
     :param cookies: html cookies
-    :param user_agent: optional user agent string
     :return: response text
 
     """
-    if cookies:
-        HTTP_client.set_cookies(cookies)  # Set cookies if provided
-
     url_list = [urls] if isinstance(urls, tuple) else urls
     responses = []
     tile_count = len(url_list)
     plural = "s" if tile_count > 1 else ""
 
     logger.info(f"Discovering roles in {tile_count} tile{plural}.")
     for url, label in url_list:
-        response = HTTP_client.get(url)  # Use the HTTPClient's get method
-
         session_url = config.okta["org"] + "/login/sessionCookieRedirect"
         params = {"token": cookies.get("sessionToken"), "redirectUrl": url}
-
         response = HTTP_client.get(session_url, params=params)
 
         saml_response_string = response.text

tokendito/config.py

@@ -46,6 +46,7 @@ class Config(object):
             password="",
             mfa=None,
             mfa_response=None,
+            client_id=None,
             tile=None,
             org=None,
             device_token=None,

tokendito/http_client.py

@@ -26,13 +26,16 @@ def set_cookies(self, cookies):
         """Update session with additional cookies."""
         self.session.cookies.update(cookies)
 
-    def get(self, url, params=None, headers=None):
+    def get(self, url, params=None, headers=None, allow_redirects=True):
         """Perform a GET request."""
         response = None
         try:
+            logger.debug(f"GET to {url}")
             logger.debug(f"Sending cookies: {self.session.cookies}")
             logger.debug(f"Sending headers: {self.session.headers}")
-            response = self.session.get(url, params=params, headers=headers)
+            response = self.session.get(
+                url, params=params, headers=headers, allow_redirects=allow_redirects
+            )
             response.raise_for_status()
             logger.debug(f"Received response from {url}: {response.text}")
             return response
@@ -51,6 +54,7 @@ def get(self, url, params=None, headers=None):
 
     def post(self, url, data=None, json=None, headers=None, params=None, return_json=False):
         """Perform a POST request."""
+        logger.debug(f"POST to {url}")
         try:
             response = self.session.post(url, data=data, json=json, params=params, headers=headers)
             response.raise_for_status()