diff --git a/AUTHORS b/AUTHORS index db49b9eb4..88fecdf64 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..78283e3ca 100644 --- a/changelog.rst +++ b/changelog.rst @@ -22,6 +22,9 @@ Features: Bug fixes: ---------- +* 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..8850b2831 100644 --- a/pgcli/auth.py +++ b/pgcli/auth.py @@ -1,4 +1,5 @@ import click +import platform from textwrap import dedent @@ -45,9 +46,33 @@ def keyring_get_password(key): return passwd +def _is_macos_keyring_backend(backend): + 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 + + 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..a72bd856c --- /dev/null +++ b/pgcli/macos_keychain.py @@ -0,0 +1,87 @@ +import ctypes + + +def _get_api(): + from keyring.backends.macOS import api + + return api + + +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 set_password(service, account, password): + """Store a password without pre-authorizing the creating executable.""" + 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_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 = [] + + def retain(value, error): + if not value: + raise RuntimeError(error) + retained.append(value) + return value + + try: + 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 == 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() + 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 = api.SecItemAdd(item, None) + api.Error.raise_for_status(status) + finally: + for value in reversed(retained): + cf_release(value) 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/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 13eed58db..010de0eef 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -25,12 +25,79 @@ 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.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.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") 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..1137d2db9 --- /dev/null +++ b/tests/test_macos_keychain.py @@ -0,0 +1,111 @@ +import ctypes +from types import SimpleNamespace +from unittest import mock + +import pytest + +from pgcli import macos_keychain + + +def keyring_api(update_status): + found = mock.MagicMock() + security = mock.MagicMock() + 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 + + 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_preserves_access_list_when_updating(): + api = keyring_api(update_status=0) + + with mock.patch("pgcli.macos_keychain._get_api", return_value=api): + macos_keychain.set_password("pgcli", "user@host@5432", "secret") + + 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 + + with ( + 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") + + api._sec.SecAccessCreate.assert_not_called() + api.SecItemAdd.assert_not_called() 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")))