Merge pull request #134 from britive/1.7.0rc1

v1.7.0rc1 develop

Author: twratl

CHANGELOG.md

@@ -2,6 +2,22 @@
 
 * As of v1.4.0 release candidates will be published in an effort to get new features out faster while still allowing time for full QA testing before moving the release candidate to a full release.
 
+## v1.7.0rc1 [2024-01-19]
+#### What's New
+* Display system announcement/banner if one is present for the tenant
+
+#### Enhancements
+* New checkout mode of `gcloudauthexec` which will invoke, via sub-shell, the `gcloud auth activate-service-account` command to switch credentials for `gcloud`. Additionally, a `checkin` will reset this configuration.
+
+#### Bug Fixes
+* Fix issue related to the `cache` and `clear` commands when no global default tenant is set
+
+#### Dependencies
+* `britive>=2.24.0rc1`
+
+#### Other
+* None
+
 ## v1.6.1 [2023-12-18]
 #### What's New
 * None

requirements.txt

@@ -1,5 +1,5 @@
 boto3
-britive>=2.23.0
+britive>=2.24.0rc1
 certifi>=2022.12.7
 charset-normalizer==2.1.0
 click~=8.1.3

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = pybritive
-version = 1.6.1
+version = 1.7.0rc1
 author = Britive Inc.
 author_email = support@britive.com
 description = A pure Python CLI for Britive
@@ -27,7 +27,7 @@ install_requires =
     toml
     cryptography>=41.0.0
     python-dateutil
-    britive>=2.23.0
+    britive>=2.24.0rc1
     jmespath
     pyjwt
 

src/pybritive/britive_cli.py

@@ -1,4 +1,5 @@
 import csv
+import hashlib
 from datetime import datetime
 from datetime import timezone
 from datetime import timedelta
@@ -29,7 +30,7 @@
 
 
 class BritiveCli:
-    def __init__(self, tenant_name: str = None, token: str = None, silent: bool = False,passphrase: str = None,
+    def __init__(self, tenant_name: str = None, token: str = None, silent: bool = False, passphrase: str = None,
                  federation_provider: str = None, from_helper_console_script: bool = False):
         self.silent = silent
         self.from_helper_console_script = from_helper_console_script
@@ -168,6 +169,22 @@ def login(self, explicit: bool = False, browser: str = None):
         if explicit and should_get_profiles:
             self._set_available_profiles()  # will handle calling cache_profiles() and construct_kube_config()
 
+        # handle printing the banner
+        self._display_banner()
+
+    def _display_banner(self):
+        if self.silent:
+            return
+
+        if not Cache().banner_expired(tenant=self.tenant_name):  # if banner is not expired yet then nothing to do
+            return
+
+        # if we get here then we need to at least grab the banner and see if it has changed
+        banner = self.b.banner()
+        banner_changed = Cache().save_banner(tenant=self.tenant_name, banner=banner)
+        if banner and banner_changed:
+            self.print(f'*** {banner.get("messageType", "UNKNOWN")}: {banner.get("message", "<no message>")} ***')
+
     def _update_sdk_user_agent(self):
         # update the user agent to include the pybritive cli version
         user_agent = self.b.session.headers.get('User-Agent')
@@ -593,8 +610,10 @@ def checkin(self, profile, console):
             transaction_id=transaction_id
         )
 
-        if application_type in ('aws', 'aws standalone'):
+        if application_type in ['aws', 'aws standalone']:
             self.clear_cached_aws_credentials(profile)
+        if application_type in ['gcp']:
+            self.clear_gcloud_auth_key_files(profile=profile)
 
     def _checkout(self, profile_name, env_name, app_name, programmatic, blocktime, maxpolltime, justification):
         try:
@@ -649,7 +668,7 @@ def _split_profile_into_parts(self, profile):
     def _extend_checkout(self, profile, console):
         self.login()
         parts = self._split_profile_into_parts(profile)
