From e5f6a2ce5d4c7503f407ca089e9959f6fd300bb9 Mon Sep 17 00:00:00 2001 From: MelvinCERBA Date: Tue, 8 Sep 2026 13:45:25 +0200 Subject: [PATCH 1/4] Avoid rewriting passwords loaded from keyring --- AUTHORS | 1 + changelog.rst | 3 +++ pgcli/main.py | 5 ++++- tests/test_main.py | 37 +++++++++++++++++++++++++++++++++++++ 4 files changed, 45 insertions(+), 1 deletion(-) diff --git a/AUTHORS b/AUTHORS index db49b9eb4..1691c3ff2 100644 --- a/AUTHORS +++ b/AUTHORS @@ -154,6 +154,7 @@ Contributors: * Diego * Chris (ChrisJr404) * Pieter Ouwerkerk (pouwerkerk) + * Melvin Cerba (melvincerba) Creator: -------- diff --git a/changelog.rst b/changelog.rst index fbdbc4917..dee412a92 100644 --- a/changelog.rst +++ b/changelog.rst @@ -22,6 +22,9 @@ Features: Bug fixes: ---------- +* Avoid rewriting passwords loaded from the keyring after every successful + connection. On macOS, rewriting recreated Keychain items and restored Python + as an application allowed to access them without confirmation. * Fix special commands being broken while explain mode (F5) is on. Every input was prefixed with ``EXPLAIN (...)`` and sent to the server as SQL, including backslash commands and the bare words ``exit``/``quit``, so ``\q``, ``\d``, diff --git a/pgcli/main.py b/pgcli/main.py index 8c172b85d..f7170d378 100644 --- a/pgcli/main.py +++ b/pgcli/main.py @@ -748,8 +748,10 @@ def connect(self, database="", host="", user="", port="", passwd="", dsn="", **k key = f"{user}@{host}@{port}" + password_loaded_from_keyring = False if not passwd and auth.keyring: passwd = auth.keyring_get_password(key) + password_loaded_from_keyring = bool(passwd) def should_ask_for_password(exc): # Prompt for a password after 1st attempt to connect @@ -847,6 +849,7 @@ def should_ask_for_password(exc): show_default=False, type=str, ) + password_loaded_from_keyring = False pgexecute = PGExecute( database, user, @@ -859,7 +862,7 @@ def should_ask_for_password(exc): ) else: raise e - if passwd and auth.keyring: + if passwd and auth.keyring and not password_loaded_from_keyring: auth.keyring_set_password(key, passwd) except Exception as e: # Connecting to a database could fail. diff --git a/tests/test_main.py b/tests/test_main.py index c8a28b419..dff38b395 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -7,6 +7,7 @@ import pytest from click.testing import CliRunner +from psycopg import OperationalError try: import setproctitle @@ -644,6 +645,42 @@ def test_pg_service_file(tmpdir): del os.environ["PGSERVICEFILE"] +def test_connect_does_not_resave_keyring_password(tmpdir): + cli = PGCli(pgclirc_file=str(tmpdir.join("rcfile"))) + + with ( + mock.patch.dict(os.environ, {"PGPASSWORD": ""}), + mock.patch("pgcli.main.auth.keyring", True), + mock.patch("pgcli.main.auth.keyring_get_password", return_value="keyring-password") as get_password, + mock.patch("pgcli.main.auth.keyring_set_password") as set_password, + mock.patch("pgcli.main.PGExecute") as pgexecute, + ): + cli.connect(database="test", host="localhost", user="postgres", port=5432) + + get_password.assert_called_once_with("postgres@localhost@5432") + assert pgexecute.call_args.args[2] == "keyring-password" + set_password.assert_not_called() + + +def test_connect_saves_replacement_for_invalid_keyring_password(tmpdir): + cli = PGCli(pgclirc_file=str(tmpdir.join("rcfile"))) + + with ( + mock.patch.dict(os.environ, {"PGPASSWORD": ""}), + mock.patch("pgcli.main.auth.keyring", True), + mock.patch("pgcli.main.auth.keyring_get_password", return_value="old-password"), + mock.patch("pgcli.main.auth.keyring_set_password") as set_password, + mock.patch("pgcli.main.click.prompt", return_value="new-password"), + mock.patch( + "pgcli.main.PGExecute", + side_effect=[OperationalError("password authentication failed"), mock.Mock()], + ), + ): + cli.connect(database="test", host="localhost", user="postgres", port=5432) + + set_password.assert_called_once_with("postgres@localhost@5432", "new-password") + + def test_ssl_db_uri(tmpdir): with mock.patch.object(PGCli, "connect") as mock_connect: cli = PGCli(pgclirc_file=str(tmpdir.join("rcfile"))) From 51bfbe4e8364da6f64e6e6e8179d2bab03491862 Mon Sep 17 00:00:00 2001 From: MelvinCERBA Date: Tue, 8 Sep 2026 14:10:33 +0200 Subject: [PATCH 2/4] Correct contributor GitHub username --- AUTHORS | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/AUTHORS b/AUTHORS index 1691c3ff2..88fecdf64 100644 --- a/AUTHORS +++ b/AUTHORS @@ -154,7 +154,7 @@ Contributors: * Diego * Chris (ChrisJr404) * Pieter Ouwerkerk (pouwerkerk) - * Melvin Cerba (melvincerba) + * Melvin Cerba (MelvinCERBA) Creator: -------- From 451b09ebabbac3d6a6e64cb228d94edb3a17b894 Mon Sep 17 00:00:00 2001 From: MelvinCERBA Date: Mon, 14 Sep 2026 10:06:06 +0200 Subject: [PATCH 3/4] Preserve macOS Keychain access choices --- changelog.rst | 6 +- pgcli/auth.py | 43 ++++++++- pgcli/macos_keychain.py | 166 +++++++++++++++++++++++++++++++++++ tests/test_auth.py | 128 +++++++++++++++++++++++++-- tests/test_macos_keychain.py | 86 ++++++++++++++++++ 5 files changed, 419 insertions(+), 10 deletions(-) create mode 100644 pgcli/macos_keychain.py create mode 100644 tests/test_macos_keychain.py diff --git a/changelog.rst b/changelog.rst index dee412a92..78283e3ca 100644 --- a/changelog.rst +++ b/changelog.rst @@ -22,9 +22,9 @@ Features: Bug fixes: ---------- -* Avoid rewriting passwords loaded from the keyring after every successful - connection. On macOS, rewriting recreated Keychain items and restored Python - as an application allowed to access them without confirmation. +* Preserve macOS Keychain access choices. New credentials no longer + pre-authorize Python, and credentials loaded from the keyring are not + rewritten after successful connections. * Fix special commands being broken while explain mode (F5) is on. Every input was prefixed with ``EXPLAIN (...)`` and sent to the server as SQL, including backslash commands and the bare words ``exit``/``quit``, so ``\q``, ``\d``, diff --git a/pgcli/auth.py b/pgcli/auth.py index 513097a31..b91acf7cd 100644 --- a/pgcli/auth.py +++ b/pgcli/auth.py @@ -1,4 +1,5 @@ import click +import platform from textwrap import dedent @@ -45,9 +46,49 @@ def keyring_get_password(key): return passwd +def _is_macos_keyring_backend(backend): + return backend.__class__.__module__ in { + "keyring.backends.macOS", + "keyring.backends.OS_X", + } + + +def _macos_keyring_uses_keychain_path(backend): + if backend.__class__.__module__ == "keyring.backends.OS_X": + return True + + import importlib + + backend_module = importlib.import_module(backend.__class__.__module__) + return hasattr(backend_module.api, "SecKeychainCopyDefault") + + +def _set_password_with_backend(backend, key, passwd): + if _is_macos_keyring_backend(backend): + from pgcli import macos_keychain + + if _macos_keyring_uses_keychain_path(backend): + macos_keychain.set_password("pgcli", key, passwd, backend.keychain) + else: + macos_keychain.set_password("pgcli", key, passwd) + else: + backend.set_password("pgcli", key, passwd) + + def keyring_set_password(key, passwd): try: - keyring.set_password("pgcli", key, passwd) + configured_backend = keyring.get_keyring() if platform.system() == "Darwin" else None + if configured_backend is not None and configured_backend.__class__.__module__ == "keyring.backends.chainer": + for backend in configured_backend.backends: + try: + _set_password_with_backend(backend, key, passwd) + break + except NotImplementedError: + pass + elif configured_backend is not None and _is_macos_keyring_backend(configured_backend): + _set_password_with_backend(configured_backend, key, passwd) + else: + keyring.set_password("pgcli", key, passwd) except Exception as e: click.secho( keyring_error_message.format("Set password in keyring returned:", str(e)), diff --git a/pgcli/macos_keychain.py b/pgcli/macos_keychain.py new file mode 100644 index 000000000..0cee3c777 --- /dev/null +++ b/pgcli/macos_keychain.py @@ -0,0 +1,166 @@ +import ctypes +from ctypes.util import find_library + + +_ITEM_NOT_FOUND = -25300 +_UTF8 = 0x08000100 +_UNSCOPED = object() + + +def _load_framework(name): + path = find_library(name) + if not path: + raise RuntimeError(f"Unable to load the macOS {name} framework") + return ctypes.CDLL(path) + + +def _constant(framework, name): + return ctypes.c_void_p.in_dll(framework, name) + + +def _check_status(status): + if status != 0: + raise RuntimeError(f"macOS Keychain returned status {status}") + + +def set_password(service, account, password, keychain_path=_UNSCOPED): + """Store a password without pre-authorizing the creating executable.""" + security = _load_framework("Security") + core_foundation = _load_framework("CoreFoundation") + + cf_string_create = core_foundation.CFStringCreateWithCString + cf_string_create.restype = ctypes.c_void_p + cf_string_create.argtypes = (ctypes.c_void_p, ctypes.c_char_p, ctypes.c_uint32) + + cf_data_create = core_foundation.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_array_create = core_foundation.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_dictionary_create = core_foundation.CFDictionaryCreate + cf_dictionary_create.restype = ctypes.c_void_p + cf_dictionary_create.argtypes = ( + ctypes.c_void_p, + ctypes.c_void_p, + ctypes.c_void_p, + ctypes.c_long, + ctypes.c_void_p, + ctypes.c_void_p, + ) + + cf_release = core_foundation.CFRelease + cf_release.argtypes = (ctypes.c_void_p,) + + sec_access_create = security.SecAccessCreate + sec_access_create.restype = ctypes.c_int32 + sec_access_create.argtypes = (ctypes.c_void_p, ctypes.c_void_p, ctypes.POINTER(ctypes.c_void_p)) + + sec_item_add = security.SecItemAdd + sec_item_add.restype = ctypes.c_int32 + sec_item_add.argtypes = (ctypes.c_void_p, ctypes.c_void_p) + + sec_item_update = security.SecItemUpdate + sec_item_update.restype = ctypes.c_int32 + sec_item_update.argtypes = (ctypes.c_void_p, ctypes.c_void_p) + + retained = [] + + def create_string(value): + result = cf_string_create(None, value.encode("utf-8"), _UTF8) + if not result: + raise RuntimeError("Unable to allocate a Keychain string") + retained.append(result) + return result + + def create_dictionary(entries): + keys = (ctypes.c_void_p * len(entries))(*(key for key, value in entries)) + values = (ctypes.c_void_p * len(entries))(*(value for key, value in entries)) + result = cf_dictionary_create( + None, + keys, + values, + len(entries), + core_foundation.kCFTypeDictionaryKeyCallBacks, + core_foundation.kCFTypeDictionaryValueCallBacks, + ) + if not result: + raise RuntimeError("Unable to allocate Keychain attributes") + retained.append(result) + return result + + try: + service_value = create_string(service) + account_value = create_string(account) + password_bytes = password.encode("utf-8") + password_buffer = ctypes.create_string_buffer(password_bytes) + password_value = cf_data_create(None, password_buffer, len(password_bytes)) + if not password_value: + raise RuntimeError("Unable to allocate Keychain password data") + retained.append(password_value) + + identity_entries = [ + (_constant(security, "kSecClass"), _constant(security, "kSecClassGenericPassword")), + (_constant(security, "kSecAttrService"), service_value), + (_constant(security, "kSecAttrAccount"), account_value), + ] + search_entries = identity_entries.copy() + add_entries = identity_entries.copy() + + if keychain_path is not _UNSCOPED: + keychain = ctypes.c_void_p() + if keychain_path is None: + sec_keychain_copy_default = security.SecKeychainCopyDefault + sec_keychain_copy_default.restype = ctypes.c_int32 + sec_keychain_copy_default.argtypes = (ctypes.POINTER(ctypes.c_void_p),) + status = sec_keychain_copy_default(ctypes.byref(keychain)) + else: + sec_keychain_open = security.SecKeychainOpen + sec_keychain_open.restype = ctypes.c_int32 + sec_keychain_open.argtypes = (ctypes.c_char_p, ctypes.POINTER(ctypes.c_void_p)) + status = sec_keychain_open(keychain_path.encode(), ctypes.byref(keychain)) + _check_status(status) + retained.append(keychain) + + keychain_values = (ctypes.c_void_p * 1)(keychain) + search_list = cf_array_create( + None, + keychain_values, + 1, + core_foundation.kCFTypeArrayCallBacks, + ) + if not search_list: + raise RuntimeError("Unable to allocate a Keychain search list") + retained.append(search_list) + + search_entries.append((_constant(security, "kSecMatchSearchList"), search_list)) + add_entries.append((_constant(security, "kSecUseKeychain"), keychain)) + + search = create_dictionary(search_entries) + attributes = create_dictionary([(_constant(security, "kSecValueData"), password_value)]) + + status = sec_item_update(search, attributes) + if status == _ITEM_NOT_FOUND: + trusted_apps = cf_array_create(None, None, 0, None) + if not trusted_apps: + raise RuntimeError("Unable to allocate Keychain access controls") + retained.append(trusted_apps) + + access = ctypes.c_void_p() + _check_status(sec_access_create(service_value, trusted_apps, ctypes.byref(access))) + retained.append(access) + + item = create_dictionary( + add_entries + + [ + (_constant(security, "kSecValueData"), password_value), + (_constant(security, "kSecAttrAccess"), access), + ] + ) + status = sec_item_add(item, None) + _check_status(status) + finally: + for value in reversed(retained): + cf_release(value) diff --git a/tests/test_auth.py b/tests/test_auth.py index 13eed58db..99e62e059 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -25,12 +25,128 @@ def test_keyring_get_password_exception(): def test_keyring_set_password_ok(): - with mock.patch("pgcli.auth.keyring", return_value=mock.MagicMock()): - with mock.patch("pgcli.auth.keyring.set_password"): - auth.keyring_set_password("test", "abc123") + with mock.patch("pgcli.auth.platform.system", return_value="Linux"): + with mock.patch("pgcli.auth.keyring", return_value=mock.MagicMock()): + with mock.patch("pgcli.auth.keyring.set_password"): + auth.keyring_set_password("test", "abc123") + + +@pytest.mark.parametrize("chained", [False, True]) +def test_keyring_set_password_macos_has_no_trusted_apps(chained): + backend_class = type("Keyring", (), {"__module__": "keyring.backends.macOS"}) + backend = backend_class() + if chained: + chainer_class = type("ChainerBackend", (), {"__module__": "keyring.backends.chainer"}) + configured_backend = chainer_class() + configured_backend.backends = [backend] + else: + configured_backend = backend + keyring = mock.Mock() + keyring.get_keyring.return_value = configured_backend + + with ( + mock.patch("pgcli.auth.platform.system", return_value="Darwin"), + mock.patch("pgcli.auth.keyring", keyring), + mock.patch("pgcli.auth._macos_keyring_uses_keychain_path", return_value=False), + mock.patch("pgcli.macos_keychain.set_password") as set_password, + ): + auth.keyring_set_password("test", "abc123") + + set_password.assert_called_once_with("pgcli", "test", "abc123") + keyring.set_password.assert_not_called() + + +def test_keyring_set_password_macos_custom_backend(): + native_backend_class = type("Keyring", (), {"__module__": "keyring.backends.macOS"}) + backend_class = type("CompositeKeyring", (), {"__module__": "custom.keyring"}) + backend = backend_class() + backend.backends = [native_backend_class()] + keyring = mock.Mock() + keyring.get_keyring.return_value = backend + + with ( + mock.patch("pgcli.auth.platform.system", return_value="Darwin"), + mock.patch("pgcli.auth.keyring", keyring), + mock.patch("pgcli.macos_keychain.set_password") as set_password, + ): + auth.keyring_set_password("test", "abc123") + + keyring.set_password.assert_called_once_with("pgcli", "test", "abc123") + set_password.assert_not_called() + + +def test_keyring_set_password_macos_after_read_only_chained_backend(): + read_only_backend = mock.Mock() + read_only_backend.set_password.side_effect = NotImplementedError + native_backend_class = type("Keyring", (), {"__module__": "keyring.backends.macOS"}) + native_backend = native_backend_class() + chainer_class = type("ChainerBackend", (), {"__module__": "keyring.backends.chainer"}) + configured_backend = chainer_class() + configured_backend.backends = [read_only_backend, native_backend] + keyring = mock.Mock() + keyring.get_keyring.return_value = configured_backend + + with ( + mock.patch("pgcli.auth.platform.system", return_value="Darwin"), + mock.patch("pgcli.auth.keyring", keyring), + mock.patch("pgcli.auth._macos_keyring_uses_keychain_path", return_value=False), + mock.patch("pgcli.macos_keychain.set_password") as set_password, + ): + auth.keyring_set_password("test", "abc123") + + read_only_backend.set_password.assert_called_once_with("pgcli", "test", "abc123") + set_password.assert_called_once_with("pgcli", "test", "abc123") + + +@pytest.mark.parametrize("keychain_path", [None, "/tmp/test.keychain"]) +def test_keyring_set_password_legacy_macos_preserves_keychain(keychain_path): + backend_class = type("Keyring", (), {"__module__": "keyring.backends.OS_X"}) + backend = backend_class() + backend.keychain = keychain_path + keyring = mock.Mock() + keyring.get_keyring.return_value = backend + + with ( + mock.patch("pgcli.auth.platform.system", return_value="Darwin"), + mock.patch("pgcli.auth.keyring", keyring), + mock.patch("pgcli.macos_keychain.set_password") as set_password, + ): + auth.keyring_set_password("test", "abc123") + + set_password.assert_called_once_with("pgcli", "test", "abc123", keychain_path) + + +def test_keyring_set_password_transitional_macos_preserves_keychain(): + backend_class = type("Keyring", (), {"__module__": "keyring.backends.macOS"}) + backend = backend_class() + backend.keychain = "/tmp/test.keychain" + keyring = mock.Mock() + keyring.get_keyring.return_value = backend + + with ( + mock.patch("pgcli.auth.platform.system", return_value="Darwin"), + mock.patch("pgcli.auth.keyring", keyring), + mock.patch("pgcli.auth._macos_keyring_uses_keychain_path", return_value=True), + mock.patch("pgcli.macos_keychain.set_password") as set_password, + ): + auth.keyring_set_password("test", "abc123") + + set_password.assert_called_once_with("pgcli", "test", "abc123", "/tmp/test.keychain") + + +@pytest.mark.parametrize("has_scoped_api", [False, True]) +def test_macos_keyring_path_capability_detection(has_scoped_api): + backend_class = type("Keyring", (), {"__module__": "keyring.backends.macOS"}) + backend_api = object() + if has_scoped_api: + backend_api = mock.Mock(SecKeychainCopyDefault=mock.Mock()) + + with mock.patch("importlib.import_module", return_value=mock.Mock(api=backend_api)): + assert auth._macos_keyring_uses_keychain_path(backend_class()) is has_scoped_api def test_keyring_set_password_exception(): - with mock.patch("pgcli.auth.keyring", return_value=mock.MagicMock()): - with mock.patch("pgcli.auth.keyring.set_password", side_effect=Exception("Boom!")): - auth.keyring_set_password("test", "abc123") + with mock.patch("pgcli.auth.platform.system", return_value="Linux"): + with mock.patch("pgcli.auth.keyring", return_value=mock.MagicMock()): + with mock.patch("pgcli.auth.keyring.set_password", side_effect=Exception("Boom!")): + auth.keyring_set_password("test", "abc123") diff --git a/tests/test_macos_keychain.py b/tests/test_macos_keychain.py new file mode 100644 index 000000000..8d3a46df4 --- /dev/null +++ b/tests/test_macos_keychain.py @@ -0,0 +1,86 @@ +import ctypes +from unittest import mock + +import pytest + +from pgcli import macos_keychain + + +def test_set_password_preserves_access_list_when_updating(): + security = mock.MagicMock() + core_foundation = mock.MagicMock() + + core_foundation.CFStringCreateWithCString.side_effect = [101, 102] + core_foundation.CFDataCreate.return_value = 103 + core_foundation.CFDictionaryCreate.side_effect = [201, 202] + security.SecItemUpdate.return_value = 0 + + constants = (ctypes.c_void_p(value) for value in range(301, 310)) + with ( + mock.patch("pgcli.macos_keychain._load_framework", side_effect=[security, core_foundation]), + mock.patch("pgcli.macos_keychain._constant", side_effect=constants), + ): + macos_keychain.set_password("pgcli", "user@host@5432", "secret") + + core_foundation.CFArrayCreate.assert_not_called() + security.SecAccessCreate.assert_not_called() + security.SecItemAdd.assert_not_called() + + +def test_set_password_uses_empty_access_list_when_creating(): + security = mock.MagicMock() + core_foundation = mock.MagicMock() + + core_foundation.CFStringCreateWithCString.side_effect = [101, 102] + core_foundation.CFDataCreate.return_value = 103 + core_foundation.CFArrayCreate.return_value = 104 + core_foundation.CFDictionaryCreate.side_effect = [201, 202, 203] + security.SecAccessCreate.return_value = 0 + security.SecItemUpdate.return_value = macos_keychain._ITEM_NOT_FOUND + security.SecItemAdd.return_value = 0 + + constants = (ctypes.c_void_p(value) for value in range(301, 310)) + with ( + mock.patch("pgcli.macos_keychain._load_framework", side_effect=[security, core_foundation]), + mock.patch("pgcli.macos_keychain._constant", side_effect=constants), + ): + macos_keychain.set_password("pgcli", "user@host@5432", "secret") + + core_foundation.CFArrayCreate.assert_called_once_with(None, None, 0, None) + security.SecAccessCreate.assert_called_once() + security.SecItemAdd.assert_called_once() + + +@pytest.mark.parametrize("keychain_path", [None, "/tmp/test.keychain"]) +def test_set_password_scopes_legacy_operations_to_keychain(keychain_path): + security = mock.MagicMock() + core_foundation = mock.MagicMock() + + core_foundation.CFStringCreateWithCString.side_effect = [101, 102] + core_foundation.CFDataCreate.return_value = 103 + core_foundation.CFArrayCreate.side_effect = [104, 105] + core_foundation.CFDictionaryCreate.side_effect = [201, 202, 203] + security.SecAccessCreate.return_value = 0 + security.SecKeychainCopyDefault.return_value = 0 + security.SecKeychainOpen.return_value = 0 + security.SecItemUpdate.return_value = macos_keychain._ITEM_NOT_FOUND + security.SecItemAdd.return_value = 0 + + constants = (ctypes.c_void_p(value) for value in range(301, 312)) + with ( + mock.patch("pgcli.macos_keychain._load_framework", side_effect=[security, core_foundation]), + mock.patch("pgcli.macos_keychain._constant", side_effect=constants), + ): + macos_keychain.set_password("pgcli", "user@host@5432", "secret", keychain_path) + + if keychain_path is None: + security.SecKeychainCopyDefault.assert_called_once() + security.SecKeychainOpen.assert_not_called() + else: + security.SecKeychainOpen.assert_called_once() + security.SecKeychainCopyDefault.assert_not_called() + assert core_foundation.CFArrayCreate.call_args_list[0].args[2:] == ( + 1, + core_foundation.kCFTypeArrayCallBacks, + ) + assert core_foundation.CFArrayCreate.call_args_list[1] == mock.call(None, None, 0, None) From bc44b6f0a87cefb58cd9902b27e29e9f982cdfbb Mon Sep 17 00:00:00 2001 From: MelvinCERBA Date: Mon, 14 Sep 2026 10:59:06 +0200 Subject: [PATCH 4/4] Reuse keyring macOS API --- pgcli/auth.py | 20 +--- pgcli/macos_keychain.py | 191 ++++++++++------------------------- pyproject.toml | 4 +- tests/test_auth.py | 49 --------- tests/test_macos_keychain.py | 157 ++++++++++++++++------------ 5 files changed, 151 insertions(+), 270 deletions(-) diff --git a/pgcli/auth.py b/pgcli/auth.py index b91acf7cd..8850b2831 100644 --- a/pgcli/auth.py +++ b/pgcli/auth.py @@ -47,30 +47,14 @@ def keyring_get_password(key): def _is_macos_keyring_backend(backend): - return backend.__class__.__module__ in { - "keyring.backends.macOS", - "keyring.backends.OS_X", - } - - -def _macos_keyring_uses_keychain_path(backend): - if backend.__class__.__module__ == "keyring.backends.OS_X": - return True - - import importlib - - backend_module = importlib.import_module(backend.__class__.__module__) - return hasattr(backend_module.api, "SecKeychainCopyDefault") + return backend.__class__.__module__ == "keyring.backends.macOS" def _set_password_with_backend(backend, key, passwd): if _is_macos_keyring_backend(backend): from pgcli import macos_keychain - if _macos_keyring_uses_keychain_path(backend): - macos_keychain.set_password("pgcli", key, passwd, backend.keychain) - else: - macos_keychain.set_password("pgcli", key, passwd) + macos_keychain.set_password("pgcli", key, passwd) else: backend.set_password("pgcli", key, passwd) diff --git a/pgcli/macos_keychain.py b/pgcli/macos_keychain.py index 0cee3c777..a72bd856c 100644 --- a/pgcli/macos_keychain.py +++ b/pgcli/macos_keychain.py @@ -1,166 +1,87 @@ import ctypes -from ctypes.util import find_library -_ITEM_NOT_FOUND = -25300 -_UTF8 = 0x08000100 -_UNSCOPED = object() +def _get_api(): + from keyring.backends.macOS import api + return api -def _load_framework(name): - path = find_library(name) - if not path: - raise RuntimeError(f"Unable to load the macOS {name} framework") - return ctypes.CDLL(path) +def _create_string(api, value): + create_string = getattr(api, "create_cfstr", None) or api.create_cf + result = create_string(value) + return result if isinstance(result, ctypes.c_void_p) else ctypes.c_void_p(result) -def _constant(framework, name): - return ctypes.c_void_p.in_dll(framework, name) - -def _check_status(status): - if status != 0: - raise RuntimeError(f"macOS Keychain returned status {status}") - - -def set_password(service, account, password, keychain_path=_UNSCOPED): +def set_password(service, account, password): """Store a password without pre-authorizing the creating executable.""" - security = _load_framework("Security") - core_foundation = _load_framework("CoreFoundation") - - cf_string_create = core_foundation.CFStringCreateWithCString - cf_string_create.restype = ctypes.c_void_p - cf_string_create.argtypes = (ctypes.c_void_p, ctypes.c_char_p, ctypes.c_uint32) + api = _get_api() - cf_data_create = core_foundation.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_array_create = core_foundation.CFArrayCreate + 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_dictionary_create = core_foundation.CFDictionaryCreate - cf_dictionary_create.restype = ctypes.c_void_p - cf_dictionary_create.argtypes = ( - ctypes.c_void_p, - ctypes.c_void_p, - ctypes.c_void_p, - ctypes.c_long, - ctypes.c_void_p, - ctypes.c_void_p, - ) - - cf_release = core_foundation.CFRelease + cf_release = api._found.CFRelease + cf_release.restype = None cf_release.argtypes = (ctypes.c_void_p,) - sec_access_create = security.SecAccessCreate - sec_access_create.restype = ctypes.c_int32 + 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_add = security.SecItemAdd - sec_item_add.restype = ctypes.c_int32 - sec_item_add.argtypes = (ctypes.c_void_p, ctypes.c_void_p) - - sec_item_update = security.SecItemUpdate - sec_item_update.restype = ctypes.c_int32 + 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 = [] - def create_string(value): - result = cf_string_create(None, value.encode("utf-8"), _UTF8) - if not result: - raise RuntimeError("Unable to allocate a Keychain string") - retained.append(result) - return result - - def create_dictionary(entries): - keys = (ctypes.c_void_p * len(entries))(*(key for key, value in entries)) - values = (ctypes.c_void_p * len(entries))(*(value for key, value in entries)) - result = cf_dictionary_create( - None, - keys, - values, - len(entries), - core_foundation.kCFTypeDictionaryKeyCallBacks, - core_foundation.kCFTypeDictionaryValueCallBacks, - ) - if not result: - raise RuntimeError("Unable to allocate Keychain attributes") - retained.append(result) - return result + def retain(value, error): + if not value: + raise RuntimeError(error) + retained.append(value) + return value try: - service_value = create_string(service) - account_value = create_string(account) - password_bytes = password.encode("utf-8") - password_buffer = ctypes.create_string_buffer(password_bytes) - password_value = cf_data_create(None, password_buffer, len(password_bytes)) - if not password_value: - raise RuntimeError("Unable to allocate Keychain password data") - retained.append(password_value) - - identity_entries = [ - (_constant(security, "kSecClass"), _constant(security, "kSecClassGenericPassword")), - (_constant(security, "kSecAttrService"), service_value), - (_constant(security, "kSecAttrAccount"), account_value), - ] - search_entries = identity_entries.copy() - add_entries = identity_entries.copy() - - if keychain_path is not _UNSCOPED: - keychain = ctypes.c_void_p() - if keychain_path is None: - sec_keychain_copy_default = security.SecKeychainCopyDefault - sec_keychain_copy_default.restype = ctypes.c_int32 - sec_keychain_copy_default.argtypes = (ctypes.POINTER(ctypes.c_void_p),) - status = sec_keychain_copy_default(ctypes.byref(keychain)) - else: - sec_keychain_open = security.SecKeychainOpen - sec_keychain_open.restype = ctypes.c_int32 - sec_keychain_open.argtypes = (ctypes.c_char_p, ctypes.POINTER(ctypes.c_void_p)) - status = sec_keychain_open(keychain_path.encode(), ctypes.byref(keychain)) - _check_status(status) - retained.append(keychain) - - keychain_values = (ctypes.c_void_p * 1)(keychain) - search_list = cf_array_create( - None, - keychain_values, - 1, - core_foundation.kCFTypeArrayCallBacks, - ) - if not search_list: - raise RuntimeError("Unable to allocate a Keychain search list") - retained.append(search_list) - - search_entries.append((_constant(security, "kSecMatchSearchList"), search_list)) - add_entries.append((_constant(security, "kSecUseKeychain"), keychain)) - - search = create_dictionary(search_entries) - attributes = create_dictionary([(_constant(security, "kSecValueData"), password_value)]) + service_value = retain(_create_string(api, service), "Unable to allocate Keychain service") + account_value = retain(_create_string(api, account), "Unable to allocate Keychain account") + password_value = retain(_create_string(api, password), "Unable to allocate Keychain password") + + search = retain( + api.create_query( + kSecClass=api.k_("kSecClassGenericPassword"), + kSecAttrService=service_value, + kSecAttrAccount=account_value, + ), + "Unable to allocate Keychain search", + ) + attributes = retain( + api.create_query(kSecValueData=password_value), + "Unable to allocate Keychain attributes", + ) status = sec_item_update(search, attributes) - if status == _ITEM_NOT_FOUND: - trusted_apps = cf_array_create(None, None, 0, None) - if not trusted_apps: - raise RuntimeError("Unable to allocate Keychain access controls") - retained.append(trusted_apps) + if status == api.error.item_not_found: + trusted_apps = retain( + cf_array_create(None, None, 0, None), + "Unable to allocate Keychain access controls", + ) access = ctypes.c_void_p() - _check_status(sec_access_create(service_value, trusted_apps, ctypes.byref(access))) - retained.append(access) - - item = create_dictionary( - add_entries - + [ - (_constant(security, "kSecValueData"), password_value), - (_constant(security, "kSecAttrAccess"), access), - ] + api.Error.raise_for_status(sec_access_create(service_value, trusted_apps, ctypes.byref(access))) + retain(access, "Unable to allocate Keychain access") + + item = retain( + api.create_query( + kSecClass=api.k_("kSecClassGenericPassword"), + kSecAttrService=service_value, + kSecAttrAccount=account_value, + kSecValueData=password_value, + kSecAttrAccess=access, + ), + "Unable to allocate Keychain item", ) - status = sec_item_add(item, None) - _check_status(status) + status = api.SecItemAdd(item, None) + api.Error.raise_for_status(status) finally: for value in reversed(retained): cf_release(value) diff --git a/pyproject.toml b/pyproject.toml index 452617496..0e49d935f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,7 +49,7 @@ dynamic = ["version"] pgcli = "pgcli.main:cli" [project.optional-dependencies] -keyring = ["keyring >= 12.2.0"] +keyring = ["keyring >= 23.1.0, < 26"] sshtunnel = ["sshtunnel >= 0.4.0"] dev = [ "behave>=1.2.4", @@ -130,4 +130,4 @@ exclude = [ [tool.pytest.ini_options] minversion = "6.0" addopts = "--capture=sys --showlocals -rxs" -testpaths = ["tests"] \ No newline at end of file +testpaths = ["tests"] diff --git a/tests/test_auth.py b/tests/test_auth.py index 99e62e059..010de0eef 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -47,7 +47,6 @@ def test_keyring_set_password_macos_has_no_trusted_apps(chained): with ( mock.patch("pgcli.auth.platform.system", return_value="Darwin"), mock.patch("pgcli.auth.keyring", keyring), - mock.patch("pgcli.auth._macos_keyring_uses_keychain_path", return_value=False), mock.patch("pgcli.macos_keychain.set_password") as set_password, ): auth.keyring_set_password("test", "abc123") @@ -89,7 +88,6 @@ def test_keyring_set_password_macos_after_read_only_chained_backend(): with ( mock.patch("pgcli.auth.platform.system", return_value="Darwin"), mock.patch("pgcli.auth.keyring", keyring), - mock.patch("pgcli.auth._macos_keyring_uses_keychain_path", return_value=False), mock.patch("pgcli.macos_keychain.set_password") as set_password, ): auth.keyring_set_password("test", "abc123") @@ -98,53 +96,6 @@ def test_keyring_set_password_macos_after_read_only_chained_backend(): set_password.assert_called_once_with("pgcli", "test", "abc123") -@pytest.mark.parametrize("keychain_path", [None, "/tmp/test.keychain"]) -def test_keyring_set_password_legacy_macos_preserves_keychain(keychain_path): - backend_class = type("Keyring", (), {"__module__": "keyring.backends.OS_X"}) - backend = backend_class() - backend.keychain = keychain_path - keyring = mock.Mock() - keyring.get_keyring.return_value = backend - - with ( - mock.patch("pgcli.auth.platform.system", return_value="Darwin"), - mock.patch("pgcli.auth.keyring", keyring), - mock.patch("pgcli.macos_keychain.set_password") as set_password, - ): - auth.keyring_set_password("test", "abc123") - - set_password.assert_called_once_with("pgcli", "test", "abc123", keychain_path) - - -def test_keyring_set_password_transitional_macos_preserves_keychain(): - backend_class = type("Keyring", (), {"__module__": "keyring.backends.macOS"}) - backend = backend_class() - backend.keychain = "/tmp/test.keychain" - keyring = mock.Mock() - keyring.get_keyring.return_value = backend - - with ( - mock.patch("pgcli.auth.platform.system", return_value="Darwin"), - mock.patch("pgcli.auth.keyring", keyring), - mock.patch("pgcli.auth._macos_keyring_uses_keychain_path", return_value=True), - mock.patch("pgcli.macos_keychain.set_password") as set_password, - ): - auth.keyring_set_password("test", "abc123") - - set_password.assert_called_once_with("pgcli", "test", "abc123", "/tmp/test.keychain") - - -@pytest.mark.parametrize("has_scoped_api", [False, True]) -def test_macos_keyring_path_capability_detection(has_scoped_api): - backend_class = type("Keyring", (), {"__module__": "keyring.backends.macOS"}) - backend_api = object() - if has_scoped_api: - backend_api = mock.Mock(SecKeychainCopyDefault=mock.Mock()) - - with mock.patch("importlib.import_module", return_value=mock.Mock(api=backend_api)): - assert auth._macos_keyring_uses_keychain_path(backend_class()) is has_scoped_api - - def test_keyring_set_password_exception(): with mock.patch("pgcli.auth.platform.system", return_value="Linux"): with mock.patch("pgcli.auth.keyring", return_value=mock.MagicMock()): diff --git a/tests/test_macos_keychain.py b/tests/test_macos_keychain.py index 8d3a46df4..1137d2db9 100644 --- a/tests/test_macos_keychain.py +++ b/tests/test_macos_keychain.py @@ -1,4 +1,5 @@ import ctypes +from types import SimpleNamespace from unittest import mock import pytest @@ -6,81 +7,105 @@ from pgcli import macos_keychain -def test_set_password_preserves_access_list_when_updating(): +def keyring_api(update_status): + found = mock.MagicMock() security = mock.MagicMock() - core_foundation = mock.MagicMock() - - core_foundation.CFStringCreateWithCString.side_effect = [101, 102] - core_foundation.CFDataCreate.return_value = 103 - core_foundation.CFDictionaryCreate.side_effect = [201, 202] - security.SecItemUpdate.return_value = 0 - - constants = (ctypes.c_void_p(value) for value in range(301, 310)) - with ( - mock.patch("pgcli.macos_keychain._load_framework", side_effect=[security, core_foundation]), - mock.patch("pgcli.macos_keychain._constant", side_effect=constants), - ): - macos_keychain.set_password("pgcli", "user@host@5432", "secret") + error = SimpleNamespace(item_not_found=-25300) + api = SimpleNamespace( + _found=found, + _sec=security, + OS_status=ctypes.c_int32, + error=error, + Error=mock.MagicMock(), + create_cf=mock.Mock(side_effect=[101, 102, 103]), + create_query=mock.Mock(side_effect=[201, 202, 203]), + k_=mock.Mock(return_value=ctypes.c_void_p(301)), + SecItemAdd=mock.Mock(return_value=0), + ) + found.CFArrayCreate.return_value = 204 - core_foundation.CFArrayCreate.assert_not_called() - security.SecAccessCreate.assert_not_called() - security.SecItemAdd.assert_not_called() + def create_access(descriptor, trusted_apps, result): + ctypes.cast(result, ctypes.POINTER(ctypes.c_void_p))[0] = ctypes.c_void_p(205) + return 0 + security.SecAccessCreate.side_effect = create_access + security.SecItemUpdate.return_value = update_status + return api -def test_set_password_uses_empty_access_list_when_creating(): - security = mock.MagicMock() - core_foundation = mock.MagicMock() - core_foundation.CFStringCreateWithCString.side_effect = [101, 102] - core_foundation.CFDataCreate.return_value = 103 - core_foundation.CFArrayCreate.return_value = 104 - core_foundation.CFDictionaryCreate.side_effect = [201, 202, 203] - security.SecAccessCreate.return_value = 0 - security.SecItemUpdate.return_value = macos_keychain._ITEM_NOT_FOUND - security.SecItemAdd.return_value = 0 +def test_set_password_preserves_access_list_when_updating(): + api = keyring_api(update_status=0) - constants = (ctypes.c_void_p(value) for value in range(301, 310)) - with ( - mock.patch("pgcli.macos_keychain._load_framework", side_effect=[security, core_foundation]), - mock.patch("pgcli.macos_keychain._constant", side_effect=constants), - ): + with mock.patch("pgcli.macos_keychain._get_api", return_value=api): macos_keychain.set_password("pgcli", "user@host@5432", "secret") - core_foundation.CFArrayCreate.assert_called_once_with(None, None, 0, None) - security.SecAccessCreate.assert_called_once() - security.SecItemAdd.assert_called_once() + api._found.CFArrayCreate.assert_not_called() + api._sec.SecAccessCreate.assert_not_called() + api.SecItemAdd.assert_not_called() + api.Error.raise_for_status.assert_called_once_with(0) + assert [ + call.args[0].value if isinstance(call.args[0], ctypes.c_void_p) else call.args[0] for call in api._found.CFRelease.call_args_list + ] == [ + 202, + 201, + 103, + 102, + 101, + ] + + +@pytest.mark.parametrize("create_string", ["create_cfstr", "create_cf"]) +def test_set_password_uses_empty_access_list_when_creating(create_string): + api_item_not_found = -25300 + api = keyring_api(update_status=api_item_not_found) + if create_string == "create_cfstr": + api.create_cfstr = mock.Mock(side_effect=[101, 102, 103]) + + with mock.patch("pgcli.macos_keychain._get_api", return_value=api): + macos_keychain.set_password("pgcli", "user@host@5432", "secret") + assert api.error.item_not_found == api_item_not_found + api._found.CFArrayCreate.assert_called_once_with(None, None, 0, None) + api._sec.SecAccessCreate.assert_called_once() + api.SecItemAdd.assert_called_once_with(203, None) + assert getattr(api, create_string).call_args_list == [ + mock.call("pgcli"), + mock.call("user@host@5432"), + mock.call("secret"), + ] + assert api.create_query.call_args_list == [ + mock.call( + kSecClass=api.k_.return_value, + kSecAttrService=mock.ANY, + kSecAttrAccount=mock.ANY, + ), + mock.call(kSecValueData=mock.ANY), + mock.call( + kSecClass=api.k_.return_value, + kSecAttrService=mock.ANY, + kSecAttrAccount=mock.ANY, + kSecValueData=mock.ANY, + kSecAttrAccess=mock.ANY, + ), + ] + search = api.create_query.call_args_list[0].kwargs + attributes = api.create_query.call_args_list[1].kwargs + item = api.create_query.call_args_list[2].kwargs + assert search["kSecAttrService"].value == item["kSecAttrService"].value == 101 + assert search["kSecAttrAccount"].value == item["kSecAttrAccount"].value == 102 + assert attributes["kSecValueData"].value == item["kSecValueData"].value == 103 + assert api.create_query.call_args_list[2].kwargs["kSecAttrAccess"].value == 205 + + +def test_set_password_rejects_missing_access_list(): + api = keyring_api(update_status=-25300) + api._found.CFArrayCreate.return_value = None -@pytest.mark.parametrize("keychain_path", [None, "/tmp/test.keychain"]) -def test_set_password_scopes_legacy_operations_to_keychain(keychain_path): - security = mock.MagicMock() - core_foundation = mock.MagicMock() - - core_foundation.CFStringCreateWithCString.side_effect = [101, 102] - core_foundation.CFDataCreate.return_value = 103 - core_foundation.CFArrayCreate.side_effect = [104, 105] - core_foundation.CFDictionaryCreate.side_effect = [201, 202, 203] - security.SecAccessCreate.return_value = 0 - security.SecKeychainCopyDefault.return_value = 0 - security.SecKeychainOpen.return_value = 0 - security.SecItemUpdate.return_value = macos_keychain._ITEM_NOT_FOUND - security.SecItemAdd.return_value = 0 - - constants = (ctypes.c_void_p(value) for value in range(301, 312)) with ( - mock.patch("pgcli.macos_keychain._load_framework", side_effect=[security, core_foundation]), - mock.patch("pgcli.macos_keychain._constant", side_effect=constants), + mock.patch("pgcli.macos_keychain._get_api", return_value=api), + pytest.raises(RuntimeError, match="Unable to allocate Keychain access controls"), ): - macos_keychain.set_password("pgcli", "user@host@5432", "secret", keychain_path) - - if keychain_path is None: - security.SecKeychainCopyDefault.assert_called_once() - security.SecKeychainOpen.assert_not_called() - else: - security.SecKeychainOpen.assert_called_once() - security.SecKeychainCopyDefault.assert_not_called() - assert core_foundation.CFArrayCreate.call_args_list[0].args[2:] == ( - 1, - core_foundation.kCFTypeArrayCallBacks, - ) - assert core_foundation.CFArrayCreate.call_args_list[1] == mock.call(None, None, 0, None) + macos_keychain.set_password("pgcli", "user@host@5432", "secret") + + api._sec.SecAccessCreate.assert_not_called() + api.SecItemAdd.assert_not_called()