DR-1295 Add pamGitHubConfiguration record support

keepercommander/__init__.py

@@ -10,4 +10,4 @@
 # Contact: commander@keepersecurity.com
 #
 
-__version__ = '18.0.10'
+__version__ = '18.0.11'

keepercommander/auth/console_ui.py

@@ -387,7 +387,11 @@ def on_sso_redirect(self, step):
                     token = pyperclip.paste()
                 except:
                     token = ''
-                    logging.warning('Failed to paste from clipboard')
+                if len(token) < 10:
+                    try:
+                        token = input(f'{Fore.GREEN}Paste SSO Token: {Fore.RESET}').strip()
+                    except KeyboardInterrupt:
+                        token = ''
             else:
                 if len(token) < 10:
                     logging.warning(f'Unsupported menu option: {token}')

keepercommander/commands/discoveryrotation.py

@@ -2337,7 +2337,8 @@ def print_root_rotation_setting(params, is_verbose=False, format_type='table'):
 
         for c in configurations:  # type: vault.TypedRecord
             if c.record_type in ('pamAwsConfiguration', 'pamAzureConfiguration', 'pamGcpConfiguration',
-                                 'pamDomainConfiguration', 'pamNetworkConfiguration', 'pamOciConfiguration'):
+                                 'pamDomainConfiguration', 'pamNetworkConfiguration', 'pamOciConfiguration',
+                                 'pamGitHubConfiguration'):
                 facade.record = c
                 shared_folder_parents = find_parent_top_folder(params, c.record_uid)
                 if shared_folder_parents:
@@ -2399,7 +2400,7 @@ def print_root_rotation_setting(params, is_verbose=False, format_type='table'):
 
 common_parser = argparse.ArgumentParser(add_help=False)
 common_parser.add_argument('--environment', '-env', dest='config_type', action='store',
-                           choices=['local', 'aws', 'azure', 'gcp', 'domain', 'oci'], help='PAM Configuration Type')
+                           choices=['local', 'aws', 'azure', 'gcp', 'domain', 'oci', 'github'], help='PAM Configuration Type')
 common_parser.add_argument('--title', '-t', dest='title', action='store', help='Title of the PAM Configuration')
 common_parser.add_argument('--gateway', '-g', dest='gateway_uid', action='store', help='Gateway UID or Name')
 common_parser.add_argument('--shared-folder', '-sf', dest='shared_folder_uid', action='store',
@@ -2458,6 +2459,12 @@ def print_root_rotation_setting(params, is_verbose=False, format_type='table'):
                        help='Google Workspace Administrator Email Address')
 gcp_group.add_argument('--gcp-region', dest='region_names', action='append', help='GCP Region Names')
 
+github_group = common_parser.add_argument_group('github', 'GitHub configuration')
+github_group.add_argument('--github-id', dest='github_id', action='store', help='GitHub Id')
+github_group.add_argument('--personal-access-token', dest='personal_access_token', action='store',
+                       help='Personal Access Token')
+github_group.add_argument('--github-base-url', dest='github_base_url', action='store',
+                       help='GitHub Base URL')
 
 class PamConfigurationEditMixin(RecordEditMixin):
     pam_record_types = None
@@ -2639,6 +2646,16 @@ def parse_properties(self, params, record, **kwargs):  # type: (KeeperParams, va
             if gcp_region:
                 regions = '\n'.join(gcp_region)
                 extra_properties.append(f'multiline.pamGcpRegionName={regions}')
+        elif record.record_type == 'pamGitHubConfiguration':
+            github_id = kwargs.get('github_id')
+            if github_id:
+                extra_properties.append(f'text.pamGitHubId={github_id}')
+            personal_access_token = kwargs.get('personal_access_token')
+            if personal_access_token:
+                extra_properties.append(f'secret.personalAccessToken={personal_access_token}')
+            github_base_url = kwargs.get('github_base_url')
+            if github_base_url:
+                extra_properties.append(f'text.pamGitHubBaseUrl={github_base_url}')
         elif record.record_type == 'pamAzureConfiguration':
             azure_id = kwargs.get('azure_id')
             if azure_id:
