Update Python version requirement and refactor common functions

- Updated the Python version requirement in README.md from 3.7+ to 3.10+.
- Refactored common functions in common.py to improve code organization and reusability, including user agent and request header generation.
- Enhanced the handling of SAML responses and role attributes across multiple files, improving code clarity and maintainability.
- Updated various classes to utilize the new common functions for consistency in HTTP requests and SAML processing.

Author: epierce

README.md

@@ -16,7 +16,7 @@ Okta is a registered trademark of Okta, Inc. and this tool has no affiliation wi
 
 [Okta SAML integration to AWS using the AWS App](https://help.okta.com/en/prod/Content/Topics/Miscellaneous/References/OktaAWSMulti-AccountConfigurationGuide.pdf)
 
-Python 3.7+
+Python 3.10+
 
 ####  A Note on Python 3.10+ Compatibility on Windows
 

docs/index.md

@@ -7,7 +7,8 @@
 **Type:** CLI (Python command-line tool)  
 **Language:** Python 3.7+  
 **Architecture:** Modular authentication system with pluggable MFA providers  
-**Version:** 2.8.2
+**Released Version:** 2.8.2
+**Latest Version:** 2.9.0-pre
 
 ### Quick Reference
 

docs/project-overview.md

@@ -9,7 +9,8 @@
 | Property | Value |
 |----------|-------|
 | **Name** | gimme-aws-creds |
-| **Version** | 2.8.2 |
+| **Released Version** | 2.8.2 |
+| **Latest Version** | 2.9.0-pre |
 | **License** | Apache License 2.0 |
 | **Python** | 3.7+ |
 | **Maintainer** | Eric Pierce |

gimme_aws_creds/__init__.py

@@ -1,2 +1,2 @@
 __all__ = ['alibaba_cloud', 'config', 'aws', 'main', 'ui', 'common', 'default', 'duo', 'errors', 'okta_classic', 'okta_identity_engine', 'registered_authenticators', 'u2f', 'webauthn']
-version = '2.8.2-pre'
+version = '2.9.0-pre'

gimme_aws_creds/alibaba_cloud.py

@@ -3,16 +3,13 @@
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 """
-import base64
 import re
-import sys
-import platform
-import xml.etree.ElementTree as ET
 from collections import namedtuple
 from urllib.parse import quote
 
 import requests
-from bs4 import BeautifulSoup
+
+from .common import user_agent, request_headers_json, parse_saml_form, parse_saml_role_attributes, okta_token_exchange
 
 try:
     from alibabacloud_sts20150401 import client as _sts_client
@@ -23,7 +20,7 @@
 except ImportError:
     ALIBABA_CLOUD_SDK_AVAILABLE = False
 
-from . import errors, version
+from . import errors
 
 ALIBABA_CLOUD_SAML_ROLE_ATTRIBUTE = 'https://www.aliyun.com/SAML-Role/Attributes/Role'
 
@@ -35,23 +32,13 @@
 )
 
 
-def _user_agent():
-    return "gimme-aws-creds {};{};{}".format(version, sys.platform, platform.python_version())
-
-
-def _request_headers_json():
-    return {
-        'User-Agent': _user_agent(),
-        'Accept': 'application/json',
-    }
-
 def _account_id_from_role_arn(role_arn):
     m = re.match(r'acs:ram::(\d+):', role_arn)
     if not m:
         return ''
     return m.group(1)
 
-class AlibabaCloudClient(object):
+class AlibabaCloudClient:
     """Alibaba Cloud RAM credentials via Okta Native-to-Web SSO (interclient token) and STS AssumeRoleWithSAML."""
 
     HTTP_TIMEOUT = 30
@@ -71,36 +58,12 @@ def __init__(self, http_client, okta_org_url, client_id, verify_ssl_certs=True):
         self._verify_ssl_certs = verify_ssl_certs
 
     def _interclient_token_exchange(self, app_id, access_token, id_token):
-        response = self._http_client.post(
-            self._okta_org_url + '/oauth2/v1/token',
-            headers=_request_headers_json(),
-            data={
-                'actor_token': access_token,
-                'actor_token_type': 'urn:ietf:params:oauth:token-type:access_token',
-                'client_id': self._client_id,
-                'audience': 'urn:okta:apps:{}'.format(app_id),
-                'grant_type': 'urn:ietf:params:oauth:grant-type:token-exchange',
-                'requested_token_type': 'urn:okta:params:oauth:token-type:interclient_token',
-                'subject_token': id_token,
-                'subject_token_type': 'urn:ietf:params:oauth:token-type:id_token',
-            },
-            verify=self._verify_ssl_certs,
-            timeout=self.HTTP_TIMEOUT,
+        return okta_token_exchange(
+            self._http_client, self._okta_org_url, self._client_id,
+            app_id, access_token, id_token,
+            requested_token_type='urn:okta:params:oauth:token-type:interclient_token',
+            verify_ssl=self._verify_ssl_certs, timeout=self.HTTP_TIMEOUT,
         )
-        try:
-            response_data = response.json()
-        except ValueError as e:
-            raise errors.GimmeAWSCredsError(
-                'Invalid JSON response from token exchange endpoint: {}'.format(str(e)), 2)
-
-        if response.status_code == 200:
-            return response_data
-        if response.status_code == 400:
-            raise errors.GimmeAWSCredsError(
-                'LOGIN ERROR: Interclient token exchange failed: {}'.format(
-                    response_data.get('error_description', 'Unknown error')), 2)
-        response.raise_for_status()
-        return None
 
     @staticmethod
     def _saml_app_fetch_url(saml_app_url, interclient_token):
@@ -117,31 +80,22 @@ def get_saml_response(self, saml_sso_url, saml_app_url, auth_session):
         fetch_url = self._saml_app_fetch_url(saml_sso_url, interclient_token)
         response = self._http_client.get(
             fetch_url,
-            headers=_request_headers_json(),
+            headers=request_headers_json(),
             verify=self._verify_ssl_certs,
             timeout=self.HTTP_TIMEOUT,
         )
 
         if response.status_code != 200:
             response.raise_for_status()
 
-        saml_response = None
-        relay_state = None
-        form_action = None
-
-        saml_soup = BeautifulSoup(response.text, 'html.parser')
-        if saml_soup.find('form') is not None:
-            form_action = saml_soup.find('form').get('action')
-        for input_tag in saml_soup.find_all('input'):
-            if input_tag.get('name') == 'SAMLResponse':
-                saml_response = input_tag.get('value')
-            elif input_tag.get('name') == 'RelayState':
-                relay_state = input_tag.get('value')
+        saml_response, relay_state, form_action = parse_saml_form(response.text)
 
         if saml_response is None:
             saml_error = 'Did not receive SAML Response after successful authentication [{}]'.format(saml_app_url)
-            if saml_soup.find(class_='error-content') is not None:
-                saml_error += '\n' + saml_soup.find(class_='error-content').get_text()
+            from bs4 import BeautifulSoup
+            error_soup = BeautifulSoup(response.text, 'html.parser')
+            if error_soup.find(class_='error-content') is not None:
+                saml_error += '\n' + error_soup.find(class_='error-content').get_text()
             raise errors.GimmeAWSCredsError(saml_error, 2)
 
         return {'SAMLResponse': saml_response, 'RelayState': relay_state, 'TargetUrl': form_action}
@@ -153,25 +107,18 @@ def enumerate_saml_roles(assertion_b64):
 
     @staticmethod
     def _enumerate_saml_roles_impl(assertion_b64):
-        root = ET.fromstring(base64.b64decode(assertion_b64))
         roles = []
-        for attr in root.iter('{urn:oasis:names:tc:SAML:2.0:assertion}Attribute'):
-            if attr.get('Name') != ALIBABA_CLOUD_SAML_ROLE_ATTRIBUTE:
-                continue
-            for val in attr.iter('{urn:oasis:names:tc:SAML:2.0:assertion}AttributeValue'):
-                text = (val.text or '').strip()
-                if not text:
-                    continue
-                parts = [p.strip() for p in text.split(',')]
-                if len(parts) != 2:
-                    raise errors.GimmeAWSCredsError(
-                        'Invalid Alibaba Cloud role pair (expected role_arn,saml_provider_arn): {}'.format(text), 2)
-                role_arn, saml_provider_arn = parts[0], parts[1]
-                roles.append(AlibabaCloudRoleSet(
-                    role_arn=role_arn,
-                    saml_provider_arn=saml_provider_arn,
-                    account_id=_account_id_from_role_arn(role_arn),
-                ))
+        for text in parse_saml_role_attributes(assertion_b64, ALIBABA_CLOUD_SAML_ROLE_ATTRIBUTE):
+            parts = [p.strip() for p in text.split(',')]
+            if len(parts) != 2:
+                raise errors.GimmeAWSCredsError(
+                    'Invalid Alibaba Cloud role pair (expected role_arn,saml_provider_arn): {}'.format(text), 2)
+            role_arn, saml_provider_arn = parts[0], parts[1]
+            roles.append(AlibabaCloudRoleSet(
+                role_arn=role_arn,
+                saml_provider_arn=saml_provider_arn,
+                account_id=_account_id_from_role_arn(role_arn),
+            ))
         return roles
 
     ASSUME_ROLE_MAX_DURATION = 3600

gimme_aws_creds/aws.py

@@ -11,19 +11,15 @@
 """
 import base64
 import json
-import xml.etree.ElementTree as ET
 
-import urllib3
-import requests
 from bs4 import BeautifulSoup
-from requests.adapters import HTTPAdapter
-from urllib3.util.retry import Retry
 
 import gimme_aws_creds.common as commondef
+from .common import parse_saml_role_attributes, create_http_session
 from . import errors
 
 
-class AwsResolver(object):
+class AwsResolver:
     """
        The Aws Client Class performes post request on AWS sign-in page
        to fetch friendly names/alias for account and IAM roles
@@ -34,15 +30,7 @@ def __init__(self, verify_ssl_certs=True):
         :param verify_ssl_certs: Enable/disable SSL verification
         """
         self._verify_ssl_certs = verify_ssl_certs
-
-        if verify_ssl_certs is False:
-            urllib3.disable_warnings()
-
-        # Allow up to 5 retries on requests to AWS in case we have network issues
-        self._http_client = requests.Session()
-        retries = Retry(total=5, backoff_factor=1,
-                        allowed_methods=['POST'])
-        self._http_client.mount('https://', HTTPAdapter(max_retries=retries))
+        self._http_client = create_http_session(verify_ssl=verify_ssl_certs, allowed_methods=['POST'])
 
     def get_signinpage(self, saml_token, saml_target_url):
         """ Post SAML token to aws sign in page and get back html result"""
@@ -62,12 +50,7 @@ def _enumerate_saml_roles(self, assertion, saml_target_url):
         signin_page = self.get_signinpage(assertion, saml_target_url)
 
         """ using the assertion to fetch aws sign-in page, parse it and return aws sts creds """
-        role_pairs = []
-        root = ET.fromstring(base64.b64decode(assertion))
-        for saml2_attribute in root.iter('{urn:oasis:names:tc:SAML:2.0:assertion}Attribute'):
-            if saml2_attribute.get('Name') == 'https://aws.amazon.com/SAML/Attributes/Role':
-                for saml2_attribute_value in saml2_attribute.iter('{urn:oasis:names:tc:SAML:2.0:assertion}AttributeValue'):
-                    role_pairs.append(saml2_attribute_value.text)
+        role_pairs = list(parse_saml_role_attributes(assertion, 'https://aws.amazon.com/SAML/Attributes/Role'))
 
         # build a temp hash table
         table = {}