Merge pull request #15 from britive/dynamic-aws-credentials-file

dynamic aws credentials file, force renew, sanitize configure tenant input

Author: twratl

requirements.txt

@@ -1,4 +1,4 @@
-https://github.com/britive/python-sdk/releases/download/v2.7.0/britive-2.7.0.tar.gz
+https://github.com/britive/python-sdk/releases/download/v2.7.1/britive-2.7.1.tar.gz
 certifi==2022.6.15
 charset-normalizer==2.1.0
 click==8.1.3

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = pybritive
-version = 0.1.6
+version = 0.2.0
 author = Britive Inc.
 author_email = support@britive.com
 description = A pure Python CLI for Britive
@@ -25,7 +25,7 @@ install_requires =
     tabulate
     toml
     cryptography
-    britive @ https://github.com/britive/python-sdk/releases/download/v2.7.0/britive-2.7.0.tar.gz#egg=britive-2.7.0
+    britive @ https://github.com/britive/python-sdk/releases/download/v2.7.1/britive-2.7.1.tar.gz#egg=britive-2.7.1
 
 [options.packages.find]
 where = src

src/pybritive/britive_cli.py

@@ -1,4 +1,6 @@
 import io
+import time
+
 from britive.britive import Britive
 from .helpers.config import ConfigManager
 from .helpers.credentials import FileCredentialManager, EncryptedFileCredentialManager
@@ -240,15 +242,17 @@ def _get_app_type(self, application_id):
                 return profile['app_type']
         raise click.ClickException(f'Application {application_id} not found')
 
-    def __get_cloud_credential_printer(self, app_type, console, mode, profile, silent, credentials):
+    def __get_cloud_credential_printer(self, app_type, console, mode, profile, silent, credentials,
+                                       aws_credentials_file):
         if app_type in ['AWS', 'AWS Standalone']:
             return printer.AwsCloudCredentialPrinter(
                 console=console,
                 mode=mode,
                 profile=profile,
                 credentials=credentials,
                 silent=silent,
-                cli=self
+                cli=self,
+                aws_credentials_file=aws_credentials_file
             )
         if app_type in ['Azure']:
             return printer.AzureCloudCredentialPrinter(
@@ -285,12 +289,34 @@ def checkin(self, profile):
             application_name=app_name
         )
 
-    def checkout(self, alias, blocktime, console, justification, mode, maxpolltime, profile, passphrase):
-        # first check if this is a profile alias
-        profile_or_alias = alias or profile
+    def _checkout(self, profile_name, env_name, app_name, programmatic, blocktime, maxpolltime, justification):
+        try:
+            self.login()
+            return self.b.my_access.checkout_by_name(
+                profile_name=profile_name,
+                environment_name=env_name,
+                application_name=app_name,
+                programmatic=programmatic,
+                include_credentials=True,
+                wait_time=blocktime,
+                max_wait_time=maxpolltime,
+                justification=justification
+            )
+        except exceptions.ApprovalRequiredButNoJustificationProvided:
+            raise click.ClickException('approval required and no justification provided.')
+        except ValueError as e:
+            raise click.BadParameter(str(e))
+
+    @staticmethod
+    def _should_check_force_renew(app, force_renew, console):
+        return app in ['AWS', 'AWS Standalone'] and force_renew and not console
 
+    def checkout(self, alias, blocktime, console, justification, mode, maxpolltime, profile, passphrase,
+                 force_renew, aws_credentials_file):
         credentials = None
         app_type = None
+        credential_process_creds_found = False
+        response = None
 
         if mode == 'awscredentialprocess':
             self.silent = True  # the aws credential process CANNOT output anything other than the expected JSON
