Merge pull request #186 from britive/v2.1.0-rc.4

v2.1.0-rc.4

Author: theborch

CHANGELOG.md

@@ -3,6 +3,28 @@
 > 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.
 
+## v2.1.0-rc.4 [2025-03-06]
+
+__What's New:__
+
+* Added "Global Settings" section to docs site.
+
+__Enhancements:__
+
+* Additional `global` config settings: `my_[access|resources]_retrieval_limit` to limit size of retrieved items.
+
+__Bug Fixes:__
+
+* Fixed missing `exceptions.StepUpAuthRequiredButNotProvided` catch during `checkout`.
+
+__Dependencies:__
+
+* None
+
+__Other:__
+
+* Allow `_` uniformity for `auto_refresh_[kube_config|profile_cache]` in `global` config.
+
 ## v2.1.0-rc.3 [2025-02-28]
 
 __What's New:__

docs/index.md

@@ -113,6 +113,60 @@ __windows (cmd):__
 set REQUESTS_CA_BUNDLE="C:\Users\User\AppData\Local\corp-proxy\cacert.pem"
 ```
 
+### Global Settings
+
+#### `credential_backend`
+
+The backend used to store temporary access tokens to authenticate against the Britive tenant.
+
+_Allowed value:_ `encrypted-file` or `file`
+
+#### `default_tenant`
+
+The name of the tenant used by default: [tenant].britive-app.com.
+
+_Allowed value:_ the name of a configured tenant alias, e.g. `[tenant-sigma]` would be `sigma`.
+
+#### `output_format`
+
+Display output format.
+
+If `table` is used, an optional table format can be specified as `table-format`, formats can be found here: [table_format](https://github.com/astanin/python-tabulate#table_format).
+
+_Allowed value:_ `json`, `yaml`, `csv`, or `table[-format]`
+
+> _NOTE:_ the following global config settings are NOT available directly via `pybritive configure global`
+
+#### `auto_refresh_kube_config`
+
+Auto refresh the cached Britive managed kube config.
+
+_Allowed value:_ `true` or `false`
+
+#### `auto_refresh_profile_cache`
+
+Auto refresh the cached Britive profiles.
+
+_Allowed value:_ `true` or `false`
+
+#### `ca_bundle`
+
+The custom TLS certificate to use when making HTTP requests.
+
+_Allowed value:_ the path to a custom TLS certificate, e.g. `/location/of/the/CA_BUNDLE_FILE.pem`
+
+#### `my_access_retrieval_limit`
+
+Limit the number of "My Access" profiles to be retrieved.
+
+_Allowed value:_ an integer greater than `0`
+
+#### `my_resources_retrieval_limit`
+
+Limit the number of "My Resources" items to be retrieved.
+
+_Allowed value:_ an integer greater than `0`
+
 ## Tenant Configuration
 
 Before `pybritive` can connect to a Britive tenant, it needs to know some details about that tenant.
@@ -758,13 +812,13 @@ The cache will not be updated over time. In order to update the cache more regul
 Note that this config flag is NOT available directly via `pybritive configure global ...`.
 
 ```sh
-pybritive configure update global auto-refresh-profile-cache true
+pybritive configure update global auto_refresh_profile_cache true
 ```
 
 To turn the feature off run
 
 ```sh