@@ -2800,13 +2817,15 @@ def execute(self, params, **kwargs):
             record_type = 'pamNetworkConfiguration'
         elif config_type == 'gcp':
             record_type = 'pamGcpConfiguration'
+        elif config_type == 'github':
+            record_type = 'pamGitHubConfiguration'
         elif config_type == 'domain':
             record_type = 'pamDomainConfiguration'
         elif config_type == 'oci':
             record_type = 'pamOciConfiguration'
         else:
             raise CommandError('pam-config-new', f'--environment {config_type} is not supported'
-                                                 ' - supported options: local, aws, azure, gcp, domain, oci')
+                                                 ' - supported options: local, aws, azure, gcp, domain, oci, github')
 
         title = kwargs.get('title')
         if not title:
@@ -2955,6 +2974,8 @@ def execute(self, params, **kwargs):
                 record_type = 'pamNetworkConfiguration'
             elif config_type == 'gcp':
                 record_type = 'pamGcpConfiguration'
+            elif config_type == 'github':
+                record_type = 'pamGitHubConfiguration'
             elif config_type == 'domain':
                 record_type = 'pamDomainConfiguration'
             elif config_type == 'oci':

keepercommander/importer/imp_exp.py

@@ -1796,7 +1796,7 @@ def value_to_token(value): # type: (any) -> str
     return str(value)
 
 
-def tokenize_record_key(record):   # type: (ImportRecord) -> Iterator[str]
+def tokenize_record_key(record, folder):   # type: (ImportRecord, str) -> Iterator[str]
     """
     Turn a record-to-import into an iterable of str's for hashing.  This is really about import --update.
 
@@ -1807,6 +1807,7 @@ def tokenize_record_key(record):   # type: (ImportRecord) -> Iterator[str]
     yield f'$title:{record.title or ""}'
     yield f'$login:{record.login or ""}'
     yield f'$url:{record.login_url or ""}'
+    yield f'$folder:{folder}'
 
     if record_type in {'', 'login'}:
         return
@@ -2071,7 +2072,7 @@ def prepare_record_add_or_update(update_flag, params, records):
         If a 100% match is found for a record, then just skip requesting anything; it doesn't need to be changed.
         Otherwise import the record, risking creating an almost-duplicate.
     If update_flag is True:
-       if a unique field match (on title, login, and url) is found, then request a change in password only.
+       if a unique field match (on title, login, url, and folder) is found, then request a change in password only.
     """
     preexisting_entire_record_hash = {}
     preexisting_partial_record_hash = {}
@@ -2081,8 +2082,13 @@ def prepare_record_add_or_update(update_flag, params, records):
             record_hash = build_record_hash(tokenize_full_import_record(import_record))
             preexisting_entire_record_hash[record_hash] = record_uid
             if update_flag:
-                record_hash = build_record_hash(tokenize_record_key(import_record))
-                preexisting_partial_record_hash[record_hash] = record_uid
+                folders = [get_folder_path(params, x) for x in find_folders(params, record_uid)]
+                folders = [x for x in folders if x]
+                if len(folders) == 0:
+                    folders.append(record_uid)
+                for folder in folders:
+                    record_hash = build_record_hash(tokenize_record_key(import_record, folder))
+                    preexisting_partial_record_hash[record_hash] = record_uid
         else:
             pass
 
@@ -2128,15 +2134,26 @@ def prepare_record_add_or_update(update_flag, params, records):
             record_to_import.append(import_record)
             continue
         elif update_flag:
-            record_hash = build_record_hash(tokenize_record_key(import_record))
-            if record_hash in preexisting_partial_record_hash:
-                record_uid = preexisting_partial_record_hash[record_hash]
-                if import_record.uid:
-                    external_lookup[import_record.uid] = record_uid
-                if record_uid not in record_uid_to_update:
-                    record_uid_to_update.add(record_uid)
-                    import_record.uid = record_uid
-                    record_to_import.append(import_record)
+            folders = []
+            if import_record.folders:
+                folders.extend([x.get_folder_path() for x in import_record.folders])
+            folders = [x for x in folders if x]
+            if len(folders) == 0:
+                folders.append('')
+            found = False
+            for folder in folders:
+                record_hash = build_record_hash(tokenize_record_key(import_record, folder))
+                if record_hash in preexisting_partial_record_hash:
+                    record_uid = preexisting_partial_record_hash[record_hash]
+                    if import_record.uid:
+                        external_lookup[import_record.uid] = record_uid
+                    if record_uid not in record_uid_to_update:
+                        record_uid_to_update.add(record_uid)
+                        import_record.uid = record_uid
+                        record_to_import.append(import_record)
+                    found = True
+                    break
+            if found:
                 continue
 
         record_uid = utils.generate_uid()

keepercommander/utils.py

@@ -90,6 +90,19 @@ def generate_aes_key():         # type: () -> bytes
     return crypto.get_random_bytes(32)
 
 
