derivepassphrase.git

1) #!/usr/bin/python3
2) 
3) from __future__ import annotations
4) 
5) import abc
6) import base64
7) import json
8) import logging
9) import warnings

10) from typing import TYPE_CHECKING
11) 
12) from derivepassphrase import exporter, vault
13) 
14) if TYPE_CHECKING:
15)     from typing import Any
16) 
17)     from typing_extensions import Buffer

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

39)         class DummyModule:  # pragma: no cover

40)             def __init__(self, exc: type[Exception]) -> None:
41)                 self.exc = exc
42) 
43)             def __getattr__(self, name: str) -> Any:
44)                 def func(*args: Any, **kwargs: Any) -> Any:  # noqa: ARG001
45)                     raise self.exc
46) 
47)                 return func
48) 
49)         crypt_exceptions = crypt_utils = DummyModule(exc)
50)         ciphers = hashes = hmac = padding = DummyModule(exc)
51)         algorithms = modes = pbkdf2 = DummyModule(exc)
52)         STUBBED = True
53)     else:
54)         STUBBED = False

55) 
56) logger = logging.getLogger(__name__)
57) 
58) 
59) def _h(bs: bytes | bytearray) -> str:
60)     return 'bytes.fromhex({!r})'.format(bs.hex(' '))
61) 
62) 

63) class VaultNativeConfigParser(abc.ABC):
64)     def __init__(self, contents: Buffer, password: str | Buffer) -> None:

65)         if not password:

66)             msg = 'Password must not be empty'

67)             raise ValueError(msg)

68)         self._contents = bytes(contents)

69)         self.iv_size = 0
70)         self.mac_size = 0
71)         self.encryption_key = b''
72)         self.encryption_key_size = 0
73)         self.signing_key = b''
74)         self.signing_key_size = 0

75)         self.message = b''
76)         self.message_tag = b''
77)         self.iv = b''
78)         self.payload = b''
79)         self._password = password
80)         self._sentinel: object = object()
81)         self._data: Any = self._sentinel
82) 
83)     def __call__(self) -> Any:
84)         if self._data is self._sentinel:
85)             self._parse_contents()
86)             self._derive_keys()
87)             self._check_signature()
88)             self._data = self._decrypt_payload()

89)         return self._data
90) 
91)     @staticmethod
92)     def pbkdf2(

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

94)     ) -> bytes:
95)         if isinstance(password, str):
96)             password = password.encode('utf-8')
97)         raw_key = pbkdf2.PBKDF2HMAC(
98)             algorithm=hashes.SHA1(),  # noqa: S303
99)             length=key_size // 2,
100)             salt=vault.Vault._UUID,  # noqa: SLF001
101)             iterations=iterations,

102)         ).derive(bytes(password))

103)         logger.debug(
104)             'binary = pbkdf2(%s, %s, %s, %s, %s) = %s -> %s',
105)             repr(password),
106)             repr(vault.Vault._UUID),  # noqa: SLF001
107)             iterations,
108)             key_size // 2,
109)             repr('sha1'),
110)             _h(raw_key),
111)             _h(raw_key.hex().lower().encode('ASCII')),
112)         )
113)         return raw_key.hex().lower().encode('ASCII')
114) 
115)     def _parse_contents(self) -> None:
116)         logger.info('Parsing IV, payload and signature from the file contents')
117) 

118)         if len(self._contents) < self.iv_size + 16 + self.mac_size:
119)             msg = 'Invalid vault configuration file: file is truncated'

120)             raise ValueError(msg)
121) 

122)         def cut(buffer: bytes, cutpoint: int) -> tuple[bytes, bytes]:
123)             return buffer[:cutpoint], buffer[cutpoint:]
124) 
125)         cutpos1 = len(self._contents) - self.mac_size
126)         cutpos2 = self.iv_size

127) 

128)         self.message, self.message_tag = cut(self._contents, cutpos1)
129)         self.iv, self.payload = cut(self.message, cutpos2)

130) 
131)         logger.debug(
132)             'buffer %s = [[%s, %s], %s]',

133)             _h(self._contents),

134)             _h(self.iv),
135)             _h(self.payload),
136)             _h(self.message_tag),
137)         )
138) 
139)     def _derive_keys(self) -> None:
140)         logger.info('Deriving an encryption and signing key')
141)         self._generate_keys()
142)         assert (
143)             len(self.encryption_key) == self.encryption_key_size

144)         ), 'Derived encryption key is invalid'

145)         assert (
146)             len(self.signing_key) == self.signing_key_size

147)         ), 'Derived signing key is invalid'

148) 
149)     @abc.abstractmethod
150)     def _generate_keys(self) -> None:
151)         raise AssertionError
152) 
153)     def _check_signature(self) -> None:
154)         logger.info('Checking HMAC signature')
155)         mac = hmac.HMAC(self.signing_key, hashes.SHA256())
156)         mac_input = self._hmac_input()
157)         logger.debug(
158)             'mac_input = %s, expected_tag = %s',
159)             _h(mac_input),
160)             _h(self.message_tag),
161)         )
162)         mac.update(mac_input)
163)         try:
164)             mac.verify(self.message_tag)
165)         except crypt_exceptions.InvalidSignature:

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

167)             raise ValueError(msg) from None
168) 
169)     @abc.abstractmethod
170)     def _hmac_input(self) -> bytes:
171)         raise AssertionError
172) 

173)     def _decrypt_payload(self) -> Any:

