derivepassphrase.git

1) # SPDX-FileCopyrightText: 2024 Marco Ricci <m@the13thletter.info>
2) #
3) # SPDX-License-Identifier: MIT
4) 
5) """Exporter for the vault native configuration format (v0.2 or v0.3)."""

6) 
7) from __future__ import annotations
8) 
9) import abc
10) import base64
11) import json
12) import logging
13) import warnings

14) from typing import TYPE_CHECKING
15) 
16) from derivepassphrase import exporter, vault
17) 
18) if TYPE_CHECKING:
19)     from typing import Any
20) 
21)     from typing_extensions import Buffer

22) 
23) if TYPE_CHECKING:
24)     from cryptography import exceptions as crypt_exceptions
25)     from cryptography import utils as crypt_utils
26)     from cryptography.hazmat.primitives import ciphers, hashes, hmac, padding
27)     from cryptography.hazmat.primitives.ciphers import algorithms, modes
28)     from cryptography.hazmat.primitives.kdf import pbkdf2
29) else:
30)     try:
31)         from cryptography import exceptions as crypt_exceptions
32)         from cryptography import utils as crypt_utils
33)         from cryptography.hazmat.primitives import (
34)             ciphers,
35)             hashes,
36)             hmac,
37)             padding,
38)         )
39)         from cryptography.hazmat.primitives.ciphers import algorithms, modes
40)         from cryptography.hazmat.primitives.kdf import pbkdf2
41)     except ModuleNotFoundError as exc:
42) 

43)         class _DummyModule:  # pragma: no cover

44)             def __init__(self, exc: type[Exception]) -> None:
45)                 self.exc = exc
46) 

47)             def __getattr__(self, name: str) -> Any:  # noqa: ANN401
48)                 def func(*args: Any, **kwargs: Any) -> Any:  # noqa: ANN401,ARG001

49)                     raise self.exc
50) 
51)                 return func
52) 

53)         crypt_exceptions = crypt_utils = _DummyModule(exc)
54)         ciphers = hashes = hmac = padding = _DummyModule(exc)
55)         algorithms = modes = pbkdf2 = _DummyModule(exc)

56)         STUBBED = True
57)     else:
58)         STUBBED = False

59) 
60) logger = logging.getLogger(__name__)
61) 
62) 
63) def _h(bs: bytes | bytearray) -> str:
64)     return 'bytes.fromhex({!r})'.format(bs.hex(' '))
65) 
66) 

67) class VaultNativeConfigParser(abc.ABC):

68)     """A base parser for vault's native configuration format.
69) 
70)     Certain details are specific to the respective vault versions, and
71)     are abstracted out.  This class by itself is not instantiable
72)     because of this.
73) 
74)     """
75) 

76)     def __init__(self, contents: Buffer, password: str | Buffer) -> None:

77)         """Initialize the parser.
78) 
79)         Args:
80)             contents:
81)                 The binary contents of the encrypted configuration file.
82) 
83)                 Note: On disk, these are usually stored in
84)                 base64-encoded form, not in the "raw" form as needed
85)                 here.
86) 
87)             password:
88)                 The vault master key/master passphrase the file is
89)                 encrypted with.  Must be non-empty.  See
90)                 [`derivepassphrase.exporter.get_vault_key`][] for
91)                 details.
92) 
93)                 If this is a text string, then the UTF-8 encoding of the
94)                 string is used as the binary password.
95) 
96)         """

97)         if not password:

98)             msg = 'Password must not be empty'

99)             raise ValueError(msg)  # noqa: DOC501

100)         self._contents = bytes(contents)

101)         self._iv_size = 0
102)         self._mac_size = 0
103)         self._encryption_key = b''
104)         self._encryption_key_size = 0
105)         self._signing_key = b''
106)         self._signing_key_size = 0
107)         self._message = b''
108)         self._message_tag = b''
109)         self._iv = b''
110)         self._payload = b''

111)         self._password = password
112)         self._sentinel: object = object()
113)         self._data: Any = self._sentinel
114) 

115)     def __call__(self) -> Any:  # noqa: ANN401
116)         """Return the decrypted and parsed vault configuration.
117) 
118)         Raises:
119)             cryptography.exceptions.InvalidSignature:
120)                 The encrypted configuration does not contain a valid
121)                 signature.
122)             ValueError:
123)                 The format is invalid, in a non-cryptographic way.  (For
124)                 example, it contains an unsupported version marker, or
125)                 unexpected extra contents, or invalid padding.)
126) 
127)         """

128)         if self._data is self._sentinel:
129)             self._parse_contents()
130)             self._derive_keys()
131)             self._check_signature()
132)             self._data = self._decrypt_payload()

133)         return self._data
134) 
135)     @staticmethod

136)     def _pbkdf2(

137)         password: str | Buffer, key_size: int, iterations: int

