1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
|
# SecretStorage module for Python
# Access passwords using the SecretService DBus API
# Author: Dmitry Shachnev, 2013-2025
# License: 3-clause BSD, see LICENSE file
"""This module provides some utility functions, but these shouldn't
normally be used by external applications."""
import os
from typing import Any
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from jeepney import (
DBusAddress,
DBusErrorResponse,
MatchRule,
Message,
MessageType,
Properties,
new_method_call,
)
from jeepney.io.blocking import DBusConnection
from secretstorage.defines import (
ALGORITHM_DH,
ALGORITHM_PLAIN,
DBUS_EXEC_FAILED,
DBUS_NO_REPLY,
DBUS_NO_SUCH_OBJECT,
DBUS_NOT_SUPPORTED,
DBUS_SERVICE_UNKNOWN,
DBUS_UNKNOWN_METHOD,
DBUS_UNKNOWN_OBJECT,
SS_PATH,
SS_PREFIX,
)
from secretstorage.dhcrypto import Session
from secretstorage.exceptions import (
ItemNotFoundException,
SecretServiceNotAvailableException,
)
BUS_NAME = 'org.freedesktop.secrets'
SERVICE_IFACE = SS_PREFIX + 'Service'
PROMPT_IFACE = SS_PREFIX + 'Prompt'
class DBusAddressWrapper(DBusAddress): # type: ignore
"""A wrapper class around :class:`jeepney.wrappers.DBusAddress`
that adds some additional methods for calling and working with
properties, and converts error responses to SecretStorage
exceptions.
.. versionadded:: 3.0
"""
def __init__(self, path: str, interface: str,
connection: DBusConnection) -> None:
DBusAddress.__init__(self, path, BUS_NAME, interface)
self._connection = connection
def send_and_get_reply(self, msg: Message) -> Any:
try:
resp_msg: Message = self._connection.send_and_get_reply(msg)
if resp_msg.header.message_type == MessageType.error:
raise DBusErrorResponse(resp_msg)
return resp_msg.body
except DBusErrorResponse as resp:
if resp.name in (
DBUS_UNKNOWN_METHOD,
DBUS_NO_SUCH_OBJECT,
DBUS_UNKNOWN_OBJECT,
):
raise ItemNotFoundException('Item does not exist!') from resp
elif resp.name in (DBUS_SERVICE_UNKNOWN, DBUS_EXEC_FAILED,
DBUS_NO_REPLY):
data = resp.data
if isinstance(data, tuple):
data = data[0]
raise SecretServiceNotAvailableException(data) from resp
raise
def call(self, method: str, signature: str, *body: Any) -> Any:
msg = new_method_call(self, method, signature, body)
return self.send_and_get_reply(msg)
def get_property(self, name: str) -> Any:
msg = Properties(self).get(name)
(signature, value), = self.send_and_get_reply(msg)
return value
def set_property(self, name: str, signature: str, value: Any) -> None:
msg = Properties(self).set(name, signature, value)
self.send_and_get_reply(msg)
def open_session(connection: DBusConnection) -> Session:
"""Returns a new Secret Service session."""
service = DBusAddressWrapper(SS_PATH, SERVICE_IFACE, connection)
session = Session()
try:
output, result = service.call(
'OpenSession', 'sv',
ALGORITHM_DH,
('ay', session.my_public_key.to_bytes(128, 'big')))
except DBusErrorResponse as resp:
if resp.name != DBUS_NOT_SUPPORTED:
raise
output, result = service.call(
'OpenSession', 'sv',
ALGORITHM_PLAIN,
('s', ''))
session.encrypted = False
else:
signature, value = output
assert signature == 'ay'
key = int.from_bytes(value, 'big')
session.set_server_public_key(key)
session.object_path = result
return session
def format_secret(session: Session, secret: bytes,
content_type: str) -> tuple[str, bytes, bytes, str]:
"""Formats `secret` to make possible to pass it to the
Secret Service API."""
if isinstance(secret, str):
secret = secret.encode('utf-8')
elif not isinstance(secret, bytes):
raise TypeError('secret must be bytes')
assert session.object_path is not None
if not session.encrypted:
return (session.object_path, b'', secret, content_type)
assert session.aes_key is not None
# PKCS-7 style padding
padding = 0x10 - (len(secret) & 0xf)
secret += bytes((padding,) * padding)
aes_iv = os.urandom(0x10)
aes = algorithms.AES(session.aes_key)
encryptor = Cipher(aes, modes.CBC(aes_iv), default_backend()).encryptor()
encrypted_secret = encryptor.update(secret) + encryptor.finalize()
return (
session.object_path,
aes_iv,
encrypted_secret,
content_type
)
def exec_prompt(
connection: DBusConnection,
prompt_path: str,
*,
timeout: float | None = None,
) -> tuple[bool, tuple[str, Any]]:
"""Executes the prompt in a blocking mode.
:returns: a two-element tuple:
- The first element is a boolean value indicating whether the operation was
dismissed.
- The second element is a (signature, result) tuple. For creating items and
collections, ``signature`` is ``'o'`` and ``result`` is a single object
path. For unlocking, ``signature`` is ``'ao'`` and ``result`` is a list of
object paths.
.. versionchanged:: 3.5
Added ``timeout`` keyword argument.
"""
prompt = DBusAddressWrapper(prompt_path, PROMPT_IFACE, connection)
rule = MatchRule(
path=prompt_path,
interface=PROMPT_IFACE,
member='Completed',
type=MessageType.signal,
)
with connection.filter(rule) as signals:
prompt.call('Prompt', 's', '')
message = connection.recv_until_filtered(signals, timeout=timeout)
dismissed, result = message.body
assert dismissed is not None
assert result is not None
return dismissed, result
def unlock_objects(
connection: DBusConnection,
paths: list[str],
*,
timeout: float | None = None,
) -> bool:
"""Requests unlocking objects specified in `paths`.
Returns a boolean representing whether the operation was dismissed.
.. versionadded:: 2.1.2
.. versionchanged:: 3.5
Added ``timeout`` keyword argument.
"""
service = DBusAddressWrapper(SS_PATH, SERVICE_IFACE, connection)
unlocked_paths, prompt = service.call('Unlock', 'ao', paths)
if len(prompt) > 1:
dismissed, (signature, unlocked) = exec_prompt(
connection,
prompt,
timeout=timeout,
)
assert signature == 'ao'
return dismissed
return False
def add_match_rules(connection: DBusConnection) -> None:
"""Adds match rules for the given connection.
Currently it matches all messages from the Prompt interface, as the
mock service (unlike GNOME Keyring) does not specify the signal
destination.
.. versionadded:: 3.1
"""
rule = MatchRule(sender=BUS_NAME, interface=PROMPT_IFACE)
dbus = DBusAddressWrapper(path='/org/freedesktop/DBus',
interface='org.freedesktop.DBus',
connection=connection)
dbus.bus_name = 'org.freedesktop.DBus'
dbus.call('AddMatch', 's', rule.serialise())
|