@@ -299,59 +325,68 @@ def checkout(self, alias, blocktime, console, justification, mode, maxpolltime,
             # if not simply return those credentials
             # if they are expired
             app_type = 'AWS'  # just hardcode as we know for sure this is for AWS
-            credentials = Cache(passphrase=passphrase).get_awscredentialprocess(profile_name=profile_or_alias)
+            credentials = Cache(passphrase=passphrase).get_awscredentialprocess(profile_name=alias or profile)
             if credentials:
                 expiration_timestamp_str = credentials['expirationTime'].replace('Z', '')
                 expires = datetime.fromisoformat(expiration_timestamp_str)
                 now = datetime.utcnow()
                 if now >= expires:  # check to ensure the credentials are still valid, if not, set to None and get new
                     credentials = None
+                else:
+                    credential_process_creds_found = True
 
-        if not credentials:  # nothing found via aws credential process or not aws credential process mode
-            self.login()
-            profile = self.config.profile_aliases.get(profile, profile)
-            parts = profile.split('/')
-            if len(parts) != 3:
-                raise click.ClickException('Provided profile string does not have the required 3 parts.')
-            app_name = parts[0]
-            env_name = parts[1]
-            profile_name = parts[2]
+        profile_real = self.config.profile_aliases.get(profile, profile)
+        parts = profile_real.split('/')
+        if len(parts) != 3:
+            raise click.ClickException('Provided profile string does not have the required 3 parts.')
 
-            try:
-                response = self.b.my_access.checkout_by_name(
-                    profile_name=profile_name,
-                    environment_name=env_name,
-                    application_name=app_name,
-                    programmatic=False if console else True,
-                    include_credentials=True,
-                    wait_time=blocktime,
-                    max_wait_time=maxpolltime,
-                    justification=justification
-                )
+        # create this params once so we can use it multiple places
+        params = {
+            'profile_name': parts[2],
+            'env_name': parts[1],
+            'app_name': parts[0],
+            'programmatic': False if console else True,
+            'blocktime': blocktime,
+            'maxpolltime': maxpolltime,
+            'justification': justification
+        }
+
+        if not credential_process_creds_found:  # nothing found via aws cred process or not aws cred process mode
+            response = self._checkout(**params)
+            app_type = self._get_app_type(response['appContainerId'])
+            credentials = response['credentials']
+
+        # this handles the --force-renew flag
+        # lets check to see if the we should checkin this profile first and check it out again
+        if self._should_check_force_renew(app_type, force_renew, console):
+            expiration = datetime.fromisoformat(credentials['expirationTime'].replace('Z', ''))
+            now = datetime.utcnow()
+            diff = (expiration - now).total_seconds() / 60.0
+            if diff < force_renew:  # time to checkin the profile so we can refresh creds
+                self.print('checking in the profile to get renewed credentials....standby')
+                self.checkin(profile=profile_real)
+                response = self._checkout(**params)
+                credential_process_creds_found = False  # need to write new creds to cache
                 credentials = response['credentials']
-                app_type = self._get_app_type(response['appContainerId'])
-            except exceptions.ApprovalRequiredButNoJustificationProvided:
-                raise click.ClickException('approval required and no justification provided.')
-            except ValueError as e:
-                raise click.BadParameter(str(e))
-
-            if alias:  # do this down here so we know that the profile is valid and a checkout was successful
-                self.config.save_profile_alias(alias=alias, profile=profile)
-            if mode == 'awscredentialprocess':
-                Cache(passphrase=passphrase).save_awscredentialprocess(
-                    profile_name=profile_or_alias,
-                    credentials=credentials
-                )
 
-        cc_printer = self.__get_cloud_credential_printer(
+        if alias:  # do this down here so we know that the profile is valid and a checkout was successful
+            self.config.save_profile_alias(alias=alias, profile=profile)
+
+        if mode == 'awscredentialprocess' and not credential_process_creds_found:
+            Cache(passphrase=passphrase).save_awscredentialprocess(
+                profile_name=alias or profile,
+                credentials=credentials
+            )
+
+        self.__get_cloud_credential_printer(
             app_type,
             console,
             mode,
-            profile_or_alias,
+            alias or profile,
             self.silent,
-            credentials
-        )
-        cc_printer.print()
+            credentials,
+            aws_credentials_file
+        ).print()
 
     def import_existing_npm_config(self):
         profile_aliases = self.config.import_global_npm_config()

src/pybritive/commands/checkout.py

@@ -6,10 +6,11 @@
 
 @click.command()
 @build_britive
-@britive_options(names='alias,blocktime,console,justification,mode,maxpolltime,silent,tenant,token,passphrase')
+@britive_options(names='alias,blocktime,console,justification,mode,maxpolltime,silent,force_renew,aws_credentials_file,'
+                       'tenant,token,passphrase')
 @click.argument('profile', shell_complete=profile_completer)
-def checkout(ctx, alias, blocktime, console, justification, mode, maxpolltime, silent, tenant, token,
-             passphrase, profile):
+def checkout(ctx, alias, blocktime, console, justification, mode, maxpolltime, silent, force_renew,
+             aws_credentials_file, tenant, token, passphrase, profile):
     """Checkout a profile.
 
     This command takes 1 required argument `PROFILE`. This should be a string representation of the profile
@@ -25,5 +26,7 @@ def checkout(ctx, alias, blocktime, console, justification, mode, maxpolltime, s
         mode=mode,
         maxpolltime=maxpolltime,
         profile=profile,
-        passphrase=passphrase
+        passphrase=passphrase,
+        force_renew=force_renew,
+        aws_credentials_file=aws_credentials_file
     )

src/pybritive/helpers/cloud_credential_printer.py

@@ -83,8 +83,9 @@ def _not_implemented(self):
 
 
 class AwsCloudCredentialPrinter(CloudCredentialPrinter):
-    def __init__(self, console, mode, profile, silent, credentials, cli):
+    def __init__(self, console, mode, profile, silent, credentials, cli, aws_credentials_file):
         super().__init__('AWS', console, mode, profile, silent, credentials, cli)
+        self.aws_credentials_file = aws_credentials_file
 
     def print_text(self):
         self.cli.print('AWS_ACCESS_KEY_ID', ignore_silent=True)
@@ -123,7 +124,7 @@ def print_env(self):
 
     def print_integrate(self):
         # get path to aws credentials file
-        env_path = os.getenv('AWS_SHARED_CREDENTIALS_FILE')
+        env_path = self.aws_credentials_file
         if not env_path:
             path = Path.home() / '.aws' / 'credentials'  # handle os specific separators properly
         else:

src/pybritive/helpers/config.py

@@ -141,7 +141,10 @@ def save(self):
 
     def save_tenant(self, tenant: str, alias: str = None, output_format: str = None):
         self.load()
-        if not alias:
+        tenant = tenant.replace('.britive-app.com', '')
+        if alias:
+            alias = alias.replace('.britive-app.com', '')
+        else:
             alias = tenant
         if f'tenant-{alias}' not in self.config.keys():
             self.config[f'tenant-{alias}'] = {}
@@ -157,7 +160,7 @@ def save_global(self, default_tenant_name: str = None, output_format: str = None
         if 'global' not in self.config.keys():
             self.config['global'] = {}
         if default_tenant_name:
-            self.config['global']['default_tenant'] = default_tenant_name
+            self.config['global']['default_tenant'] = default_tenant_name.replace('.britive-app.com', '')
         if output_format:
             self.config['global']['output_format'] = output_format
         if backend:

src/pybritive/options/aws_credentials_file.py

@@ -0,0 +1,12 @@
+import click
+
+
+option = click.option(
+    '--aws-credentials-file',
+    default=None,
+    help='AWS Programmatic Only - When mode is `integrate` specify a non-default location for the AWS '
+         'credentials file.',
+    envvar='AWS_SHARED_CREDENTIALS_FILE',
+    show_envvar=True,
+    show_default=True
+)

src/pybritive/options/britive_options.py

@@ -17,6 +17,8 @@
 from ..options.file import option as file
 from ..options.configure_backend import option as configure_backend
 from ..options.passphrase import option as passphrase
+from ..options.force_renew import option as force_renew
+from ..options.aws_credentials_file import option as aws_credentials_file
 
 options_map = {
     'tenant': tenant,
@@ -37,7 +39,9 @@
     'checked_out': checked_out,
     'file': file,
     'configure_backend': configure_backend,
-    'passphrase': passphrase
+    'passphrase': passphrase,
+    'force_renew': force_renew,
+    'aws_credentials_file': aws_credentials_file
 }
 
 

src/pybritive/options/force_renew.py

@@ -0,0 +1,11 @@
+import click
+
+option = click.option(
+    '--force-renew',
+    default=None,
+    show_default=True,
+    type=int,
+    help='AWS Programmatic Only - If the credentials are to expire within the specified number of minutes, check in '
+         'the profile first and check it out again to get a new set of credentials.'
+)
+