Preserve macOS Keychain access choices

AUTHORS

@@ -154,6 +154,7 @@ Contributors:
     * Diego
     * Chris (ChrisJr404)
     * Pieter Ouwerkerk (pouwerkerk)
+    * Melvin Cerba (MelvinCERBA)
 
 Creator:
 --------

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``,

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

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)

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.

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"]
+testpaths = ["tests"]

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

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