-        response = self.b.my_access.extend_checkout_by_name(
+        self.b.my_access.extend_checkout_by_name(
             profile_name=parts['profile'],
             environment_name=parts['env'],
             application_name=parts['app'],
@@ -927,8 +946,49 @@ def request_withdraw(self, profile):
             environment_id=ids['environment_id']
         )
 
-    def clear_gcloud_auth_key_files(self):
-        self.config.clear_gcloud_auth_key_files()
+    @staticmethod
+    def build_gcloud_key_file_for_gcloudauthexec(profile: str):
+        profile_hash = hashlib.sha256(string=profile.encode('utf-8')).hexdigest()
+        return f'gcloudauthexec-{profile_hash}.json'
+
+    def clear_gcloud_auth_key_files(self, profile: str = None):
+        if profile:  # we want to attempt a gcloud cli command
+            import subprocess  # lazy load as this will not always be needed
+
+            # build the path to the key file in question
+            key_file = self.build_gcloud_key_file_for_gcloudauthexec(profile=profile)
+            path = Path(self.config.gcloud_key_file_path) / key_file
+
+            if path.exists():  # we have a valid gcloudauthexec key file, so we know there was a checkout with this mode
+                try:
+                    with open(str(path), 'r') as f:
+                        credentials = json.loads(f.read())
+                    commands = [
+                        'gcloud',
+                        'auth',
+                        'revoke',
+                        credentials['client_email'],
+                        '--verbosity=error'
+                    ]
+                    self.debug(' '.join(commands))
+                    subprocess.run(commands, check=True)
+
+                    gcloud_default_account = self.config.gcloud_default_account()
+
+                    if gcloud_default_account:
+                        commands = [
+                            'gcloud',
+                            'config',
+                            'set',
+                            'account',
+                            gcloud_default_account,  # no need for "" here as subprocess will properly escape
+                            '--verbosity=error'
+                        ]
+                        self.debug(' '.join(commands))
+                        subprocess.run(commands, check=True)
+                except Exception as e:
+                    self.print(f'could not reset gcloud CLI active account due to issue: {str(e)}')
+        self.config.clear_gcloud_auth_key_files(profile=profile)
 
     def api(self, method, parameters: dict, query=None):
         self.login()

src/pybritive/choices/mode.py

@@ -25,6 +25,7 @@
         'browser-chrome',
         'browser-chromium',
         'kube-exec',                # bake into kubeconfig with oidc exec output and additional caching to make kubectl more performant
+        'gcloudauthexec',           # will effectively execute results of gcloudauth in a sub-shell
     ],
     case_sensitive=False
 )

src/pybritive/commands/clear.py

@@ -2,6 +2,7 @@
 from ..helpers.build_britive import build_britive
 from ..helpers.profile_argument_decorator import click_smart_profile_argument
 
+
 @click.group()
 def clear():
     """Clear various local settings and configurations."""

src/pybritive/helpers/build_britive.py

@@ -4,13 +4,26 @@
 import click
 from merge_args import merge_args
 from ..britive_cli import BritiveCli
+from click import Context
 
 
 @dataclass
 class Common:
     britive: BritiveCli
 
 
+def should_set_output_format(ctx: Context) -> bool:
+    parent_command = ctx.parent.command.name
+    command = ctx.command.name
+    if parent_command in ['configure', 'clear']:
+        return False
+
+    if parent_command == 'cache' and command in ['clear']:
+        return False
+
+    return True
+
+
 # this wrapper exists to centralize all "common" CLI options (options that exist for all commands)
 def build_britive(f):
     @merge_args(f)
@@ -27,8 +40,7 @@ def wrapper(
             passphrase=kwargs.get('passphrase'),
             federation_provider=kwargs.get('federation_provider')
         ))
-        parent_command = ctx.parent.command.name
-        if parent_command != 'configure':
+        if should_set_output_format(ctx=ctx):
             ctx.obj.britive.set_output_format(kwargs.get('output_format'))
         return f(ctx=ctx, **kwargs)
     return wrapper

src/pybritive/helpers/cache.py

@@ -1,7 +1,9 @@
+import hashlib
 import json
 import os
 from pathlib import Path
 from .encryption import StringEncryption, InvalidPassphraseException
+import time
 
 
 class Cache:
@@ -15,7 +17,8 @@ def __init__(self, passphrase: str = None):
         self.default_key_values = {
             'profiles': [],
             'awscredentialprocess': {},
-            'kube-exec': {}
+            'kube-exec': {},
+            'banners': {}
         }
         self.load()
 
