diff --git a/keepercommander/__init__.py b/keepercommander/__init__.py index ded16103e..0c9cce431 100644 --- a/keepercommander/__init__.py +++ b/keepercommander/__init__.py @@ -10,4 +10,4 @@ # Contact: commander@keepersecurity.com # -__version__ = '18.0.10' +__version__ = '18.0.11' diff --git a/keepercommander/auth/console_ui.py b/keepercommander/auth/console_ui.py index eca6bf3b4..0af2ccf58 100644 --- a/keepercommander/auth/console_ui.py +++ b/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}') diff --git a/keepercommander/commands/discoveryrotation.py b/keepercommander/commands/discoveryrotation.py index 031665ac2..f8f3e6434 100644 --- a/keepercommander/commands/discoveryrotation.py +++ b/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': diff --git a/keepercommander/importer/imp_exp.py b/keepercommander/importer/imp_exp.py index 38703a9dc..48b29176f 100644 --- a/keepercommander/importer/imp_exp.py +++ b/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() diff --git a/keepercommander/utils.py b/keepercommander/utils.py index 3a6f84119..41d3f3829 100644 --- a/keepercommander/utils.py +++ b/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}') diff --git a/keepercommander/vault.py b/keepercommander/vault.py index 0fa854f82..52a16489f 100644 --- a/keepercommander/vault.py +++ b/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 '' diff --git a/unit-tests/test_importer.py b/unit-tests/test_importer.py index 19217776a..aaa203611 100644 --- a/unit-tests/test_importer.py +++ b/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 * * ?') diff --git a/unit-tests/test_windows_file_permissions.py b/unit-tests/test_windows_file_permissions.py new file mode 100644 index 000000000..c058da26e --- /dev/null +++ b/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)