174)         decryptor = self._make_decryptor()
175)         padded_plaintext = bytearray()
176)         padded_plaintext.extend(decryptor.update(self.payload))
177)         padded_plaintext.extend(decryptor.finalize())
178)         logger.debug('padded plaintext = %s', _h(padded_plaintext))
179)         unpadder = padding.PKCS7(self.iv_size * 8).unpadder()
180)         plaintext = bytearray()
181)         plaintext.extend(unpadder.update(padded_plaintext))
182)         plaintext.extend(unpadder.finalize())
183)         logger.debug('plaintext = %s', _h(plaintext))

184)         return json.loads(plaintext)

185) 
186)     @abc.abstractmethod
187)     def _make_decryptor(self) -> ciphers.CipherContext:
188)         raise AssertionError
189) 
190) 

191) class VaultNativeV03ConfigParser(VaultNativeConfigParser):

192)     KEY_SIZE = 32
193) 
194)     def __init__(self, *args: Any, **kwargs: Any) -> None:
195)         super().__init__(*args, **kwargs)
196)         self.iv_size = 16
197)         self.mac_size = 32
198) 

199)     def __call__(self) -> Any:
200)         if self._data is self._sentinel:
201)             logger.info('Attempting to parse as v0.3 configuration')
202)             return super().__call__()
203)         return self._data

204) 
205)     def _generate_keys(self) -> None:

206)         self.encryption_key = self.pbkdf2(self._password, self.KEY_SIZE, 100)
207)         self.signing_key = self.pbkdf2(self._password, self.KEY_SIZE, 200)

208)         self.encryption_key_size = self.signing_key_size = self.KEY_SIZE
209) 
210)     def _hmac_input(self) -> bytes:
211)         return self.message.hex().lower().encode('ASCII')
212) 
213)     def _make_decryptor(self) -> ciphers.CipherContext:
214)         return ciphers.Cipher(
215)             algorithms.AES256(self.encryption_key), modes.CBC(self.iv)
216)         ).decryptor()
217) 
218) 

219) class VaultNativeV02ConfigParser(VaultNativeConfigParser):

220)     def __init__(self, *args: Any, **kwargs: Any) -> None:
221)         super().__init__(*args, **kwargs)
222)         self.iv_size = 16
223)         self.mac_size = 64
224) 

225)     def __call__(self) -> Any:
226)         if self._data is self._sentinel:
227)             logger.info('Attempting to parse as v0.2 configuration')
228)             return super().__call__()
229)         return self._data

230) 
231)     def _parse_contents(self) -> None:
232)         super()._parse_contents()
233)         logger.debug('Decoding payload (base64) and message tag (hex)')
234)         self.payload = base64.standard_b64decode(self.payload)
235)         self.message_tag = bytes.fromhex(self.message_tag.decode('ASCII'))
236) 
237)     def _generate_keys(self) -> None:

238)         self.encryption_key = self.pbkdf2(self._password, 8, 16)
239)         self.signing_key = self.pbkdf2(self._password, 16, 16)

240)         self.encryption_key_size = 8
241)         self.signing_key_size = 16
242) 
243)     def _hmac_input(self) -> bytes:
244)         return base64.standard_b64encode(self.message)
245) 
246)     def _make_decryptor(self) -> ciphers.CipherContext:

247)         def evp_bytestokey_md5_one_iteration_no_salt(
248)             data: bytes, key_size: int, iv_size: int

249)         ) -> tuple[bytes, bytes]:
250)             total_size = key_size + iv_size
251)             buffer = bytearray()
252)             last_block = b''

253)             salt = b''

254)             logging.debug(
255)                 (
256)                     'data = %s, salt = %s, key_size = %s, iv_size = %s, '
257)                     'buffer length = %s, buffer = %s'
258)                 ),
259)                 _h(data),
260)                 _h(salt),
261)                 key_size,
262)                 iv_size,
263)                 len(buffer),
264)                 _h(buffer),
265)             )
266)             while len(buffer) < total_size:
267)                 with warnings.catch_warnings():
268)                     warnings.simplefilter(
269)                         'ignore', crypt_utils.CryptographyDeprecationWarning
270)                     )
271)                     block = hashes.Hash(hashes.MD5())  # noqa: S303
272)                 block.update(last_block)
273)                 block.update(data)
274)                 block.update(salt)
275)                 last_block = block.finalize()
276)                 buffer.extend(last_block)
277)                 logging.debug(
278)                     'buffer length = %s, buffer = %s', len(buffer), _h(buffer)
279)                 )
280)             logging.debug(
281)                 'encryption_key = %s, iv = %s',
282)                 _h(buffer[:key_size]),
283)                 _h(buffer[key_size:total_size]),
284)             )
285)             return bytes(buffer[:key_size]), bytes(buffer[key_size:total_size])
286) 
287)         data = base64.standard_b64encode(self.iv + self.encryption_key)

288)         encryption_key, iv = evp_bytestokey_md5_one_iteration_no_salt(
289)             data, key_size=32, iv_size=16

290)         )
291)         return ciphers.Cipher(
292)             algorithms.AES256(encryption_key), modes.CBC(iv)
293)         ).decryptor()
294) 
295) 
296) if __name__ == '__main__':
297)     import os
298) 
299)     logging.basicConfig(level=('DEBUG' if os.getenv('DEBUG') else 'WARNING'))

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

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

302)     password = exporter.get_vault_key()

303)     try:

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

305)     except ValueError:

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