From 5e35109441839f1c75d0f3cf0f24ede56ec7eecf Mon Sep 17 00:00:00 2001 From: uglyegg Date: Sun, 13 Sep 2026 09:06:54 -0500 Subject: [PATCH] fix: support desktop-selected Secret Service providers Treat org.freedesktop.secrets as the runtime contract instead of requiring GNOME Keyring. Avoid validation-time secret access, resolve a missing default alias conservatively, and reuse one serialized D-Bus connection so provider authorization can persist across operations. Keep GNOME Keyring as a packaging suggestion and cover validation, collection selection, and connection lifetime with focused tests. Development disclosure: I developed this patch with material assistance from OpenAI Codex under my direction, then reviewed, tested, and signed the final change. --- debian/control | 3 +- .../secretservice/secretservice_backend.py | 163 ++++++++++- rpmbuild/SPECS/package.spec | 3 +- tests/test_secretservice_backend.py | 255 ++++++++++++++++++ 4 files changed, 412 insertions(+), 12 deletions(-) create mode 100644 tests/test_secretservice_backend.py diff --git a/debian/control b/debian/control index c8652fb..68da75c 100644 --- a/debian/control +++ b/debian/control @@ -8,7 +8,8 @@ X-Python3-Version: >= 3.9 Package: python3-proton-keyring-linux Architecture: all -Depends: ${python3:Depends}, ${misc:Depends}, python3-proton-core, python3-keyring, python3-secretstorage, gnome-keyring +Depends: ${python3:Depends}, ${misc:Depends}, python3-proton-core, python3-keyring, python3-secretstorage +Suggests: gnome-keyring Replaces: python3-proton-keyring-linux-secretservice Breaks: python3-proton-keyring-linux-secretservice (<< 0.1.0) Description: Python3 Proton linux keyring base implementation diff --git a/proton/keyring_linux/secretservice/secretservice_backend.py b/proton/keyring_linux/secretservice/secretservice_backend.py index dc8c61b..aa25510 100644 --- a/proton/keyring_linux/secretservice/secretservice_backend.py +++ b/proton/keyring_linux/secretservice/secretservice_backend.py @@ -19,16 +19,161 @@ along with ProtonVPN. If not, see . """ import json +import logging import os +import threading +import weakref -import logging import keyring +from keyring.credentials import SimpleCredential from proton.keyring.exceptions import KeyringLocked, KeyringError from proton.keyring_linux.core import KeyringBackendLinux logger = logging.getLogger(__name__) +def _get_secret_service_backend(): + """Return Python Keyring's Secret Service backend with safer collection lookup.""" + # pylint: disable=import-outside-toplevel + import secretstorage + from keyring.backends import SecretService + + class ProviderAgnosticSecretServiceKeyring(SecretService.Keyring): + """Secret Service backend with provider-agnostic collection handling. + + A Secret Service client is identified by its unique D-Bus connection. + Keep one connection for the lifetime of this backend so providers can + remember authorization decisions across multiple keyring operations. + """ + + def __init__(self): + super().__init__() + self._connection = None + self._connection_finalizer = None + self._operation_lock = threading.RLock() + + def _get_connection(self): + if self._connection is None: + self._connection = secretstorage.dbus_init() + self._connection_finalizer = weakref.finalize( + self, self._connection.close + ) + return self._connection + + def close(self): + """Close the persistent D-Bus connection, if it was opened.""" + with self._operation_lock: + if ( + self._connection_finalizer is not None + and self._connection_finalizer.alive + ): + self._connection_finalizer() + self._connection = None + self._connection_finalizer = None + + def get_preferred_collection(self): + connection = self._get_connection() + try: + if hasattr(self, "preferred_collection"): + collection = secretstorage.Collection( + connection, self.preferred_collection + ) + else: + collection = self._get_default_or_only_collection( + secretstorage, connection + ) + except secretstorage.exceptions.SecretStorageException as excp: + raise keyring.errors.InitError( + f"Failed to get a Secret Service collection: {excp}." + ) from excp + + if collection.is_locked(): + collection.unlock() + if collection.is_locked(): + raise keyring.errors.KeyringLocked( + "Failed to unlock the collection!" + ) + return collection + + def get_password(self, service, username): + """Get a password without discarding the D-Bus client identity.""" + with self._operation_lock: + collection = self.get_preferred_collection() + items = collection.search_items(self._query(service, username)) + for item in items: + self.unlock(item) + return item.get_secret().decode("utf-8") + return None + + def set_password(self, service, username, password): + """Set a password using the persistent D-Bus connection.""" + with self._operation_lock: + collection = self.get_preferred_collection() + attributes = self._query( + service, username, application=self.appid + ) + label = f"Password for '{username}' on '{service}'" + collection.create_item( + label, attributes, password, replace=True + ) + + def delete_password(self, service, username): + """Delete a password using the persistent D-Bus connection.""" + with self._operation_lock: + collection = self.get_preferred_collection() + items = collection.search_items(self._query(service, username)) + for item in items: + return item.delete() + raise keyring.errors.PasswordDeleteError("No such password!") + + def get_credential(self, service, username): + """Get a credential using the persistent D-Bus connection.""" + with self._operation_lock: + scheme = self.schemes[self.scheme] + query = self._query(service, username) + collection = self.get_preferred_collection() + items = collection.search_items(query) + for item in items: + self.unlock(item) + item_username = item.get_attributes().get( + scheme["username"] + ) + return SimpleCredential( + item_username, item.get_secret().decode("utf-8") + ) + return None + + @staticmethod + def _get_default_or_only_collection(secretstorage_module, connection): + try: + return secretstorage_module.get_collection_by_alias( + connection, "default" + ) + except secretstorage_module.exceptions.ItemNotFoundException: + collections = list( + secretstorage_module.get_all_collections(connection) + ) + + if len(collections) == 1: + logger.warning( + "The default Secret Service collection alias is unavailable; " + "using the only advertised collection" + ) + return collections[0] + + if not collections: + return secretstorage_module.create_collection( + connection, "Default", "default" + ) + + raise keyring.errors.InitError( + "The default Secret Service collection alias is unavailable " + "and the provider advertises multiple collections" + ) + + return ProviderAgnosticSecretServiceKeyring() + + # pylint: disable=too-few-public-methods class KeyringBackendLinuxSecretService(KeyringBackendLinux): """Implements the Secret Service keyring backend.""" @@ -42,17 +187,17 @@ def _validate(cls): if os.environ.get("SNAP") is not None: return False try: - # pylint: disable=import-outside-toplevel - from keyring.backends import SecretService - return cls._is_backend_working(SecretService.Keyring()) - except ModuleNotFoundError: - logger.debug("Gnome-Keyring module not found") + # Python Keyring's priority check verifies that SecretStorage is + # installed and org.freedesktop.secrets is owned or activatable. + # Do not probe with get_password(): that requires a usable default + # collection and can create one as a side effect. + return _get_secret_service_backend().priority > 0 + except (ModuleNotFoundError, RuntimeError): + logger.debug("Secret Service backend not available", exc_info=True) return False def __init__(self): - # pylint: disable=import-outside-toplevel - from keyring.backends import SecretService - self._backend = SecretService.Keyring() + self._backend = _get_secret_service_backend() super().__init__(self._backend) def _get_item(self, key): diff --git a/rpmbuild/SPECS/package.spec b/rpmbuild/SPECS/package.spec index 3a73cdd..0f3d1e9 100644 --- a/rpmbuild/SPECS/package.spec +++ b/rpmbuild/SPECS/package.spec @@ -20,15 +20,14 @@ BuildRoot: %{_tmppath}/%{pep_625_name}-%{version}-%{release}-buildroot BuildRequires: python3-devel BuildRequires: python3-setuptools -BuildRequires: gnome-keyring BuildRequires: python3-keyring BuildRequires: python3-secretstorage BuildRequires: python3-proton-core -Requires: gnome-keyring Requires: python3-keyring Requires: python3-secretstorage Requires: python3-proton-core +Suggests: gnome-keyring Conflicts: python3-proton-keyring-linux-secretservice < 0.1.0 Obsoletes: python3-proton-keyring-linux-secretservice diff --git a/tests/test_secretservice_backend.py b/tests/test_secretservice_backend.py new file mode 100644 index 0000000..83912c0 --- /dev/null +++ b/tests/test_secretservice_backend.py @@ -0,0 +1,255 @@ +""" +Tests for the provider-agnostic Secret Service backend. + +Copyright (c) 2026 Proton AG + +This file is part of Proton VPN. + +Proton VPN is free software: you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +Proton VPN is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License +along with ProtonVPN. If not, see . +""" +from unittest import mock + +import keyring +import pytest +import secretstorage + +from proton.keyring_linux.secretservice import secretservice_backend +from proton.keyring_linux.secretservice.secretservice_backend import ( + KeyringBackendLinuxSecretService, +) + + +def test_validate_checks_availability_without_reading_a_secret(monkeypatch): + backend = mock.Mock(priority=5) + get_backend = mock.Mock(return_value=backend) + monkeypatch.delenv("SNAP", raising=False) + monkeypatch.setattr( + secretservice_backend, "_get_secret_service_backend", get_backend + ) + + assert KeyringBackendLinuxSecretService._validate() + get_backend.assert_called_once_with() + backend.get_password.assert_not_called() + + +def test_validate_rejects_unavailable_secret_service(monkeypatch): + monkeypatch.delenv("SNAP", raising=False) + monkeypatch.setattr( + secretservice_backend, + "_get_secret_service_backend", + mock.Mock(side_effect=RuntimeError("not owned or activatable")), + ) + + assert not KeyringBackendLinuxSecretService._validate() + + +def test_default_collection_is_preferred(): + backend = secretservice_backend._get_secret_service_backend() + secretstorage_module = mock.Mock() + connection = mock.sentinel.connection + default_collection = mock.sentinel.default_collection + secretstorage_module.get_collection_by_alias.return_value = default_collection + + result = backend._get_default_or_only_collection( + secretstorage_module, connection + ) + + assert result is default_collection + secretstorage_module.get_collection_by_alias.assert_called_once_with( + connection, "default" + ) + secretstorage_module.get_all_collections.assert_not_called() + + +def test_only_collection_is_used_when_default_alias_is_stale(): + backend = secretservice_backend._get_secret_service_backend() + secretstorage_module = mock.Mock() + secretstorage_module.exceptions.ItemNotFoundException = type( + "ItemNotFoundException", (Exception,), {} + ) + secretstorage_module.get_collection_by_alias.side_effect = ( + secretstorage_module.exceptions.ItemNotFoundException() + ) + collection = mock.sentinel.collection + secretstorage_module.get_all_collections.return_value = iter([collection]) + + result = backend._get_default_or_only_collection( + secretstorage_module, mock.sentinel.connection + ) + + assert result is collection + secretstorage_module.create_collection.assert_not_called() + + +def test_empty_service_may_create_a_default_collection(): + backend = secretservice_backend._get_secret_service_backend() + secretstorage_module = mock.Mock() + secretstorage_module.exceptions.ItemNotFoundException = type( + "ItemNotFoundException", (Exception,), {} + ) + secretstorage_module.get_collection_by_alias.side_effect = ( + secretstorage_module.exceptions.ItemNotFoundException() + ) + secretstorage_module.get_all_collections.return_value = iter([]) + created_collection = mock.sentinel.created_collection + secretstorage_module.create_collection.return_value = created_collection + connection = mock.sentinel.connection + + result = backend._get_default_or_only_collection( + secretstorage_module, connection + ) + + assert result is created_collection + secretstorage_module.create_collection.assert_called_once_with( + connection, "Default", "default" + ) + + +def test_multiple_collections_without_default_are_rejected(): + backend = secretservice_backend._get_secret_service_backend() + secretstorage_module = mock.Mock() + secretstorage_module.exceptions.ItemNotFoundException = type( + "ItemNotFoundException", (Exception,), {} + ) + secretstorage_module.get_collection_by_alias.side_effect = ( + secretstorage_module.exceptions.ItemNotFoundException() + ) + secretstorage_module.get_all_collections.return_value = iter( + [mock.sentinel.first, mock.sentinel.second] + ) + + with pytest.raises(keyring.errors.InitError, match="multiple collections"): + backend._get_default_or_only_collection( + secretstorage_module, mock.sentinel.connection + ) + + secretstorage_module.create_collection.assert_not_called() + + +def test_connection_is_reused_across_secret_reads(monkeypatch): + backend = secretservice_backend._get_secret_service_backend() + connection = mock.Mock() + monkeypatch.setattr( + secretstorage, "dbus_init", mock.Mock(return_value=connection) + ) + collection = mock.Mock() + collection.is_locked.return_value = False + first_item = mock.Mock() + first_item.is_locked.return_value = False + first_item.get_secret.return_value = b"first" + second_item = mock.Mock() + second_item.is_locked.return_value = False + second_item.get_secret.return_value = b"second" + collection.search_items.side_effect = [iter([first_item]), iter([second_item])] + monkeypatch.setattr( + backend, + "_get_default_or_only_collection", + mock.Mock(return_value=collection), + ) + + assert backend.get_password("Proton", "first") == "first" + assert backend.get_password("Proton", "second") == "second" + + secretstorage.dbus_init.assert_called_once_with() + assert backend._get_default_or_only_collection.call_count == 2 + assert all( + call.args == (secretstorage, connection) + for call in backend._get_default_or_only_collection.call_args_list + ) + connection.close.assert_not_called() + + backend.close() + connection.close.assert_called_once_with() + + +def test_closed_backend_opens_a_new_connection_on_next_operation(monkeypatch): + backend = secretservice_backend._get_secret_service_backend() + first_connection = mock.Mock() + second_connection = mock.Mock() + monkeypatch.setattr( + secretstorage, + "dbus_init", + mock.Mock(side_effect=[first_connection, second_connection]), + ) + collection = mock.Mock() + collection.is_locked.return_value = False + monkeypatch.setattr( + backend, + "_get_default_or_only_collection", + mock.Mock(return_value=collection), + ) + + backend.get_preferred_collection() + backend.close() + backend.get_preferred_collection() + backend.close() + + assert secretstorage.dbus_init.call_count == 2 + first_connection.close.assert_called_once_with() + second_connection.close.assert_called_once_with() + + +def test_get_credential_does_not_close_persistent_connection(monkeypatch): + backend = secretservice_backend._get_secret_service_backend() + connection = mock.Mock() + monkeypatch.setattr( + secretstorage, "dbus_init", mock.Mock(return_value=connection) + ) + collection = mock.Mock() + collection.is_locked.return_value = False + item = mock.Mock() + item.is_locked.return_value = False + item.get_attributes.return_value = {"username": "alice"} + item.get_secret.return_value = b"secret" + collection.search_items.return_value = iter([item]) + monkeypatch.setattr( + backend, + "_get_default_or_only_collection", + mock.Mock(return_value=collection), + ) + + credential = backend.get_credential("Proton", "alice") + + assert credential.username == "alice" + assert credential.password == "secret" + connection.close.assert_not_called() + backend.close() + connection.close.assert_called_once_with() + + +def test_write_and_delete_reuse_persistent_connection(monkeypatch): + backend = secretservice_backend._get_secret_service_backend() + connection = mock.Mock() + monkeypatch.setattr( + secretstorage, "dbus_init", mock.Mock(return_value=connection) + ) + collection = mock.Mock() + collection.is_locked.return_value = False + item = mock.Mock() + collection.search_items.return_value = iter([item]) + monkeypatch.setattr( + backend, + "_get_default_or_only_collection", + mock.Mock(return_value=collection), + ) + + backend.set_password("Proton", "alice", "secret") + backend.delete_password("Proton", "alice") + + secretstorage.dbus_init.assert_called_once_with() + collection.create_item.assert_called_once() + item.delete.assert_called_once_with() + connection.close.assert_not_called() + backend.close() + connection.close.assert_called_once_with()