138)     ) -> bytes:
139)         if isinstance(password, str):
140)             password = password.encode('utf-8')
141)         raw_key = pbkdf2.PBKDF2HMAC(

142)             algorithm=hashes.SHA1(),

143)             length=key_size // 2,
144)             salt=vault.Vault._UUID,  # noqa: SLF001
145)             iterations=iterations,

146)         ).derive(bytes(password))

147)         logger.debug(
148)             'binary = pbkdf2(%s, %s, %s, %s, %s) = %s -> %s',
149)             repr(password),
150)             repr(vault.Vault._UUID),  # noqa: SLF001
151)             iterations,
152)             key_size // 2,
153)             repr('sha1'),
154)             _h(raw_key),
155)             _h(raw_key.hex().lower().encode('ASCII')),
156)         )
157)         return raw_key.hex().lower().encode('ASCII')
158) 
159)     def _parse_contents(self) -> None:
160)         logger.info('Parsing IV, payload and signature from the file contents')
161) 

162)         if len(self._contents) < self._iv_size + 16 + self._mac_size:

163)             msg = 'Invalid vault configuration file: file is truncated'

164)             raise ValueError(msg)
165) 

166)         def cut(buffer: bytes, cutpoint: int) -> tuple[bytes, bytes]:
167)             return buffer[:cutpoint], buffer[cutpoint:]
168) 

169)         cutpos1 = len(self._contents) - self._mac_size
170)         cutpos2 = self._iv_size

171) 

172)         self._message, self._message_tag = cut(self._contents, cutpos1)
173)         self._iv, self._payload = cut(self._message, cutpos2)

174) 
175)         logger.debug(
176)             'buffer %s = [[%s, %s], %s]',

177)             _h(self._contents),

178)             _h(self._iv),
179)             _h(self._payload),
180)             _h(self._message_tag),

181)         )
182) 
183)     def _derive_keys(self) -> None:
184)         logger.info('Deriving an encryption and signing key')
185)         self._generate_keys()
186)         assert (

187)             len(self._encryption_key) == self._encryption_key_size

188)         ), 'Derived encryption key is invalid'

189)         assert (

190)             len(self._signing_key) == self._signing_key_size

191)         ), 'Derived signing key is invalid'

192) 
193)     @abc.abstractmethod
194)     def _generate_keys(self) -> None:
195)         raise AssertionError
196) 
197)     def _check_signature(self) -> None:
198)         logger.info('Checking HMAC signature')

199)         mac = hmac.HMAC(self._signing_key, hashes.SHA256())

200)         mac_input = self._hmac_input()
201)         logger.debug(
202)             'mac_input = %s, expected_tag = %s',
203)             _h(mac_input),

204)             _h(self._message_tag),

205)         )
206)         mac.update(mac_input)
207)         try:

208)             mac.verify(self._message_tag)

209)         except crypt_exceptions.InvalidSignature:

210)             msg = 'File does not contain a valid signature'

211)             raise ValueError(msg) from None
212) 
213)     @abc.abstractmethod
214)     def _hmac_input(self) -> bytes:
215)         raise AssertionError
216) 

217)     def _decrypt_payload(self) -> Any:  # noqa: ANN401

218)         decryptor = self._make_decryptor()
219)         padded_plaintext = bytearray()

220)         padded_plaintext.extend(decryptor.update(self._payload))

221)         padded_plaintext.extend(decryptor.finalize())
222)         logger.debug('padded plaintext = %s', _h(padded_plaintext))

223)         unpadder = padding.PKCS7(self._iv_size * 8).unpadder()

224)         plaintext = bytearray()
225)         plaintext.extend(unpadder.update(padded_plaintext))
226)         plaintext.extend(unpadder.finalize())
227)         logger.debug('plaintext = %s', _h(plaintext))

228)         return json.loads(plaintext)

229) 
230)     @abc.abstractmethod
231)     def _make_decryptor(self) -> ciphers.CipherContext:
232)         raise AssertionError
233) 
234) 

235) class VaultNativeV03ConfigParser(VaultNativeConfigParser):

236)     """A parser for vault's native configuration format (v0.3).
237) 
238)     This is the modern, pre-storeroom configuration format.
239) 
240)     """
241) 

242)     KEY_SIZE = 32
243) 

244)     def __init__(self, *args: Any, **kwargs: Any) -> None:  # noqa: ANN401,D107

245)         super().__init__(*args, **kwargs)

246)         self._iv_size = 16
247)         self._mac_size = 32

248) 

249)     def __call__(self) -> Any:  # noqa: ANN401,D102

250)         if self._data is self._sentinel:
251)             logger.info('Attempting to parse as v0.3 configuration')
252)             return super().__call__()
253)         return self._data

254) 
255)     def _generate_keys(self) -> None:

