KC-1412: Fix email-config title lookup so shared-in records cannot redirect --send-email SMTP

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

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

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

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()