-pybritive configure update global auto-refresh-profile-cache false
+pybritive configure update global auto_refresh_profile_cache false
 pybritive cache clear
 ```
 

src/pybritive/__init__.py

@@ -1 +1 @@
-__version__ = '2.1.0-rc.3'
+__version__ = '2.1.0-rc.4'

src/pybritive/britive_cli.py

@@ -162,7 +162,6 @@ def login(self, explicit: bool = False, browser: str = default_browser):
         should_get_profiles = any([self.config.auto_refresh_profile_cache(), self.config.auto_refresh_kube_config()])
         if explicit and should_get_profiles:
             self._set_available_profiles()  # will handle calling cache_profiles() and construct_kube_config()
-
         self._display_banner()
 
     def _display_banner(self):
@@ -343,7 +342,11 @@ def list_resources(self):
         self.login()
         found_resource_names = []
         resources = []
-        for item in self.b.my_resources.list_profiles():
+        if resource_profile_limit := int(self.config.my_resources_retrieval_limit):
+            profiles = self.b.my_resources.list(size=resource_profile_limit)['data']
+        else:
+            profiles = self.b.my_resources.list_profiles()
+        for item in profiles:
             name = item['resourceName']
             if name not in found_resource_names:
                 resources.append(
@@ -389,7 +392,6 @@ def list_profiles(self, checked_out: bool = False, profile_type: Optional[str] =
                 if profile_is_checked_out:
                     row['Expiration'] = checked_out_profiles[key]['expiration']
                     total_seconds = checked_out_profiles[key]['expires_in_seconds']
-
                     hours, remainder = divmod(total_seconds, 3600)
                     minutes, seconds = divmod(remainder, 60)
                     time_format = f'{hours:02d}:{minutes:02d}:{seconds:02d}'
@@ -406,7 +408,7 @@ def list_profiles(self, checked_out: bool = False, profile_type: Optional[str] =
                     if profile['2_part_profile_format_allowed']:
                         row.pop('Environment', None)
                 elif self.output_format == 'json':
-                    row['Name'] = f"{row['Application']}/{row['Environment']}/{row['Profile']}"
+                    row['Name'] = f'{row["Application"]}/{row["Environment"]}/{row["Profile"]}'
 
                 data.append(row)
 
@@ -462,29 +464,43 @@ def _set_available_profiles(self, from_cache_command=False, profile_type: Option
         if not self.available_profiles:
             data = []
             if not profile_type or profile_type == 'my-access':
-                for app in self.b.my_access.list_profiles():
-                    for profile in app.get('profiles', []):
-                        for env in profile.get('environments', []):
-                            row = {
-                                'app_name': app['appName'],
-                                'app_id': app['appContainerId'],
-                                'app_type': app['catalogAppName'],
-                                'app_description': app['appDescription'],
-                                'env_name': env['environmentName'],
-                                'env_id': env['environmentId'],
-                                'env_short_name': env['alternateEnvironmentName'],
-                                'env_description': env['environmentDescription'],
-                                'profile_name': profile['profileName'],
-                                'profile_id': profile['profileId'],
-                                'profile_allows_console': profile['consoleAccess'],
-                                'profile_allows_programmatic': profile['programmaticAccess'],
-                                'profile_description': profile['profileDescription'],
-                                '2_part_profile_format_allowed': app['requiresHierarchicalModel'],
-                                'env_properties': env.get('profileEnvironmentProperties', {}),
-                            }
-                            data.append(row)
+                access_profile_limit = int(self.config.my_access_retrieval_limit)
+                access_data = self.b.my_access.list(size=access_profile_limit)
+                apps = {app['appContainerId']: app for app in access_data['apps']}
+                envs = {env['environmentId']: env for env in access_data['environments']}
+                accs = {
+                    p: {'env_id': v, 'types': [t['accessType'] for t in access_data['accesses'] if t['papId'] == p]}
+                    for p, v in {(a['papId'], a['environmentId']) for a in access_data['accesses']}
+                }
+                for profile in access_data['profiles']:
+                    acc = accs.get(profile['papId'], {})
+                    env = envs.get(acc.get('env_id'), {})  # Get environment info or default to empty dict
+                    app = apps.get(profile['appContainerId'], {})
+                    row = {
+                        'app_name': profile['catalogAppDisplayName'],
+                        'app_id': profile['appContainerId'],
+                        'app_type': profile['catalogAppName'],
+                        'app_description': app.get('appDescription'),
+                        'env_name': env.get('environmentName', ''),  # Pull from `environments`
+                        'env_id': env.get('environmentId', ''),  # Pull from `environments`
+                        'env_short_name': env.get('alternateEnvironmentName', ''),  # Pull from `environments`
+                        'env_description': env.get('environmentDescription', ''),  # Pull from `environments`
+                        'profile_name': profile['papName'],
+                        'profile_id': profile['papId'],
+                        'profile_allows_console': 'CONSOLE' in acc.get('types'),
+                        'profile_allows_programmatic': 'PROGRAMMATIC' in acc.get('types'),
+                        'profile_description': profile['papDescription'],
+                        '2_part_profile_format_allowed': app['requiresHierarchicalModel'],
+                        'env_properties': env.get('profileEnvironmentProperties', {}),  # Pull from `environments`
+                    }
+                    data.append(row)
             if self.b.feature_flags.get('server-access') and (not profile_type or profile_type == 'my-resources'):
-                for item in self.b.my_resources.list_profiles():
+                if not (resource_profile_limit := int(self.config.my_resources_retrieval_limit)):
+                    profiles = self.b.my_resources.list_profiles()
+                else:
+                    profiles = self.b.my_resources.list(size=resource_profile_limit)
+                    profiles = profiles['data']
+                for item in profiles:
                     row = {
                         'app_name': None,
                         'app_id': None,
@@ -697,6 +713,8 @@ def _checkout(
             if mode == 'awscredentialprocess':
                 raise e
             raise click.ClickException('approval required and no justification provided.') from e
+        except exceptions.StepUpAuthRequiredButNotProvided as e:
+            raise click.ClickException('Step Up Authentication required and no OTP provided.') from e
         except ValueError as e:
             raise click.BadParameter(str(e)) from e
         except Exception as e:
@@ -761,21 +779,25 @@ def _profile_is_for_resource(self, profile, profile_type):
         return real_profile_name.startswith(f'{self.resource_profile_prefix}')
 
     def _resource_checkout(self, blocktime, justification, maxpolltime, profile, ticket_id, ticket_type):
-        self.login()
-        resource_name, profile_name = self._split_resource_profile_into_parts(profile=profile)
-        response = self.b.my_resources.checkout_by_name(
-            include_credentials=True,
-            justification=justification,
-            max_wait_time=maxpolltime,
-            profile_name=profile_name[0],
-            progress_func=self.checkout_callback_printer,  # callback will handle silent, isatty, etc.
-            resource_name=resource_name,
-            response_template=profile_name[1] if len(profile_name) > 1 else None,
-            ticket_id=ticket_id,
-            ticket_type=ticket_type,
-            wait_time=blocktime,
-        )
-        return response['credentials']
+        try:
+            self.login()
+            resource_name, profile_name = self._split_resource_profile_into_parts(profile=profile)
+            return self.b.my_resources.checkout_by_name(
+                include_credentials=True,
+                justification=justification,
+                max_wait_time=maxpolltime,
+                profile_name=profile_name[0],
+                progress_func=self.checkout_callback_printer,  # callback will handle silent, isatty, etc.
+                resource_name=resource_name,
+                response_template=profile_name[1] if len(profile_name) > 1 else None,
+                ticket_id=ticket_id,
+                ticket_type=ticket_type,
+                wait_time=blocktime,
+            )['credentials']
+        except exceptions.ApprovalRequiredButNoJustificationProvided as e:
+            raise click.ClickException('approval required and no justification provided.') from e
+        except exceptions.StepUpAuthRequiredButNotProvided as e:
+            raise click.ClickException('Step Up Authentication required and no OTP provided.') from e
 
     def _access_checkout(
         self,
@@ -1455,7 +1477,7 @@ def clear_cached_aws_credentials(self, profile):
         # profile name as well - it will not hurt anything to try to clear
         # both versions
         parts = self._split_profile_into_parts(profile)
-        Cache().clear_credentials(profile_name=f"{parts['app']}/{parts['env']}/{parts['profile']}")
+        Cache().clear_credentials(profile_name=f'{parts["app"]}/{parts["env"]}/{parts["profile"]}')
 
     def ssh_gcp_identity_aware_proxy(self, username, hostname, push_public_key, port_number, key_source):
         self.silent = True

src/pybritive/helpers/aws_credential_process.py

@@ -160,8 +160,8 @@ def get_args():
 
 def usage():
     print(
-        f'Usage : {argv[0]} --profile <profile> [-t/--tenant, -T/--token, -p/--passphrase, -f/--force-renew, '
-        f'-F/--federation-provider]'
+        f'Usage : {argv[0]} -P/--profile <profile> [-t/--tenant, -T/--token, -p/--passphrase, -f/--force-renew,'
+        ' -F/--federation-provider]'
     )
     raise SystemExit
 

src/pybritive/helpers/config.py

@@ -40,12 +40,14 @@ def coalesce(*arg):
 non_tenant_sections = ['global', 'profile-aliases', 'aws', 'gcp']
 
 global_fields = [
+    'auto_refresh_kube_config',
+    'auto_refresh_profile_cache',
+    'ca_bundle',
+    'credential_backend',
     'default_tenant',
     'output_format',
-    'credential_backend',
-    'auto-refresh-profile-cache',
-    'auto-refresh-kube-config',
-    'ca_bundle',
+    'my_access_retrieval_limit',
+    'my_resources_retrieval_limit',
 ]
 
 tenant_fields = ['name', 'output_format', 'sso_idp']
@@ -73,6 +75,8 @@ def __init__(self, cli: object, tenant_name: Optional[str] = None):
         self.validation_error_messages = []
         self.gcloud_key_file_path: str = str(Path(self.path).parent / 'pybritive-gcloud-key-files')
         self.global_ca_bundle = None
+        self.my_access_retrieval_limit = None
+        self.my_resources_retrieval_limit = None
 
     def clear_gcloud_auth_key_files(self, profile=None):
         path = Path(self.gcloud_key_file_path)
@@ -121,7 +125,9 @@ def load(self, force=False):
                     self.tenants_by_name[name] = item
         self.aliases_and_names = {**self.tenants, **self.tenants_by_name}
         self.profile_aliases = self.config.get('profile-aliases', {})
-        self.global_ca_bundle = self.config.get('ca_bundle', {})
+        self.global_ca_bundle = self.config.get('global', {}).get('ca_bundle')
+        self.my_access_retrieval_limit = self.config.get('global', {}).get('my_access_retrieval_limit', '0')
+        self.my_resources_retrieval_limit = self.config.get('global', {}).get('my_resources_retrieval_limit', '0')
         self.loaded = True
 
     def get_tenant(self):
@@ -260,10 +266,10 @@ def validate_global(self, section, fields):
             if field == 'credential_backend' and value not in backend_choices.choices:
                 error = f'Invalid {section} field {field} value {value} provided. Invalid value choice.'
                 self.validation_error_messages.append(error)
-            if field == 'auto-refresh-profile-cache' and value not in ['true', 'false']:
+            if field.replace('-', '_') == 'auto_refresh_profile_cache' and value not in ['true', 'false']:
                 error = f'Invalid {section} field {field} value {value} provided. Invalid value choice.'
                 self.validation_error_messages.append(error)
-            if field == 'auto-refresh-kube-config' and value not in ['true', 'false']:
+            if field.replace('-', '_') == 'auto_refresh_kube_config' and value not in ['true', 'false']:
                 error = f'Invalid {section} field {field} value {value} provided. Invalid value choice.'
                 self.validation_error_messages.append(error)
             if field == 'default_tenant':
@@ -276,6 +282,9 @@ def validate_global(self, section, fields):
                 if not Path.is_file(ca_bundle_file_path):
                     error = f'Invalid {field} file {ca_bundle_file_path}. File does not exist.'
                     self.validation_error_messages.append(error)
+            if field in ['my_access_retrieval_limit', 'my_resources_retrieval_limit'] and not value.isnumeric():
+                error = f'Invalid {section} field {field} value {value} provided. Must be an integer.'
+                self.validation_error_messages.append(error)
 
     def validate_profile_aliases(self, section, fields):
         for field, value in fields.items():
@@ -314,10 +323,14 @@ def validate_tenant(self, section, fields):
 
     def auto_refresh_profile_cache(self):
         self.load()
-        value = self.config.get('global', {}).get('auto-refresh-profile-cache', 'false')
+        value = self.config.get('global', {}).get(
+            'auto_refresh_profile_cache', self.config.get('global', {}).get('auto-refresh-profile-cache', 'false')
+        )
         return value == 'true'
 
     def auto_refresh_kube_config(self):
         self.load()
-        value = self.config.get('global', {}).get('auto-refresh-kube-config', 'false')
+        value = self.config.get('global', {}).get(
+            'auto_refresh_kube_config', self.config.get('global', {}).get('auto-refresh-kube-config', 'false')
+        )
         return value == 'true'

src/pybritive/helpers/credentials.py

@@ -83,8 +83,7 @@ def _setup_requests_session(self):
         self.session = requests.Session()
         retries = Retry(total=5, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
         self.session.mount('https://', HTTPAdapter(max_retries=retries))
-        global_ca_bundle = self.cli.config.get_tenant().get('ca_bundle')
-        if global_ca_bundle:
+        if global_ca_bundle := self.cli.config.global_ca_bundle:
             os.environ['PYBRITIVE_CA_BUNDLE'] = global_ca_bundle
             self.session.verify = global_ca_bundle
         # allow the disabling of TLS/SSL verification for testing in development (mostly local development)