256)         self._encryption_key = self._pbkdf2(self._password, self.KEY_SIZE, 100)
257)         self._signing_key = self._pbkdf2(self._password, self.KEY_SIZE, 200)
258)         self._encryption_key_size = self._signing_key_size = self.KEY_SIZE

259) 
260)     def _hmac_input(self) -> bytes:

261)         return self._message.hex().lower().encode('ASCII')

262) 
263)     def _make_decryptor(self) -> ciphers.CipherContext:
264)         return ciphers.Cipher(

265)             algorithms.AES256(self._encryption_key), modes.CBC(self._iv)

266)         ).decryptor()
267) 
268) 

269) class VaultNativeV02ConfigParser(VaultNativeConfigParser):

270)     """A parser for vault's native configuration format (v0.2).
271) 
272)     This is the classic configuration format.  Compared to v0.3, it
273)     contains an (accidental) API misuse for the generation of the master
274)     keys, a low-entropy method of generating initialization vectors for
275)     the AES-CBC encryption step, and extra layers of base64 encoding.
276)     Because of these significantly weakened confidentiality guarantees,
277)     v0.2 configurations should be upgraded to at least v0.3 as soon as
278)     possible.

279) 
280)     """
281) 
282)     def __init__(self, *args: Any, **kwargs: Any) -> None:  # noqa: ANN401,D107

283)         super().__init__(*args, **kwargs)

284)         self._iv_size = 16
285)         self._mac_size = 64

286) 

287)     def __call__(self) -> Any:  # noqa: ANN401,D102

288)         if self._data is self._sentinel:
289)             logger.info('Attempting to parse as v0.2 configuration')
290)             return super().__call__()
291)         return self._data

292) 
293)     def _parse_contents(self) -> None:
294)         super()._parse_contents()
295)         logger.debug('Decoding payload (base64) and message tag (hex)')

296)         self._payload = base64.standard_b64decode(self._payload)
297)         self._message_tag = bytes.fromhex(self._message_tag.decode('ASCII'))

298) 
299)     def _generate_keys(self) -> None:

300)         self._encryption_key = self._pbkdf2(self._password, 8, 16)
301)         self._signing_key = self._pbkdf2(self._password, 16, 16)
302)         self._encryption_key_size = 8
303)         self._signing_key_size = 16

304) 
305)     def _hmac_input(self) -> bytes:

306)         return base64.standard_b64encode(self._message)

307) 
308)     def _make_decryptor(self) -> ciphers.CipherContext:

309)         def evp_bytestokey_md5_one_iteration_no_salt(
310)             data: bytes, key_size: int, iv_size: int

311)         ) -> tuple[bytes, bytes]:
312)             total_size = key_size + iv_size
313)             buffer = bytearray()
314)             last_block = b''

315)             salt = b''

316)             logging.debug(
317)                 (
318)                     'data = %s, salt = %s, key_size = %s, iv_size = %s, '
319)                     'buffer length = %s, buffer = %s'
320)                 ),
321)                 _h(data),
322)                 _h(salt),
323)                 key_size,
324)                 iv_size,
325)                 len(buffer),
326)                 _h(buffer),
327)             )
328)             while len(buffer) < total_size:
329)                 with warnings.catch_warnings():
330)                     warnings.simplefilter(
331)                         'ignore', crypt_utils.CryptographyDeprecationWarning
332)                     )

333)                     block = hashes.Hash(hashes.MD5())

334)                 block.update(last_block)
335)                 block.update(data)
336)                 block.update(salt)
337)                 last_block = block.finalize()
338)                 buffer.extend(last_block)
339)                 logging.debug(
340)                     'buffer length = %s, buffer = %s', len(buffer), _h(buffer)
341)                 )
342)             logging.debug(
343)                 'encryption_key = %s, iv = %s',
344)                 _h(buffer[:key_size]),
345)                 _h(buffer[key_size:total_size]),
346)             )
347)             return bytes(buffer[:key_size]), bytes(buffer[key_size:total_size])
348) 

349)         data = base64.standard_b64encode(self._iv + self._encryption_key)

350)         encryption_key, iv = evp_bytestokey_md5_one_iteration_no_salt(
351)             data, key_size=32, iv_size=16

352)         )
353)         return ciphers.Cipher(
354)             algorithms.AES256(encryption_key), modes.CBC(iv)
355)         ).decryptor()
356) 
357) 
358) if __name__ == '__main__':
359)     import os
360) 
361)     logging.basicConfig(level=('DEBUG' if os.getenv('DEBUG') else 'WARNING'))

362)     with open(exporter.get_vault_path(), 'rb') as infile:

363)         contents = base64.standard_b64decode(infile.read())

364)     password = exporter.get_vault_key()

365)     try:

366)         config = VaultNativeV03ConfigParser(contents, password)()

367)     except ValueError:

368)         config = VaultNativeV02ConfigParser(contents, password)()