new Mac keyring entries don't trust all Python

Previously, Mac keyring/keychain entries would trust Python, meaning
that any Python app could read the credential.  With this change, that
only happens if the user follows up with "always allow" to the keychain
dialog.

This still is not perfect!  But apparently the only way to do better on
Mac would be to have a mycli shim binary.

Based on dbcli/pgcli#1633 .

Author: rolandwalker

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