diff --git a/keepercommander/commands/discoveryrotation.py b/keepercommander/commands/discoveryrotation.py index 028693d2f..2d9069d10 100644 --- a/keepercommander/commands/discoveryrotation.py +++ b/keepercommander/commands/discoveryrotation.py @@ -3812,6 +3812,10 @@ def execute(self, params, **kwargs): # Find and load email config to validate provider and dependencies try: config_uid = find_email_config_record(params, self.email_config) + if not config_uid: + raise CommandError( + 'pam action rotate', + f'Email configuration "{self.email_config}" not found') email_config_obj = load_email_config_from_record(params, config_uid) # Check if required dependencies are installed for this provider diff --git a/keepercommander/commands/email_commands.py b/keepercommander/commands/email_commands.py index 0e195c26d..5edb037de 100644 --- a/keepercommander/commands/email_commands.py +++ b/keepercommander/commands/email_commands.py @@ -153,7 +153,12 @@ def find_email_config_record(params: KeeperParams, name: str) -> Optional[str]: """ - Find email config record by name. + Find an owned email config record by name. + + Only records owned by the current account are eligible. Shared or + non-owned records (even with a matching title and ``__email_config__`` + marker) are ignored so SMTP/provider settings cannot be supplied by + another user. Args: params: KeeperParams session @@ -175,8 +180,18 @@ def find_email_config_record(params: KeeperParams, name: str) -> Optional[str]: custom_fields = record_dict.get('custom', []) for field in custom_fields: if field.get('type') == 'text' and field.get('label') == '__email_config__': - if record.title == name: - return record_uid + if record.title != name: + continue + + owner = (params.record_owner_cache or {}).get(record_uid) + if not owner or not owner.owner: + logging.warning( + 'Ignoring email configuration "%s" (%s): ' + 'not owned by the current account (or ownership unknown)', + name, record_uid) + continue + + return record_uid except: continue @@ -591,13 +606,17 @@ def execute(self, params: KeeperParams, **kwargs): class EmailConfigListCommand(Command): - """List all email configurations.""" + """List owned email configurations.""" def get_parser(self): return email_config_list_parser def execute(self, params: KeeperParams, **kwargs): - """Execute email-config list command.""" + """Execute email-config list command. + + Only configurations owned by the current account are listed, matching + ``find_email_config_record`` eligibility used by test/delete/--send-email. + """ configs = [] # Find all email config records @@ -629,13 +648,24 @@ def execute(self, params: KeeperParams, **kwargs): if values: from_address = values[0] - if is_email_config: - configs.append({ - 'name': record.title, - 'record_uid': record_uid, - 'provider': provider or 'unknown', - 'from_address': from_address or '' - }) + if not is_email_config: + continue + + # Match find_email_config_record: only list owned configs + owner = (params.record_owner_cache or {}).get(record_uid) + if not owner or not owner.owner: + logging.debug( + 'Skipping email configuration "%s" (%s) in list: ' + 'not owned by the current account (or ownership unknown)', + record.title, record_uid) + continue + + configs.append({ + 'name': record.title, + 'record_uid': record_uid, + 'provider': provider or 'unknown', + 'from_address': from_address or '' + }) except Exception as e: logging.debug(f'Error loading record {record_uid}: {e}') continue diff --git a/keepercommander/commands/record_edit.py b/keepercommander/commands/record_edit.py index c0750fdb1..4b6bd2da3 100644 --- a/keepercommander/commands/record_edit.py +++ b/keepercommander/commands/record_edit.py @@ -881,6 +881,8 @@ def execute(self, params, **kwargs): email_config_name = kwargs.get('email_config') config_uid = find_email_config_record(params, email_config_name) + if not config_uid: + raise CommandError('record-add', f'Email configuration "{email_config_name}" not found') email_config_obj = load_email_config_from_record(params, config_uid) # Check if required dependencies are installed for this provider diff --git a/unit-tests/test_email_config_ownership.py b/unit-tests/test_email_config_ownership.py new file mode 100644 index 000000000..3ac0bae6e --- /dev/null +++ b/unit-tests/test_email_config_ownership.py @@ -0,0 +1,185 @@ +import unittest +from unittest.mock import MagicMock, patch + +from keepercommander.error import CommandError +from keepercommander.params import KeeperParams, RecordOwner + + +class TestFindEmailConfigRecordOwnership(unittest.TestCase): + """Email config records must be owned by the current account.""" + + TITLE = 'default' + + def _make_typed_record(self, uid, title=None): + from keepercommander import vault + + record = MagicMock(spec=vault.TypedRecord) + record.record_uid = uid + record.title = title or self.TITLE + record.record_type = 'login' + return record + + def _email_config_data(self): + return { + 'custom': [ + {'type': 'text', 'label': '__email_config__', 'value': ['true']}, + ] + } + + @patch('keepercommander.commands.email_commands.vault_extensions.extract_typed_record_data') + @patch('keepercommander.commands.email_commands.vault.KeeperRecord.load') + def test_skips_non_owned_record(self, mock_load, mock_extract): + from keepercommander.commands.email_commands import find_email_config_record + + shared = self._make_typed_record('SHARED_UID') + owned = self._make_typed_record('OWNED_UID') + + def load_side_effect(_params, uid): + return {'SHARED_UID': shared, 'OWNED_UID': owned}[uid] + + mock_load.side_effect = load_side_effect + mock_extract.return_value = self._email_config_data() + + params = MagicMock(spec=KeeperParams) + params.record_cache = {'SHARED_UID': {}, 'OWNED_UID': {}} + params.record_owner_cache = { + 'SHARED_UID': RecordOwner(False, 'attacker'), + 'OWNED_UID': RecordOwner(True, 'operator'), + } + + self.assertEqual(find_email_config_record(params, self.TITLE), 'OWNED_UID') + + @patch('keepercommander.commands.email_commands.vault_extensions.extract_typed_record_data') + @patch('keepercommander.commands.email_commands.vault.KeeperRecord.load') + def test_returns_none_when_only_shared_match(self, mock_load, mock_extract): + from keepercommander.commands.email_commands import find_email_config_record + + mock_load.return_value = self._make_typed_record('SHARED_UID') + mock_extract.return_value = self._email_config_data() + + params = MagicMock(spec=KeeperParams) + params.record_cache = {'SHARED_UID': {}} + params.record_owner_cache = { + 'SHARED_UID': RecordOwner(False, 'attacker'), + } + + self.assertIsNone(find_email_config_record(params, self.TITLE)) + + @patch('keepercommander.commands.email_commands.vault_extensions.extract_typed_record_data') + @patch('keepercommander.commands.email_commands.vault.KeeperRecord.load') + def test_returns_none_when_owner_cache_missing(self, mock_load, mock_extract): + from keepercommander.commands.email_commands import find_email_config_record + + mock_load.return_value = self._make_typed_record('UNKNOWN_UID') + mock_extract.return_value = self._email_config_data() + + params = MagicMock(spec=KeeperParams) + params.record_cache = {'UNKNOWN_UID': {}} + params.record_owner_cache = {} + + self.assertIsNone(find_email_config_record(params, self.TITLE)) + + @patch('keepercommander.commands.email_commands.vault_extensions.extract_typed_record_data') + @patch('keepercommander.commands.email_commands.vault.KeeperRecord.load') + def test_returns_owned_record(self, mock_load, mock_extract): + from keepercommander.commands.email_commands import find_email_config_record + + mock_load.return_value = self._make_typed_record('OWNED_UID') + mock_extract.return_value = self._email_config_data() + + params = MagicMock(spec=KeeperParams) + params.record_cache = {'OWNED_UID': {}} + params.record_owner_cache = { + 'OWNED_UID': RecordOwner(True, 'operator'), + } + + self.assertEqual(find_email_config_record(params, self.TITLE), 'OWNED_UID') + + +class TestEmailConfigListOwnership(unittest.TestCase): + """email-config list must only show owned configurations.""" + + def _make_typed_record(self, uid, title): + from keepercommander import vault + + record = MagicMock(spec=vault.TypedRecord) + record.record_uid = uid + record.title = title + record.record_type = 'login' + return record + + def _email_config_data(self, provider='smtp', from_address='it@corp.example'): + return { + 'custom': [ + {'type': 'text', 'label': '__email_config__', 'value': ['true']}, + {'type': 'text', 'label': 'provider', 'value': [provider]}, + {'type': 'text', 'label': 'from_address', 'value': [from_address]}, + ] + } + + @patch('keepercommander.commands.email_commands.dump_report_data') + @patch('keepercommander.commands.email_commands.vault_extensions.extract_typed_record_data') + @patch('keepercommander.commands.email_commands.vault.KeeperRecord.load') + def test_list_skips_shared_in_config(self, mock_load, mock_extract, mock_dump): + from keepercommander.commands.email_commands import EmailConfigListCommand + + shared = self._make_typed_record('SHARED_UID', 'default') + owned = self._make_typed_record('OWNED_UID', 'corp-smtp') + + def load_side_effect(_params, uid): + return {'SHARED_UID': shared, 'OWNED_UID': owned}[uid] + + mock_load.side_effect = load_side_effect + mock_extract.return_value = self._email_config_data() + + params = MagicMock(spec=KeeperParams) + params.record_cache = {'SHARED_UID': {}, 'OWNED_UID': {}} + params.record_owner_cache = { + 'SHARED_UID': RecordOwner(False, 'attacker'), + 'OWNED_UID': RecordOwner(True, 'operator'), + } + + EmailConfigListCommand().execute(params, format='table') + + mock_dump.assert_called_once() + table = mock_dump.call_args[0][0] + self.assertEqual(len(table), 1) + self.assertEqual(table[0][0], 'corp-smtp') + self.assertEqual(table[0][3], 'OWNED_UID') + + +class TestSendEmailValidationOwnership(unittest.TestCase): + """Patched validation paths must raise when only a shared-in config matches.""" + + @patch('keepercommander.commands.email_commands.find_email_config_record', return_value=None) + def test_record_add_raises_when_only_shared_config(self, _mock_find): + from keepercommander.commands.record_edit import RecordAddCommand + + params = MagicMock(spec=KeeperParams) + with self.assertRaises(CommandError) as ctx: + RecordAddCommand().execute( + params, + send_email='newhire@corp.example', + email_config='default', + ) + self.assertIn('Email configuration "default" not found', str(ctx.exception)) + + @patch('keepercommander.commands.discoveryrotation._is_rotation_allowed_by_enforcement', + return_value=True) + @patch('keepercommander.commands.discoveryrotation.find_email_config_record', return_value=None) + def test_pam_rotate_raises_when_only_shared_config(self, _mock_find, _mock_allowed): + from keepercommander.commands.discoveryrotation import PAMGatewayActionRotateCommand + + params = MagicMock(spec=KeeperParams) + with self.assertRaises(CommandError) as ctx: + PAMGatewayActionRotateCommand().execute( + params, + record_uid='REC_UID', + send_email='newhire@corp.example', + email_config='default', + ) + self.assertIn('Email configuration "default" not found', str(ctx.exception)) + + +if __name__ == '__main__': + unittest.main()