K8s mode

setup.cfg

@@ -38,4 +38,5 @@ where = src
 [options.entry_points]
 console_scripts =
     pybritive = pybritive.cli_interface:safe_cli
-    pybritive-aws-cred-process = pybritive.helpers.aws_credential_process:main
+    pybritive-aws-cred-process = pybritive.helpers.aws_credential_process:main
+    pybritive-kube-exec = pybritive.helpers.k8s_exec:main

src/pybritive/britive_cli.py

@@ -44,6 +44,16 @@ def __init__(self, tenant_name: str = None, token: str = None, silent: bool = Fa
         self.credential_manager = None
         self.verbose_checkout = False
         self.checkout_progress_previous_message = None
+        self.cachable_modes = {
+            'awscredentialprocess': {
+                'app_type': 'AWS',
+                'expiration_jmespath': 'expirationTime'
+            },
+            'kube-exec': {
+                'app_type': 'Kubernetes',
+                'expiration_jmespath': 'expirationTime'
+            }
+        }
         self.browser = None
 
     def set_output_format(self, output_format: str):
@@ -342,7 +352,7 @@ def _get_app_type(self, application_id):
         raise click.ClickException(f'Application {application_id} not found')
 
     def __get_cloud_credential_printer(self, app_type, console, mode, profile, silent, credentials,
-                                       aws_credentials_file, gcloud_key_file):
+                                       aws_credentials_file, gcloud_key_file, k8s_processor):
         if app_type in ['AWS', 'AWS Standalone']:
             return printer.AwsCloudCredentialPrinter(
                 console=console,
@@ -372,14 +382,25 @@ def __get_cloud_credential_printer(self, app_type, console, mode, profile, silen
                 cli=self,
                 gcloud_key_file=gcloud_key_file
             )
-        return printer.GenericCloudCredentialPrinter(
-            console=console,
-            mode=mode,
-            profile=profile,
-            credentials=credentials,
-            silent=silent,
-            cli=self
-        )
+        elif app_type in ['Kubernetes']:
+            return printer.KubernetesCredentialPrinter(
+                console=console,
+                mode=mode,
+                profile=profile,
+                credentials=credentials,
+                silent=silent,
+                cli=self,
+                k8s_processor=k8s_processor
+            )
+        else:
+            return printer.GenericCloudCredentialPrinter(
+                console=console,
+                mode=mode,
+                profile=profile,
+                credentials=credentials,
+                silent=silent,
+                cli=self
+            )
 
     def checkin(self, profile, console):
         self.login()
@@ -474,9 +495,18 @@ def checkout(self, alias, blocktime, console, justification, mode, maxpolltime,
                  force_renew, aws_credentials_file, gcloud_key_file, verbose):
         credentials = None
         app_type = None
-        credential_process_creds_found = False
+        cached_credentials_found = False
+        k8s_processor = None
         self.verbose_checkout = verbose
 
+        # handle kube-exec since the profile is actually going to be passed in via another method
+        # and perform some basic validation so we don't waste time performing a checkout when we
+        # will not be able to return a response back to kubectl via the exec command
+        if mode == 'kube-exec' or 'KUBERNETES_EXEC_INFO' in os.environ:
+            mode = 'kube-exec'  # set for downstream processes if we are basing this only on the env var being present
+            from .helpers.k8s_exec_credential_builder import KubernetesExecCredentialProcessor
+            k8s_processor = KubernetesExecCredentialProcessor()
+
         # these 2 modes implicitly say that console access should be checked out without having to provide
         # the --console flag
         if mode and (mode == 'console' or mode.startswith('browser')):
@@ -488,22 +518,23 @@ def checkout(self, alias, blocktime, console, justification, mode, maxpolltime,
 
         self._validate_justification(justification)
 
-        if mode == 'awscredentialprocess':
-            self.silent = True  # the aws credential process CANNOT output anything other than the expected JSON
-            # we need to check the credential process cache for the credentials first
-            # then check to see if they are expired
-            # 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=alias or profile)
+        if mode in self.cachable_modes:
+            self.silent = True  # CANNOT output anything other than the expected JSON
+            # we need to check the cache for the credentials first and then check to see if they are expired
+            # if not simply return those credentials, if they are expired, continue to do an actual checkout
+            app_type = self.cachable_modes[mode]['app_type']
+            credentials = Cache(passphrase=passphrase).get_credentials(profile_name=alias or profile, mode=mode)
             if credentials:
-                expiration_timestamp_str = credentials['expirationTime'].replace('Z', '')
+                expiration_timestamp_str = jmespath.search(
+                    expression=self.cachable_modes[mode]['expiration_jmespath'],
+                    data=credentials
+                ).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
+                    cached_credentials_found = True
 
         parts = self._split_profile_into_parts(profile)
 
@@ -518,7 +549,7 @@ def checkout(self, alias, blocktime, console, justification, mode, maxpolltime,
             'justification': justification
         }
 
-        if not credential_process_creds_found:  # nothing found via aws cred process or not aws cred process mode
+        if not cached_credentials_found:  # nothing found in cache, cache is expired, or not a cachable mode
             response = self._checkout(**params)
             app_type = self._get_app_type(response['appContainerId'])
             credentials = response['credentials']
@@ -533,16 +564,17 @@ def checkout(self, alias, blocktime, console, justification, mode, maxpolltime,
                 self.print('checking in the profile to get renewed credentials....standby')
                 self.checkin(profile=profile)
                 response = self._checkout(**params)
-                credential_process_creds_found = False  # need to write new creds to cache
+                cached_credentials_found = False  # need to write new creds to cache
                 credentials = response['credentials']
 
         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(
+        if mode in self.cachable_modes and not cached_credentials_found:
+            Cache(passphrase=passphrase).save_credentials(
                 profile_name=alias or profile,
-                credentials=credentials
+                credentials=credentials,
+                mode=mode
             )
 
         self.__get_cloud_credential_printer(
@@ -553,7 +585,8 @@ def checkout(self, alias, blocktime, console, justification, mode, maxpolltime,
             self.silent,
             credentials,
             aws_credentials_file,
-            gcloud_key_file
+            gcloud_key_file,
+            k8s_processor
         ).print()
 
     def import_existing_npm_config(self):

src/pybritive/choices/mode.py

@@ -23,7 +23,9 @@
         'browser-macosx',
         'browser-safari',
         'browser-chrome',
-        'browser-chromium'
+        'browser-chromium',
+        'kube-exec',                # bake into kubeconfig with oidc exec output and additional caching to make kubectl more performant
+        'kube-config',              # create a kubeconfig file with user oidc credentials and optionally cluster/context in ~/.britive/kubeconfig/
     ],
     case_sensitive=False
 )

src/pybritive/helpers/aws_credential_process.py

@@ -66,7 +66,10 @@ def main():
     creds = None
     if not args['force_renew']:  # if force renew let's defer to that the full package vs. this helper
         from .cache import Cache  # lazy load
-        creds = Cache(passphrase=args['passphrase']).get_awscredentialprocess(profile_name=args['profile'])
+        creds = Cache(passphrase=args['passphrase']).get_credentials(
+            profile_name=args['profile'],
+            mode='awscredentialprocess'
+        )
         if creds:
             from datetime import datetime  # lazy load
             expiration = datetime.fromisoformat(creds['expirationTime'].replace('Z', ''))

src/pybritive/helpers/cache.py

@@ -1,7 +1,6 @@
 import json
 import os
 from pathlib import Path
-
 from .encryption import StringEncryption, InvalidPassphraseException
 
 
@@ -12,6 +11,11 @@ def __init__(self, passphrase: str = None):
         home = os.getenv('PYBRITIVE_HOME_DIR', str(Path.home()))
         self.path = str(Path(home) / '.britive' / 'pybritive.cache')  # handle os specific separators properly
         self.cache = {}
+        self.default_key_values = {
+            'profiles': [],
+            'awscredentialprocess': {},
+            'kube-exec': {}
+        }
         self.load()
 
     def load(self):
@@ -29,10 +33,7 @@ def load(self):
             except json.decoder.JSONDecodeError:
                 self.cache = {}
 
-        if 'profiles' not in self.cache.keys():
-            self.cache['profiles'] = []
-        if 'awscredentialprocess' not in self.cache.keys():
-            self.cache['awscredentialprocess'] = {}
+        self.cache = {**self.default_key_values, **self.cache}
 
     def write(self):
         # write the new cache file
@@ -49,24 +50,23 @@ def save_profiles(self, profiles: list):
         self.write()
 
     def clear(self):
-        self.cache['profiles'] = []
-        self.cache['awscredentialprocess'] = {}
+        self.cache = self.default_key_values
         self.write()
 
-    def get_awscredentialprocess(self, profile_name: str):
+    def get_credentials(self, profile_name: str, mode: str = 'awscredentialprocess'):
         try:
-            ciphertext = self.cache['awscredentialprocess'].get(profile_name.lower())
+            ciphertext = self.cache[mode].get(profile_name.lower())
             if not ciphertext:
                 return None
             return json.loads(self.string_encryptor.decrypt(ciphertext))
         except InvalidPassphraseException:  # if we cannot decrypt don't error - just make the API call to get the creds
             return None
 
-    def save_awscredentialprocess(self, profile_name: str, credentials: dict):
+    def save_credentials(self, profile_name: str, credentials: dict, mode: str = 'awscredentialprocess'):
         ciphertext = self.string_encryptor.encrypt(json.dumps(credentials, default=str))
-        self.cache['awscredentialprocess'][profile_name.lower()] = ciphertext
+        self.cache[mode][profile_name.lower()] = ciphertext
         self.write()
 
-    def clear_awscredentialprocess(self, profile_name: str):
-        self.cache['awscredentialprocess'].pop(profile_name.lower(), None)
+    def clear_credentials(self, profile_name: str, mode: str = 'awscredentialprocess'):
+        self.cache[mode].pop(profile_name.lower(), None)
         self.write()

src/pybritive/helpers/cloud_credential_printer.py

@@ -62,6 +62,8 @@ def print(self):
             self.print_azps()
         if self.mode == 'gcloudauth':
             self.print_gcloudauth()
+        if self.mode == 'kube':
+            self.print_kube()
 
     def print_console(self):
         url = self.credentials.get('url', self.credentials)
@@ -95,6 +97,9 @@ def print_azps(self):
     def print_gcloudauth(self):
         self._not_implemented()
 
+    def print_kube(self):
+        self._not_implemented()
+
     def _not_implemented(self):
         raise click.ClickException(f'Application type {self.app_type} does not support the specified mode.')
 
@@ -253,3 +258,26 @@ def print_gcloudauth(self):
             f"gcloud auth activate-service-account {self.credentials['client_email']} --key-file {str(path)}",
             ignore_silent=True
         )
+
+
+class KubernetesCredentialPrinter(CloudCredentialPrinter):
+    def __init__(self, console, mode, profile, silent, credentials, cli, k8s_processor):
+        self.k8s_processor = k8s_processor
+        super().__init__('Kubernetes', console, mode, profile, silent, credentials, cli)
+
+    def print_json(self):
+        try:
+            self.cli.print(json.dumps(self.credentials, indent=2), ignore_silent=True)
+        except json.JSONDecodeError:
+            self.cli.print(self.credentials, ignore_silent=True)
+
+    def print_kube(self):
+        if self.mode_modifier == 'exec':
+            self.cli.print(self.k8s_processor.construct_exec_credential(self.credentials), ignore_silent=True)
+        elif self.mode_modifier == 'config':
+            # write file to ~/.britive/kubeconfig/...
+            # clean up any older config file that are no longer required
+            # print out `export KUBECONFIG=~/.britive/kubeconfig/...`
+            pass
+        else:
+            raise ValueError(f'--mode modifier {self.mode_modifier} for mode {self.mode} not supported')