diff --git a/keepercommander/__main__.py b/keepercommander/__main__.py index 58772abf9..1b1ef2b15 100644 --- a/keepercommander/__main__.py +++ b/keepercommander/__main__.py @@ -118,6 +118,8 @@ def show_brief_help(): print(' --debug Turn on debug mode') print(' --silent Turn off all logging statements') print(' --batch-mode Run in batch/non-interactive mode') + print(' --skip-scan Skip login-time BreachWatch/security scans and device registration') + print(' Vault and enterprise sync still run') print(' --proxy PROXY Proxy server') print(' --new-login Force full login (bypass persistent login)') print(' --version Display version') @@ -152,6 +154,9 @@ def show_brief_help(): parser.add_argument('--debug', dest='debug', action='store_true', help='Turn on debug mode') parser.add_argument('--silent', dest='silent', action='store_true', help='Turn off all logging statements') parser.add_argument('--batch-mode', dest='batch_mode', action='store_true', help='Run commander in batch or basic UI mode.') +parser.add_argument('--skip-scan', dest='skip_scan', action='store_true', + help='Skip login-time BreachWatch/security scans and device registration; ' + 'vault/enterprise sync still runs.') parser.add_argument('--launched-with-shortcut', '-lwsc', dest='launched_with_shortcut', action='store', help='Indicates that the app was launched using a shortcut, for example using Mac App or from ' 'Windows Start Menu.') @@ -235,7 +240,7 @@ def main(from_package=False): '--data-dir', '-ks', '-ku', '-kp', '-lwsc'] bool_main_parser_args = ['--version', '--debug', '--silent', '--batch-mode', '--unmask-all', '--fail-on-throttle', - '--new-login', '--config-file'] + '--new-login', '--config-file', '--skip-scan'] main_parser_args = value_main_parser_args + bool_main_parser_args is_main_parser_arg = False @@ -267,6 +272,7 @@ def main(from_package=False): if opts.batch_mode: params.batch_mode = True + params.skip_scan = opts.skip_scan if opts.debug: params.debug = opts.debug diff --git a/keepercommander/commands/utils.py b/keepercommander/commands/utils.py index 3be980c95..df0dbb370 100644 --- a/keepercommander/commands/utils.py +++ b/keepercommander/commands/utils.py @@ -302,6 +302,9 @@ def register_command_info(aliases, command_info): help='Store config in config.json instead of the OS-native keychain ' '(use for headless servers, Docker, or CI/CD environments). ' 'Equivalent to setting KEEPER_CONFIG_STORAGE=file.') +login_parser.add_argument('--skip-scan', dest='skip_scan', action='store_true', + help='Skip login-time BreachWatch/security scans and device registration; ' + 'vault/enterprise sync still runs.') login_parser.add_argument('email', nargs='?', type=str, help='account email') login_parser.error = raise_parse_exception login_parser.exit = suppress_exit @@ -1751,6 +1754,7 @@ def execute(self, params, **kwargs): params.password = password new_login = kwargs.get('new_login') is True skip_sync = kwargs.get('skip_sync') is True + skip_scan = kwargs.get('skip_scan') is True or params.skip_scan is True # Apply storage backend choice before login so that store_config_properties # (called inside api.login) honours the user's explicit preference. @@ -1809,20 +1813,21 @@ def execute(self, params, **kwargs): SyncDownCommand().execute(params, force=True) if params.is_enterprise_admin: api.query_enterprise(params, True) - try: - if params.breach_watch: - BreachWatchScanCommand().execute(params, suppress_no_op=True) - if params.enterprise_ec_key: - SyncSecurityDataCommand().execute(params, record='@all', suppress_no_op=True) - except Exception as e: - logging.warning(f'A problem was encountered while updating BreachWatch/security data: {e}') - logging.debug(e, exc_info=True) + if not skip_scan: + try: + if params.breach_watch: + BreachWatchScanCommand().execute(params, suppress_no_op=True) + if params.enterprise_ec_key: + SyncSecurityDataCommand().execute(params, record='@all', suppress_no_op=True) + except Exception as e: + logging.warning(f'A problem was encountered while updating BreachWatch/security data: {e}') + logging.debug(e, exc_info=True) - # Auto-register device for persistent login (stores encrypted data key on server) - try: - loginv3.LoginV3API.register_encrypted_data_key_for_device(params) - except Exception as e: - logging.debug(f'Device registration: {e}') + # Auto-register device for persistent login (stores encrypted data key on server) + try: + loginv3.LoginV3API.register_encrypted_data_key_for_device(params) + except Exception as e: + logging.debug(f'Device registration: {e}') # Print config storage confirmation, mirroring KSM CLI's post-init messages. stored_backend = (params.config or {}).get(CONFIG_STORAGE_URL, '') diff --git a/keepercommander/params.py b/keepercommander/params.py index c40cdffee..8e345046d 100644 --- a/keepercommander/params.py +++ b/keepercommander/params.py @@ -239,6 +239,7 @@ def __init__(self, config_filename='', config=None, server='keepersecurity.com') self.clone_code = None self.device_token = None self.device_private_key = None + self.skip_scan = False # Skip login-time BreachWatch/security scans and device registration only. self.account_uid_bytes = None self.session_token_bytes = None self.record_type_cache = {} # RT definitions only diff --git a/unit-tests/test_command_utils.py b/unit-tests/test_command_utils.py index a06fe846e..ae5c005e8 100644 --- a/unit-tests/test_command_utils.py +++ b/unit-tests/test_command_utils.py @@ -6,12 +6,52 @@ from data_vault import get_synced_params, get_user_params, get_connected_params, VaultEnvironment from helper import KeeperApiHelper from keepercommander.commands import utils +from keepercommander import __main__ as keeper_main +from keepercommander.__main__ import parser as keeper_parser vault_env = VaultEnvironment() ent_env = EnterpriseEnvironment() +class TestSkipScanCLI(TestCase): + def test_skip_scan_flag_before_or_after_command(self): + for argv in (['--skip-scan', 'list'], ['list', '--skip-scan']): + with self.subTest(argv=argv): + opts, remaining = keeper_parser.parse_known_args(argv) + self.assertEqual(opts.command, 'list') + self.assertTrue(opts.skip_scan) + self.assertEqual(remaining, []) + + def test_login_command_flag_is_parsed(self): + opts = utils.LoginCommand().get_parser().parse_args([ + '--skip-scan', 'user@example.com' + ]) + self.assertTrue(opts.skip_scan) + + def test_global_flag_survives_login_command_reconstruction(self): + command_lines = ( + (['keeper', '--skip-scan', 'login', 'user@example.com'], + ['login user@example.com', 'q']), + (['keeper', '--skip-scan', 'shell', 'login', 'user@example.com'], + ['login user@example.com']), + (['keeper', 'shell', '--skip-scan', 'login', 'user@example.com'], + ['login user@example.com']), + ) + for argv, expected_commands in command_lines: + with self.subTest(argv=argv): + params = get_user_params() + with mock.patch('sys.argv', argv), \ + mock.patch.object(keeper_main, 'get_params_from_config', return_value=params), \ + mock.patch.object(keeper_main.cli, 'loop', return_value=0) as loop, \ + mock.patch('sys.exit'): + keeper_main.main() + + self.assertTrue(params.skip_scan) + loop.assert_called_once() + self.assertEqual(loop.call_args.args[0].commands, expected_commands) + + class TestRegister(TestCase): enterpriseInviteCode = '987654321' @@ -58,6 +98,89 @@ def test_login(self): cmd.execute(params) mock_login.assert_not_called() + def test_login_skip_scan_keeps_vault_and_enterprise_sync(self): + params = get_user_params() + params.config['config_storage'] = 'file' + params.skip_scan = True + + def sync_down_effect(p, **kwargs): + p.sync_data = False + + with mock.patch('keepercommander.api.login', side_effect=lambda p, **kwargs: setattr(p, 'session_token', 'token')), \ + mock.patch.object(utils.SyncDownCommand, 'execute', side_effect=sync_down_effect) as sync_down, \ + mock.patch('keepercommander.api.query_enterprise') as query_enterprise, \ + mock.patch.object(utils.BreachWatchScanCommand, 'execute') as breachwatch_scan, \ + mock.patch.object(utils.SyncSecurityDataCommand, 'execute') as sync_security_data, \ + mock.patch('keepercommander.loginv3.LoginV3API.register_encrypted_data_key_for_device') as register_device: + params.is_enterprise_admin = True + params.breach_watch = True + params.enterprise_ec_key = b'enterprise-key' + utils.LoginCommand().execute( + params, + email=params.user, + password=params.password, + show_help=False + ) + + sync_down.assert_called_once_with(params, force=True) + query_enterprise.assert_called_once_with(params, True) + breachwatch_scan.assert_not_called() + sync_security_data.assert_not_called() + register_device.assert_not_called() + # Normal vault sync leaves the command-dispatch post-sync hook idle. + self.assertFalse(params.sync_data) + + def test_login_command_skip_scan_flag_uses_kwargs(self): + params = get_user_params() + params.config['config_storage'] = 'file' + + def sync_down_effect(p, **kwargs): + p.sync_data = False + + with mock.patch('keepercommander.api.login', side_effect=lambda p, **kwargs: setattr(p, 'session_token', 'token')), \ + mock.patch.object(utils.SyncDownCommand, 'execute', side_effect=sync_down_effect) as sync_down, \ + mock.patch.object(utils.BreachWatchScanCommand, 'execute') as breachwatch_scan, \ + mock.patch.object(utils.SyncSecurityDataCommand, 'execute') as sync_security_data, \ + mock.patch('keepercommander.loginv3.LoginV3API.register_encrypted_data_key_for_device') as register_device: + params.breach_watch = True + params.enterprise_ec_key = b'enterprise-key' + utils.LoginCommand().execute_args( + params, + f'--skip-scan {params.user}', + show_help=False + ) + + sync_down.assert_called_once_with(params, force=True) + breachwatch_scan.assert_not_called() + sync_security_data.assert_not_called() + register_device.assert_not_called() + + def test_login_without_skip_scan_and_registration_keeps_post_login_work(self): + params = get_user_params() + params.config['config_storage'] = 'file' + params.is_enterprise_admin = True + params.breach_watch = True + params.enterprise_ec_key = b'enterprise-key' + + with mock.patch('keepercommander.api.login', side_effect=lambda p, **kwargs: setattr(p, 'session_token', 'token')), \ + mock.patch.object(utils.SyncDownCommand, 'execute') as sync_down, \ + mock.patch('keepercommander.api.query_enterprise') as query_enterprise, \ + mock.patch.object(utils.BreachWatchScanCommand, 'execute') as breachwatch_scan, \ + mock.patch.object(utils.SyncSecurityDataCommand, 'execute') as sync_security_data, \ + mock.patch('keepercommander.loginv3.LoginV3API.register_encrypted_data_key_for_device') as register_device: + utils.LoginCommand().execute( + params, + email=params.user, + password=params.password, + show_help=False + ) + + sync_down.assert_called_once_with(params, force=True) + query_enterprise.assert_called_once_with(params, True) + breachwatch_scan.assert_called_once_with(params, suppress_no_op=True) + sync_security_data.assert_called_once_with(params, record='@all', suppress_no_op=True) + register_device.assert_called_once_with(params) + def test_logout(self): params = get_synced_params() cmd = utils.LogoutCommand()