diff --git a/AGENTS.md b/AGENTS.md index b83d397f..5cfa5cc5 100644 --- a/AGENTS.md +++ b/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 diff --git a/changelog.md b/changelog.md index c3d28d0b..5efb1229 100644 --- a/changelog.md +++ b/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`. diff --git a/mycli/client_connection.py b/mycli/client_connection.py index db8844d4..df9c5d5b 100644 --- a/mycli/client_connection.py +++ b/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}', diff --git a/mycli/keyring_utils.py b/mycli/keyring_utils.py new file mode 100644 index 00000000..1a1741cd --- /dev/null +++ b/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) diff --git a/mycli/macos_keychain.py b/mycli/macos_keychain.py new file mode 100644 index 00000000..5d725e24 --- /dev/null +++ b/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) diff --git a/test/pytests/test_client_connection.py b/test/pytests/test_client_connection.py index df1aeb4c..a1483fef 100644 --- a/test/pytests/test_client_connection.py +++ b/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,7 +351,7 @@ 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) @@ -356,6 +359,46 @@ def fail_set_password(*_args: Any) -> None: 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() diff --git a/test/pytests/test_keyring_utils.py b/test/pytests/test_keyring_utils.py new file mode 100644 index 00000000..ea0d3789 --- /dev/null +++ b/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() diff --git a/test/pytests/test_macos_keychain.py b/test/pytests/test_macos_keychain.py new file mode 100644 index 00000000..09065a78 --- /dev/null +++ b/test/pytests/test_macos_keychain.py @@ -0,0 +1,160 @@ +import ctypes +from types import SimpleNamespace +from typing import Any +from unittest.mock import Mock + +from keyring.backends import macOS +from keyring.errors import KeyringLocked, PasswordSetError +import pytest + +from mycli import macos_keychain + + +class ApiError(Exception): + @classmethod + def raise_for_status(cls, status: int) -> None: + if status == -128: + raise ApiDenied(status) + if status: + raise cls(status) + + +class ApiDenied(ApiError): + pass + + +@pytest.fixture +def api(monkeypatch: pytest.MonkeyPatch) -> SimpleNamespace: + def create_access(descriptor: ctypes.c_void_p, apps: ctypes.c_void_p, result: Any) -> int: + ctypes.cast(result, ctypes.POINTER(ctypes.c_void_p))[0] = ctypes.c_void_p(205) + return 0 + + api = SimpleNamespace( + _found=SimpleNamespace(CFArrayCreate=Mock(return_value=204), CFDataCreate=Mock(return_value=103), CFRelease=Mock()), + _sec=SimpleNamespace(SecAccessCreate=Mock(side_effect=create_access), SecItemUpdate=Mock(return_value=0)), + OS_status=ctypes.c_int32, + error=SimpleNamespace(item_not_found=-25300), + Error=ApiError, + KeychainDenied=ApiDenied, + create_cf=Mock(side_effect=[101, 102]), + create_query=Mock(side_effect=[201, 202, 203]), + k_=Mock(return_value=ctypes.c_void_p(301)), + SecItemAdd=Mock(return_value=0), + ) + monkeypatch.setattr(macos_keychain, '_get_api', lambda: api) + return api + + +def released(api: SimpleNamespace) -> list[int]: + return [call.args[0].value for call in api._found.CFRelease.call_args_list] + + +def test_api_is_imported_lazily(monkeypatch: pytest.MonkeyPatch) -> None: + sentinel = object() + monkeypatch.setattr(macOS, 'api', sentinel, raising=False) + + assert macos_keychain._get_api() is sentinel + + +def test_update_preserves_access_controls(api: SimpleNamespace) -> None: + macos_keychain.set_password('mycli.net', 'account', 'secret') + + assert list(api.create_query.call_args_list[1].kwargs) == ['kSecValueData'] + api._found.CFArrayCreate.assert_not_called() + api._sec.SecAccessCreate.assert_not_called() + api.SecItemAdd.assert_not_called() + assert released(api) == [202, 201, 103, 102, 101] + + +def test_create_has_no_trusted_applications(api: SimpleNamespace) -> None: + api._sec.SecItemUpdate.return_value = api.error.item_not_found + + macos_keychain.set_password('mycli.net', 'account', 'secret') + + api._found.CFArrayCreate.assert_called_once_with(None, None, 0, None) + assert api._sec.SecAccessCreate.call_args.args[1].value == 204 + item = api.create_query.call_args_list[2].kwargs + assert item['kSecAttrAccess'].value == 205 + assert item['kSecAttrService'].value == 101 + assert item['kSecAttrAccount'].value == 102 + assert item['kSecValueData'].value == 103 + assert api.SecItemAdd.call_args.args[0].value == 203 + assert released(api) == [203, 205, 204, 202, 201, 103, 102, 101] + + +@pytest.mark.parametrize('password', ['secret', 'p\u00e4ss\U0001f512', 'with\0null', '']) +def test_password_is_utf8_data(api: SimpleNamespace, password: str) -> None: + macos_keychain.set_password('service', 'account', password) + + _, buffer, length = api._found.CFDataCreate.call_args.args + assert ctypes.string_at(buffer, length) == password.encode('utf-8') + assert [call.args[0] for call in api.create_cf.call_args_list] == ['service', 'account'] + + +@pytest.mark.parametrize( + ('operation', 'released_values'), + [ + ('update', [202, 201, 103, 102, 101]), + ('access', [204, 202, 201, 103, 102, 101]), + ('add', [203, 205, 204, 202, 201, 103, 102, 101]), + ], +) +@pytest.mark.parametrize(('status', 'exception'), [(-128, KeyringLocked), (-50, PasswordSetError)]) +def test_native_errors_release_resources( + api: SimpleNamespace, operation: str, released_values: list[int], status: int, exception: type[Exception] +) -> None: + api._sec.SecItemUpdate.return_value = api.error.item_not_found + if operation == 'update': + api._sec.SecItemUpdate.return_value = status + elif operation == 'access': + api._sec.SecAccessCreate.side_effect = None + api._sec.SecAccessCreate.return_value = status + else: + api.SecItemAdd.return_value = status + + with pytest.raises(exception, match="Can't store password on keychain"): + macos_keychain.set_password('service', 'account', 'secret') + + assert released(api) == released_values + if operation != 'add': + api.SecItemAdd.assert_not_called() + + +@pytest.mark.parametrize( + ('allocation', 'released_values'), + [ + ('service', []), + ('account', [101]), + ('password', [102, 101]), + ('search', [103, 102, 101]), + ('attributes', [201, 103, 102, 101]), + ('access controls', [202, 201, 103, 102, 101]), + ('access', [204, 202, 201, 103, 102, 101]), + ('item', [205, 204, 202, 201, 103, 102, 101]), + ], +) +def test_allocation_failure_is_not_an_unrestricted_write(api: SimpleNamespace, allocation: str, released_values: list[int]) -> None: + api._sec.SecItemUpdate.return_value = api.error.item_not_found + if allocation == 'service': + api.create_cf.side_effect = [None] + elif allocation == 'account': + api.create_cf.side_effect = [101, None] + elif allocation == 'password': + api._found.CFDataCreate.return_value = None + elif allocation == 'search': + api.create_query.side_effect = [None] + elif allocation == 'attributes': + api.create_query.side_effect = [201, None] + elif allocation == 'access controls': + api._found.CFArrayCreate.return_value = None + elif allocation == 'access': + api._sec.SecAccessCreate.side_effect = None + api._sec.SecAccessCreate.return_value = 0 + else: + api.create_query.side_effect = [201, 202, None] + + with pytest.raises(RuntimeError, match=f'Unable to allocate Keychain {allocation}'): + macos_keychain.set_password('service', 'account', 'secret') + + api.SecItemAdd.assert_not_called() + assert released(api) == released_values