New Mac keyring entries don't trust all Python

AGENTS.md

@@ -25,8 +25,10 @@ A command line client for MySQL with auto-completion and syntax highlighting.
 ├── mycli/config.py                       # configuration file readers and utilities
 ├── mycli/constants.py                    # shared constants
 ├── mycli/key_bindings.py                 # prompt_toolkit key binding utilities
+├── mycli/keyring_utils.py                # saved credential utilities
 ├── mycli/kubectl_tunnel.py               # connection over kubectl tunnel
 ├── mycli/lexer.py                        # extends `MySqlLexer` from Pygments
+├── mycli/macos_keychain.py               # macOS Keychain credentials preserving access controls
 ├── mycli/main.py                         # processes CLI arguments
 ├── mycli/main_modes/                     # main execution paths
 ├── mycli/main_modes/batch.py             # `--batch` mode

changelog.md

@@ -1,6 +1,11 @@
 Upcoming (TBD)
 ==============
 
+Bug Fixes
+--------
+* New macOS keyring entries no longer automatically trust Python.
+
+
 Documentation
 --------
 * Update features bullets in `README.md`.

mycli/client_connection.py

@@ -22,6 +22,7 @@
     EMPTY_PASSWORD_FLAG_SENTINEL,
     ER_MUST_CHANGE_PASSWORD_LOGIN,
 )
+from mycli.keyring_utils import set_keyring_password
 from mycli.kubectl_tunnel import KubectlTunnel, KubectlTunnelError
 from mycli.packages.filepaths import guess_socket_location
 from mycli.packages.special.utils import format_connection_dsn
