rename the repo to aws-fusion to be more generic to future usecase

.github/workflows/tagging.yml

@@ -5,7 +5,7 @@ on:
     branches:
       - "main"
     paths:
-      - "aws_console/**"
+      - "aws_fusion/**"
       - "setup.py"
 
 concurrency: tagging

README.md

@@ -1,40 +1,38 @@
-# aws console
-AWS Console Login Utility
+# aws fusion
+Unified CLI tool for streamlined AWS operations, enhancing developer productivity
 
 [![Tag][tag-badge]][tag]
 [![Tagging][actions-workflow-tagging-badge]][actions-workflow-tagging]
 
 ## Command line tool
-- `aws-console`
+- `aws-fusion`
 - `aws-credential-process-from-system`
 
 Additonally this creates [aws cli alias](https://docs.aws.amazon.com/cli/latest/userguide/cli-usage-alias.html) for all the tools
 - `aws console`
 - `aws credential-process-from-system`
 
 ## Installation
-### Via Pip
+### Via Pip directly
 Install via pip install.
 _note this also requires git to be present_
 
 ```shell
-pip install git+https://github.com/snigdhasjg/aws-console.git@main
+pip install git+https://github.com/snigdhasjg/aws-fusion.git@main
 ```
 
 ### Manually
-1. Simply clone this repository
+Simply clone this repository and run pip install
 ```shell
-git clone https://github.com/snigdhasjg/aws-console.git
-```
-2. Install using [setup.py](./setup.py)
-```shell
-python setup.py install
+git clone https://github.com/snigdhasjg/aws-fusion.git
+cd aws-fusion
+pip install .
 ```
 
 ---
-## Usage of `aws-console`
+## Usage of `aws-fusion`
  - Make AWS credentials available via aws profile
- - Execute the script: `aws-console --profile my-profile`
+ - Execute the script: `aws-fusion --profile my-profile`
  - :tada: Your browser opens, and you are signed in into the AWS console
 
 ### Use cases
@@ -87,7 +85,7 @@ sso_start_url = https://my-sso-portal.awsapps.com/start
 sso_registration_scopes = sso:account:access
 ```
 
-> Try `aws-console --help` for detailed parameter
+> Try `aws-fusion --help` for detailed parameter
 
 ### Refer
 The docs
@@ -133,8 +131,8 @@ This project is licensed under the MIT License - see the [LICENSE](./LICENSE) fi
 
 <!-- badge links -->
 
-[tag]: https://github.com/snigdhasjg/aws-console/tags
-[tag-badge]: https://img.shields.io/github/v/tag/snigdhasjg/aws-console?style=for-the-badge&logo=github
+[tag]: https://github.com/snigdhasjg/aws-fusion/tags
+[tag-badge]: https://img.shields.io/github/v/tag/snigdhasjg/aws-fusion?style=for-the-badge&logo=github
 
-[actions-workflow-tagging]: https://github.com/snigdhasjg/aws-console/actions/workflows/tagging.yml
-[actions-workflow-tagging-badge]: https://img.shields.io/github/actions/workflow/status/snigdhasjg/aws-console/tagging.yml?branch=main&label=Tagging&style=for-the-badge&logo=githubactions
+[actions-workflow-tagging]: https://github.com/snigdhasjg/aws-fusion/actions/workflows/tagging.yml
+[actions-workflow-tagging-badge]: https://img.shields.io/github/actions/workflow/status/snigdhasjg/aws-fusion/tagging.yml?branch=main&label=Tagging&style=for-the-badge&logo=githubactions

aws_fusion/__init__.py

@@ -0,0 +1 @@
+__version__ = '0.5'

aws_fusion/app.py

@@ -0,0 +1,25 @@
+import os
+import argparse
+
+from importlib.metadata import version
+
+from .commands import open_browser, iam_user_credentials, generate_okta_device_auth_credentials
+
+def main():
+    global_parser = argparse.ArgumentParser(add_help=False)
+    global_parser.add_argument('-v', '--version', action='version', help="Display the version of this tool", version=version("aws_fusion"))
+
+    main_parser = argparse.ArgumentParser(prog='aws-fusion' ,description='Unified CLI tool for streamlined AWS operations, enhancing developer productivity', parents=[global_parser])
+    subparsers = main_parser.add_subparsers(dest='command', required=True, help='Available commands')
+
+    open_browser.setup(subparsers, global_parser)
+    iam_user_credentials.setup(subparsers, global_parser)
+    generate_okta_device_auth_credentials.setup(subparsers, global_parser)
+
+    args = main_parser.parse_args()
+    # Call the associated function for the selected sub-command
+    if hasattr(args, 'func'):
+        args.func(args)
+
+if __name__ == '__main__':
+    main()

aws_fusion/aws/assume_role.py

@@ -0,0 +1,73 @@
+import sys
+import boto3
+import hashlib
+import datetime
+import logging
+from botocore.exceptions import ClientError
+from botocore.utils import JSONFileCache
+
+import json
+
+LOG = logging.getLogger(__name__)
+
+class AssumeRoleWithSamlCache():
+    __jsonFileCache = JSONFileCache()
+
+    def __init__(self, role) -> None:
+        LOG.debug('Initialize AssumeRoleWithSamlCache')
+        self.__role = role
+        self.__cache_key = hashlib.sha1(role.encode('utf-8')).hexdigest()
+        self.__response = None
+
+    def does_valid_token_cache_exists(self):
+        if self.__cache_key in self.__jsonFileCache:
+            response = self.__jsonFileCache[self.__cache_key]
+            expiration = datetime.datetime.strptime(response['Credentials']['Expiration'], '%Y-%m-%dT%H:%M:%S%Z')
+            current_utc_time = datetime.datetime.utcnow()
+            if expiration - current_utc_time >= datetime.timedelta(minutes=1):
+                response['Credentials']['Expiration'] = expiration.replace(tzinfo=datetime.timezone.utc)
+                self.__response = response
+                LOG.debug('Valid token exists. Can use cache')
+                return True
+        LOG.debug('No valid token exists')
+        return False
+
+    def credential_process(self):
+        credentials = self.__response['Credentials']
+        LOG.debug(f'Giving credential as aws credential process format. The credential: {credentials}')
+
+        # https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-sourcing-external.html
+        return json.dumps({
+            "Version": 1,
+            "AccessKeyId": credentials['AccessKeyId'],
+            "SecretAccessKey": credentials['SecretAccessKey'],
+            "SessionToken": credentials['SessionToken'],
+            "Expiration": credentials['Expiration'].strftime('%Y-%m-%dT%H:%M:%S%Z')
+        })
+
+    def assume_role_with_saml(self, saml_response, roles, sessoion_duration):
+        LOG.debug(f'Started assumning role with SAML')
+        client = boto3.client('sts')
+        selected_role = self.__role
+        try:
+            response = client.assume_role_with_saml(
+                RoleArn=selected_role,
+                PrincipalArn=roles[selected_role],
+                SAMLAssertion=saml_response,
+                DurationSeconds=sessoion_duration
+            )
+            LOG.debug('Got assume role response')
+        except ClientError:
+            # Try again with a shorter session length
+            response = client.assume_role_with_saml(
+                RoleArn=selected_role,
+                PrincipalArn=roles[selected_role],
+                SAMLAssertion=saml_response,
+                DurationSeconds=3600
+            )
+            LOG.debug('Got assume role fallback response')
+
+        self.__jsonFileCache[self.__cache_key] = response
+
+        self.__response = response
+

aws_console/aws_session_credential.py → aws_fusion/aws/session.py

@@ -1,17 +1,13 @@
 from botocore.utils import JSONFileCache
 import boto3
 
-import datetime
-import os
-import json
-
 
 class TokenGenerationException(Exception):
     "Exception for credential not having token"
     pass
 
 
-def aws_session_credential(profile_name, region_name):
+def credentials(profile_name, region_name):
     session = boto3.Session(profile_name=profile_name, region_name=region_name)
 
     __update_credential_provider_cache(session)

aws_fusion/commands/generate_okta_device_auth_credentials.py

@@ -0,0 +1,34 @@
+import os
+import logging
+
+from ..aws.assume_role import AssumeRoleWithSamlCache
+from ..okta.api import device_auth, verifiction_and_token, session_and_token, saml_assertion
+
+LOG = logging.getLogger(__name__)
+
+def setup(subparsers, parent_parser):
+    summary = 'Generate AWS session credentials using SAML assertion from Okta device authentication'
+    parser = subparsers.add_parser('generate-okta-device-auth-credentials', description=summary, help=summary, parents=[parent_parser])
+
+    parser.add_argument('--org-domain', required=True, help="Full domain hostname of the Okta org e.g. example.okta.com")
+    parser.add_argument('--oidc-client-id', required=True, help="The ID is the identifier of the client is Okta app acting as the IdP for AWS")
+    parser.add_argument('--aws-acct-fed-app-id', required=True, help="ID for the AWS Account Federation integration app")
+    parser.add_argument('--aws-iam-role', required=True, help="The AWS IAM Role ARN to assume")
+    parser.add_argument('--credential-process', action='store_true', help='Output the credential in AWS credential process syntax')
+
+    parser.set_defaults(func=run)
+
+def run(args):
+    assume_role_with_cache = AssumeRoleWithSamlCache(args.aws_iam_role)
+
+    if not assume_role_with_cache.does_valid_token_cache_exists():
+        LOG.debug('Credential cache not found, invloking SAML')
+        device_code = device_auth(args.org_domain, args.oidc_client_id)
+        access_token, id_token = verifiction_and_token(args.org_domain, args.oidc_client_id, device_code)
+        session_token = session_and_token(args.org_domain, args.oidc_client_id, access_token, id_token, args.aws_acct_fed_app_id)
+        saml_response, roles, sessoion_duration = saml_assertion(args.org_domain, session_token)
+        assume_role_with_cache.assume_role_with_saml(saml_response, roles, sessoion_duration)
+
+    if args.credential_process:
+        print(assume_role_with_cache.credential_process())
+