+def _windows_icacls_principal():     # type: () -> str
+    """Return a DOMAIN\\username principal suitable for icacls /grant on Windows."""
+    username = os.environ.get('USERNAME', '')
+    if not username:
+        username = os.getlogin()
+    if '\\' in username:
+        return username
+    domain = os.environ.get('USERDOMAIN') or os.environ.get('COMPUTERNAME', '')
+    if domain:
+        return f'{domain}\\{username}'
+    return username
+
+
 def set_file_permissions(file_path):     # type: (str) -> None
     """
     Set secure file permissions (600) for configuration files containing sensitive data.
@@ -111,10 +124,10 @@ def set_file_permissions(file_path):     # type: (str) -> None
             os.chmod(file_path, stat.S_IRUSR | stat.S_IWUSR)
             logging.debug(f'Set secure permissions (600) for file: {file_path}')
         else:
-            username = os.getlogin()
+            principal = _windows_icacls_principal()
             subprocess.run(["icacls", file_path, "/inheritance:r"], check=True, capture_output=True)
             subprocess.run(["icacls", file_path, "/remove", "NT AUTHORITY\\SYSTEM", "BUILTIN\\Administrators"], check=False, capture_output=True)
-            subprocess.run(["icacls", file_path, "/grant", f"{username}:RW"], check=True, capture_output=True)
+            subprocess.run(["icacls", file_path, "/grant", f"{principal}:RW"], check=True, capture_output=True)
             logging.debug(f'Set secure permissions (owner RW only) for Windows file: {file_path}')
     except Exception:
         logging.warning(f'Failed to set file permissions for {file_path}')

keepercommander/vault.py

@@ -475,8 +475,7 @@ def export_schedule_field(value):   # type: (dict) -> Optional[str]
                 cron = value.get('cron')
                 if isinstance(cron, str):
                     comps = [x for x in cron.split(' ') if x]
-                    if len(comps) >= 6:
-                        comps = comps[1:6]
+                    if comps:
                         return ' '.join(comps)
                 return ''
 

unit-tests/test_importer.py

@@ -230,3 +230,13 @@ def test_schedule_parser(self):
         self.assertEqual(sc.get('type'), 'MONTHLY_BY_WEEKDAY')
         self.assertEqual(sc.get('weekday'), 'WEDNESDAY')
         self.assertEqual(sc.get('occurrence'), 'SECOND')
+
+    def test_export_schedule_field_preserves_quartz_cron_seconds(self):
+        schedule = {"type": "CRON", "cron": "1 5 1 * * ?", "tz": "Etc/UTC"}
+        self.assertEqual(vault.TypedField.export_schedule_field(schedule), '1 5 1 * * ?')
+
+        schedule = {"type": "CRON", "cron": "0 5 1 * * ?"}
+        self.assertEqual(vault.TypedField.export_schedule_field(schedule), '0 5 1 * * ?')
+
+        schedule = {"type": "CRON", "cron": "5 1 * * ?"}
+        self.assertEqual(vault.TypedField.export_schedule_field(schedule), '5 1 * * ?')

unit-tests/test_windows_file_permissions.py

@@ -0,0 +1,68 @@
+import os
+import tempfile
+from unittest import TestCase, mock
+
+from keepercommander import utils
+
+
+class TestWindowsIcaclsPrincipal(TestCase):
+    def test_userdomain_and_username(self):
+        with mock.patch.dict(os.environ, {'USERNAME': 'ivan', 'USERDOMAIN': 'IVAN'}, clear=False):
+            self.assertEqual(utils._windows_icacls_principal(), 'IVAN\\ivan')
+
+    def test_domain_user(self):
+        with mock.patch.dict(os.environ, {'USERNAME': 'jdoe', 'USERDOMAIN': 'CORP'}, clear=False):
+            self.assertEqual(utils._windows_icacls_principal(), 'CORP\\jdoe')
+
+    def test_falls_back_to_computername(self):
+        env = os.environ.copy()
+        env.pop('USERDOMAIN', None)
+        with mock.patch.dict(os.environ, env, clear=True):
+            os.environ['USERNAME'] = 'bob'
+            os.environ['COMPUTERNAME'] = 'MYPC'
+            self.assertEqual(utils._windows_icacls_principal(), 'MYPC\\bob')
+
+    def test_already_qualified_username(self):
+        with mock.patch.dict(os.environ, {'USERNAME': 'CORP\\jdoe', 'USERDOMAIN': 'CORP'}, clear=False):
+            self.assertEqual(utils._windows_icacls_principal(), 'CORP\\jdoe')
+
+    def test_falls_back_to_getlogin(self):
+        with mock.patch.dict(os.environ, {}, clear=True):
+            with mock.patch('os.getlogin', return_value='localuser'):
+                with mock.patch.dict(os.environ, {'COMPUTERNAME': 'MYPC'}, clear=False):
+                    self.assertEqual(utils._windows_icacls_principal(), 'MYPC\\localuser')
+
+
+class TestSetFilePermissionsWindows(TestCase):
+    def _grant_principal(self, mock_run):
+        for call in mock_run.call_args_list:
+            args = call.args[0]
+            if '/grant' in args:
+                return args[args.index('/grant') + 1]
+        self.fail('icacls /grant was not called')
+
+    @mock.patch('subprocess.run')
+    @mock.patch('platform.system', return_value='Windows')
+    @mock.patch('os.path.islink', return_value=False)
+    def test_grant_uses_qualified_principal_when_names_collide(self, _islink, _system, mock_run):
+        with tempfile.NamedTemporaryFile(delete=False) as tmp:
+            path = tmp.name
+        try:
+            with mock.patch.dict(os.environ, {'USERNAME': 'ivan', 'USERDOMAIN': 'IVAN'}, clear=False):
+                utils.set_file_permissions(path)
+            self.assertEqual(self._grant_principal(mock_run), 'IVAN\\ivan:RW')
+        finally:
+            os.unlink(path)
+
+    @mock.patch('subprocess.run')
+    @mock.patch('platform.system', return_value='Windows')
+    @mock.patch('os.path.islink', return_value=False)
+    def test_grant_uses_domain_principal(self, _islink, _system, mock_run):
+        with tempfile.NamedTemporaryFile(delete=False) as tmp:
+            path = tmp.name
+        try:
+            with mock.patch.dict(os.environ, {'USERNAME': 'jdoe', 'USERDOMAIN': 'CORP'}, clear=False):
+                utils.set_file_permissions(path)
+            self.assertEqual(self._grant_principal(mock_run), 'CORP\\jdoe:RW')
+        finally:
+            os.unlink(path)