@@ -381,7 +382,7 @@ def _update_keyring(
                 try:
                     saved_pw = keyring.get_password(keyring_domain, keyring_identifier)
                     if password != saved_pw or reset_keyring:
-                        keyring.set_password(keyring_domain, keyring_identifier, password)
+                        set_keyring_password(keyring_domain, keyring_identifier, password)
                         click.secho(
                             f'Password from source: "{password_source}" '
                             f'saved to the system keyring at {keyring_domain}/{keyring_identifier}',

mycli/keyring_utils.py

@@ -0,0 +1,34 @@
+import sys
+
+import keyring
+from keyring.backend import KeyringBackend
+
+
+def _set_password_with_backend(backend: KeyringBackend, service: str, account: str, password: str) -> None:
+    from keyring.backends.macOS import Keyring
+
+    if type(backend) is Keyring:
+        from mycli import macos_keychain
+
+        macos_keychain.set_password(service, account, password)
+    else:
+        backend.set_password(service, account, password)
+
+
+def set_keyring_password(service: str, account: str, password: str) -> None:
+    if sys.platform != 'darwin':
+        keyring.set_password(service, account, password)
+        return
+
+    from keyring.backends.chainer import ChainerBackend  # type: ignore[unreachable]
+
+    backend = keyring.get_keyring()
+    if type(backend) is ChainerBackend:
+        for candidate in backend.backends:
+            try:
+                _set_password_with_backend(candidate, service, account, password)
+                return
+            except NotImplementedError:
+                continue
+    else:
+        _set_password_with_backend(backend, service, account, password)

mycli/macos_keychain.py

@@ -0,0 +1,80 @@
+import ctypes
+from typing import Any
+
+from keyring.errors import KeyringLocked, PasswordSetError
+
+
+def _get_api() -> Any:
+    from keyring.backends.macOS import api
+
+    return api
+
+
+def set_password(service: str, account: str, password: str) -> None:
+    """Preserve existing ACLs and create credentials without trusting Python."""
+    api = _get_api()
+    cf_array_create = api._found.CFArrayCreate
+    cf_array_create.restype = ctypes.c_void_p
+    cf_array_create.argtypes = (ctypes.c_void_p, ctypes.c_void_p, ctypes.c_long, ctypes.c_void_p)
+    cf_data_create = api._found.CFDataCreate
+    cf_data_create.restype = ctypes.c_void_p
+    cf_data_create.argtypes = (ctypes.c_void_p, ctypes.c_void_p, ctypes.c_long)
+    cf_release = api._found.CFRelease
+    cf_release.restype = None
+    cf_release.argtypes = (ctypes.c_void_p,)
+    sec_access_create = api._sec.SecAccessCreate
+    sec_access_create.restype = api.OS_status
+    sec_access_create.argtypes = (ctypes.c_void_p, ctypes.c_void_p, ctypes.POINTER(ctypes.c_void_p))
+    sec_item_update = api._sec.SecItemUpdate
+    sec_item_update.restype = api.OS_status
+    sec_item_update.argtypes = (ctypes.c_void_p, ctypes.c_void_p)
+
+    retained: list[ctypes.c_void_p] = []
+
+    def retain(value: int | ctypes.c_void_p | None, name: str) -> ctypes.c_void_p:
+        if not value:
+            raise RuntimeError(f'Unable to allocate Keychain {name}')
+        pointer = value if isinstance(value, ctypes.c_void_p) else ctypes.c_void_p(value)
+        retained.append(pointer)
+        return pointer
+
+    try:
+        service_value = retain(api.create_cf(service), 'service')
+        account_value = retain(api.create_cf(account), 'account')
+        encoded = password.encode('utf-8')
+        password_value = retain(cf_data_create(None, ctypes.create_string_buffer(encoded), len(encoded)), 'password')
+        search = retain(
+            api.create_query(
+                kSecClass=api.k_('kSecClassGenericPassword'),
+                kSecAttrService=service_value,
+                kSecAttrAccount=account_value,
+            ),
+            'search',
+        )
+        attributes = retain(api.create_query(kSecValueData=password_value), 'attributes')
+        status = sec_item_update(search, attributes)
+        if status == api.error.item_not_found:
+            # An empty array trusts nobody; NULL would trust the calling executable.
+            trusted_apps = retain(cf_array_create(None, None, 0, None), 'access controls')
+            access = ctypes.c_void_p()
+            api.Error.raise_for_status(sec_access_create(service_value, trusted_apps, ctypes.byref(access)))
+            retain(access, 'access')
+            item = retain(
+                api.create_query(
+                    kSecClass=api.k_('kSecClassGenericPassword'),
+                    kSecAttrService=service_value,
+                    kSecAttrAccount=account_value,
+                    kSecValueData=password_value,
+                    kSecAttrAccess=access,
+                ),
+                'item',
+            )
+            status = api.SecItemAdd(item, None)
+        api.Error.raise_for_status(status)
+    except api.KeychainDenied as e:
+        raise KeyringLocked(f"Can't store password on keychain: {e}") from e
+    except api.Error as e:
+        raise PasswordSetError(f"Can't store password on keychain: {e}") from e
+    finally:
+        for value in reversed(retained):
+            cf_release(value)

test/pytests/test_client_connection.py

@@ -260,11 +260,14 @@ def fake_get_password(domain: str, identifier: str) -> str:
         return 'from-keyring'
 
     monkeypatch.setattr(client_connection.keyring, 'get_password', fake_get_password)
+    writes: list[tuple[str, str, str]] = []
+    monkeypatch.setattr(client_connection, 'set_keyring_password', lambda *args: writes.append(args))
 
     client.connect(user='alice', host='db', port=3307, use_keyring=True)
 
     assert FakeSQLExecute.calls[-1]['password'] == 'from-keyring'
     assert get_password_calls == [('mycli.net', 'alice@db:3307:')]
+    assert writes == []
 
 
 def test_connect_uses_mylogin_password_before_keyring(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -275,7 +278,7 @@ def test_connect_uses_mylogin_password_before_keyring(monkeypatch: pytest.Monkey
         'get_password',
         lambda domain, identifier: get_password_calls.append((domain, identifier)) or 'from-mylogin',  # type: ignore[func-returns-value]
     )
-    monkeypatch.setattr(client_connection.keyring, 'set_password', lambda *_args: None)
+    monkeypatch.setattr(client_connection, 'set_keyring_password', lambda *_args: None)
 
     client.connect(user='alice', host='db', port=3307, use_keyring=True)
 
@@ -293,7 +296,7 @@ def test_connect_resolves_supplied_candidates_after_connection_details(monkeypat
         'get_password',
         lambda domain, identifier: keyring_calls.append((domain, identifier)) or 'from-dsn',  # type: ignore[func-returns-value]
     )
-    monkeypatch.setattr(client_connection.keyring, 'set_password', lambda *_args: None)
+    monkeypatch.setattr(client_connection, 'set_keyring_password', lambda *_args: None)
 
     client.connect(user='alice', host='db', port=3307, password_candidates=candidates, use_keyring=True)
 
@@ -326,8 +329,8 @@ def test_connect_saves_selected_password_to_keyring(monkeypatch: pytest.MonkeyPa
     secho_calls: list[tuple[str, dict[str, Any]]] = []
     monkeypatch.setattr(client_connection.keyring, 'get_password', lambda *_args: 'old-secret')
     monkeypatch.setattr(
-        client_connection.keyring,
-        'set_password',
+        client_connection,
+        'set_keyring_password',
         lambda domain, identifier, password: set_password_calls.append((domain, identifier, password)),
     )
     monkeypatch.setattr(client_connection.click, 'secho', lambda message, **kwargs: secho_calls.append((message, kwargs)))
@@ -348,14 +351,54 @@ def test_connect_reports_keyring_save_error(monkeypatch: pytest.MonkeyPatch) ->
     def fail_set_password(*_args: Any) -> None:
         raise RuntimeError('locked')
 
-    monkeypatch.setattr(client_connection.keyring, 'set_password', fail_set_password)
+    monkeypatch.setattr(client_connection, 'set_keyring_password', fail_set_password)
     monkeypatch.setattr(client_connection.click, 'secho', lambda message, **kwargs: secho_calls.append((message, kwargs)))
 
     client.connect(user='alice', host='db', port=3307, password_candidates=candidates, use_keyring=True)
 
     assert secho_calls == [('Password not saved to the system keyring: locked', {'err': True, 'fg': 'red'})]
 
 
+def test_connect_saves_fallback_password_when_keyring_is_empty(monkeypatch: pytest.MonkeyPatch) -> None:
+    client = DummyClient()
+    FakeSQLExecute.effects = [op_error(1045), None]
+    writes: list[tuple[str, str, str]] = []
+    monkeypatch.setattr(client_connection.keyring, 'get_password', lambda *_args: None)
+    monkeypatch.setattr(client_connection.click, 'prompt', lambda *_args, **_kwargs: 'new-secret')
+    monkeypatch.setattr(client_connection, 'set_keyring_password', lambda *args: writes.append(args))
+
+    client.connect(user='alice', host='db', port=3307, use_keyring=True)
+
+    assert writes == [('mycli.net', 'alice@db:3307:', 'new-secret')]
+
+
+def test_connect_does_not_rewrite_rejected_keyring_password(monkeypatch: pytest.MonkeyPatch) -> None:
+    client = DummyClient()
+    FakeSQLExecute.effects = [op_error(1045)]
+    writes: list[tuple[str, str, str]] = []
+    monkeypatch.setattr(client_connection.keyring, 'get_password', lambda *_args: 'old-secret')
+    monkeypatch.setattr(client_connection, 'set_keyring_password', lambda *args: writes.append(args))
+
+    with pytest.raises(SystemExit):
+        client.connect(user='alice', host='db', port=3307, use_keyring=True)
+
+    assert writes == []
+
+
+@pytest.mark.parametrize('reset', [False, True])
+def test_connect_only_rewrites_unchanged_password_for_reset(monkeypatch: pytest.MonkeyPatch, reset: bool) -> None:
+    client = DummyClient()
+    candidates = PasswordCandidates()
+    candidates.add_value('literal', 'secret')
+    writes: list[tuple[str, str, str]] = []
+    monkeypatch.setattr(client_connection.keyring, 'get_password', lambda *_args: 'secret')
+    monkeypatch.setattr(client_connection, 'set_keyring_password', lambda *args: writes.append(args))
+
+    client.connect(user='alice', host='db', port=3307, password_candidates=candidates, use_keyring=True, reset_keyring=reset)
+
+    assert writes == ([('mycli.net', 'alice@db:3307:', 'secret')] if reset else [])
+
+
 def test_connect_uses_ssh_jump_with_remote_socket(monkeypatch: pytest.MonkeyPatch) -> None:
     tunnel_calls: list[dict[str, Any]] = []
 
@@ -485,8 +528,8 @@ def close(self) -> None:
         lambda domain, identifier: keyring_calls.append(('get', domain, identifier)),
     )
     monkeypatch.setattr(
-        client_connection.keyring,
-        'set_password',
+        client_connection,
+        'set_keyring_password',
         lambda domain, identifier, password: keyring_calls.append(('set', domain, identifier, password)),
     )
     password_candidates = PasswordCandidates()

test/pytests/test_keyring_utils.py

@@ -0,0 +1,105 @@
+from unittest.mock import Mock
+
+from keyring.backends.chainer import ChainerBackend
+from keyring.backends.macOS import Keyring
+import pytest
+
+from mycli import keyring_utils, macos_keychain
+
+
+@pytest.fixture
+def native_write(monkeypatch: pytest.MonkeyPatch) -> Mock:
+    write = Mock()
+    monkeypatch.setattr(keyring_utils.sys, 'platform', 'darwin')
+    monkeypatch.setattr(macos_keychain, 'set_password', write)
+    return write
+
+
+@pytest.mark.parametrize('platform', ['linux', 'win32'])
+def test_other_platforms_use_configured_keyring(monkeypatch: pytest.MonkeyPatch, platform: str) -> None:
+    write = Mock()
+    monkeypatch.setattr(keyring_utils.sys, 'platform', platform)
+    monkeypatch.setattr(keyring_utils.keyring, 'set_password', write)
+    monkeypatch.setattr(keyring_utils.keyring, 'get_keyring', Mock(side_effect=AssertionError('Do not inspect backend')))
+
+    keyring_utils.set_keyring_password('mycli.net', 'account', 'secret')
+
+    write.assert_called_once_with('mycli.net', 'account', 'secret')
+
+
+@pytest.mark.parametrize('chained', [False, True])
+def test_native_backend_uses_restricted_writer(monkeypatch: pytest.MonkeyPatch, native_write: Mock, chained: bool) -> None:
+    backend = Keyring()
+    monkeypatch.setattr(backend, 'set_password', Mock(side_effect=AssertionError('Unsafe native write')))
+    monkeypatch.setattr(ChainerBackend, 'backends', [backend])
+    monkeypatch.setattr(keyring_utils.keyring, 'get_keyring', lambda: ChainerBackend() if chained else backend)
+
+    keyring_utils.set_keyring_password('mycli.net', 'account', 'secret')
+
+    native_write.assert_called_once_with('mycli.net', 'account', 'secret')
+
+
+def test_custom_backend_is_not_unwrapped(monkeypatch: pytest.MonkeyPatch, native_write: Mock) -> None:
+    backend = Mock(backends=[Keyring()])
+    monkeypatch.setattr(keyring_utils.keyring, 'get_keyring', lambda: backend)
+
+    keyring_utils.set_keyring_password('service', 'account', 'secret')
+
+    backend.set_password.assert_called_once_with('service', 'account', 'secret')
+    native_write.assert_not_called()
+
+
+def test_native_subclass_retains_custom_write(monkeypatch: pytest.MonkeyPatch, native_write: Mock) -> None:
+    class CustomKeyring(Keyring):
+        pass
+
+    backend = CustomKeyring()
+    write = Mock()
+    monkeypatch.setattr(backend, 'set_password', write)
+    monkeypatch.setattr(keyring_utils.keyring, 'get_keyring', lambda: backend)
+
+    keyring_utils.set_keyring_password('service', 'account', 'secret')
+
+    write.assert_called_once_with('service', 'account', 'secret')
+    native_write.assert_not_called()
+
+
+@pytest.mark.parametrize('read_only', [False, True])
+def test_chainer_preserves_backend_order(monkeypatch: pytest.MonkeyPatch, native_write: Mock, read_only: bool) -> None:
+    first = Mock()
+    if read_only:
+        first.set_password.side_effect = NotImplementedError
+    monkeypatch.setattr(ChainerBackend, 'backends', [first, Keyring()])
+    monkeypatch.setattr(keyring_utils.keyring, 'get_keyring', ChainerBackend)
+
+    keyring_utils.set_keyring_password('service', 'account', 'secret')
+
+    first.set_password.assert_called_once_with('service', 'account', 'secret')
+    assert native_write.call_count == int(read_only)
+
+
+@pytest.mark.parametrize('chained', [False, True])
+def test_native_failure_never_falls_back(monkeypatch: pytest.MonkeyPatch, native_write: Mock, chained: bool) -> None:
+    native_write.side_effect = RuntimeError('denied')
+    fallback = Mock()
+    backend = Keyring()
+    monkeypatch.setattr(backend, 'set_password', Mock(side_effect=AssertionError('Unsafe native write')))
+    monkeypatch.setattr(ChainerBackend, 'backends', [backend, fallback])
+    monkeypatch.setattr(keyring_utils.keyring, 'get_keyring', lambda: ChainerBackend() if chained else backend)
+
+    with pytest.raises(RuntimeError, match='denied'):
+        keyring_utils.set_keyring_password('service', 'account', 'secret')
+
+    fallback.set_password.assert_not_called()
+
+
+def test_read_only_chainer_keeps_existing_behavior(monkeypatch: pytest.MonkeyPatch, native_write: Mock) -> None:
+    backend = Mock()
+    backend.set_password.side_effect = NotImplementedError
+    monkeypatch.setattr(ChainerBackend, 'backends', [backend])
+    monkeypatch.setattr(keyring_utils.keyring, 'get_keyring', ChainerBackend)
+
+    keyring_utils.set_keyring_password('service', 'account', 'secret')
+
+    backend.set_password.assert_called_once()
+    native_write.assert_not_called()