@@ -76,3 +79,31 @@ def save_credentials(self, profile_name: str, credentials: dict, mode: str = 'aw
     def clear_credentials(self, profile_name: str, mode: str = 'awscredentialprocess'):
         self.cache[mode].pop(profile_name.lower(), None)
         self.write()
+
+    @staticmethod
+    def hash_banner(banner: dict) -> str:
+        return hashlib.sha512(string=json.dumps(banner, default=str)).hexdigest()
+
+    def banner_expired(self, tenant: str) -> bool:
+        cached_banner_data = self.cache.get('banners', {}).get(tenant)
+        expires = 0
+        if cached_banner_data is not None:
+            expires = cached_banner_data.get('expires', 0)
+        return expires < int(time.time())
+
+    def save_banner(self, tenant: str, banner: dict) -> bool:
+        # if someone called this then we simply save the banner
+        # regardless of whether the cached record has expired yet
+        # as we assume the caller knows what they are doing
+
+        cached_banner_data = self.cache.get('banners', {}).get(tenant, {})
+        cached_hash = cached_banner_data.get('hash', '')
+        new_hash = hashlib.sha512(string=json.dumps(banner, default=str, sort_keys=True).encode('utf-8')).hexdigest()
+        self.cache['banners'][tenant] = {
+            'hash': new_hash,
+            'expires': int(time.time()) + (5 * 60)
+        }
+        self.write()
+
+        # return True if the hashes have changed, False is they are equal
+        return cached_hash != new_hash

src/pybritive/helpers/cloud_credential_printer.py

@@ -1,11 +1,13 @@
 import configparser
 import json
 import os
+import subprocess
 from pathlib import Path
 import platform
 import uuid
 import webbrowser
 import click
+import hashlib
 
 
 # trailing spaces matter as some options do not have the trailing space
@@ -64,6 +66,8 @@ def print(self):
             self.print_gcloudauth()
         if self.mode == 'kube':
             self.print_kube()
+        if self.mode == 'gcloudauthexec':
+            self.exec_gcloudauthexec()
 
     def print_console(self):
         url = self.credentials.get('url', self.credentials)
@@ -97,6 +101,9 @@ def print_azps(self):
     def print_gcloudauth(self):
         self._not_implemented()
 
+    def exec_gcloudautoauth(self):
+        self._not_implemented()
+
     def print_kube(self):
         self._not_implemented()
 
@@ -246,7 +253,7 @@ def print_json(self):
     def print_gcloudauth(self):
         # get path to gcloud key file
         if not self.gcloud_key_file:  # if --gcloud-key-file not provided
-            path = Path(self.cli.config.path).parent / 'pybritive-gcloud-key-files' / f'{uuid.uuid4().hex}.json'
+            path = Path(self.cli.config.gcloud_key_file_path) / f'{uuid.uuid4().hex}.json'
         else:  # we need to parse out/sanitize what was provided
             path = Path(self.gcloud_key_file).expanduser().absolute()
 
@@ -259,6 +266,29 @@ def print_gcloudauth(self):
             ignore_silent=True
         )
 
+    def exec_gcloudauthexec(self):
+        key_file = self.cli.build_gcloud_key_file_for_gcloudauthexec(profile=self.profile)
+        path = Path(self.cli.config.gcloud_key_file_path) / key_file
+
+        # key file does not yet exist so write to it
+        path.parent.mkdir(exist_ok=True, parents=True)
+        path.write_text(json.dumps(self.credentials, indent=2), encoding='utf-8')
+
+        try:
+            commands = [
+                'gcloud',
+                'auth',
+                'activate-service-account',
+                self.credentials['client_email'],
+                '--key-file',
+                str(path),
+                '--verbosity=error'
+            ]
+
+            subprocess.run(commands, check=True)
+        except Exception as e:
+            self.cli.print(f'error running `gcloud auth activate-service-account ...`: {str(e)}')
+
 
 class KubernetesCredentialPrinter(CloudCredentialPrinter):
     def __init__(self, console, mode, profile, silent, credentials, cli, k8s_processor):