Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/cryptography/fernet.py: 55%

1# This file is dual licensed under the terms of the Apache License, Version 

2# 2.0, and the BSD License. See the LICENSE file in the root of this repository 

3# for complete details. 

4 

5from __future__ import annotations 

6 

7import base64 

8import binascii 

9import os 

10import time 

11import typing 

12from collections.abc import Iterable 

13 

14from cryptography import utils 

15from cryptography.exceptions import InvalidSignature 

16from cryptography.hazmat.primitives import hashes, padding 

17from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes 

18from cryptography.hazmat.primitives.hmac import HMAC 

19 

20 

21class InvalidToken(Exception): 

22 pass 

23 

24 

25_MAX_CLOCK_SKEW = 60 

26 

27# Hoisted to module level so each operation doesn't reconstruct them. 

28_PKCS7_128 = padding.PKCS7(128) 

29_SHA256 = hashes.SHA256() 

30 

31 

32class Fernet: 

33 def __init__( 

34 self, 

35 key: bytes | str, 

36 backend: typing.Any = None, 

37 ) -> None: 

38 try: 

39 key = base64.urlsafe_b64decode(key) 

40 except binascii.Error as exc: 

41 raise ValueError( 

42 "Fernet key must be 32 url-safe base64-encoded bytes." 

43 ) from exc 

44 if len(key) != 32: 

45 raise ValueError( 

46 "Fernet key must be 32 url-safe base64-encoded bytes." 

47 ) 

48 

49 self._signing_key = key[:16] 

50 self._encryption_key = key[16:] 

51 self._aes = algorithms.AES(self._encryption_key) 

52 

53 @classmethod 

54 def generate_key(cls) -> bytes: 

55 return base64.urlsafe_b64encode(os.urandom(32)) 

56 

57 def encrypt(self, data: bytes) -> bytes: 

58 return self.encrypt_at_time(data, int(time.time())) 

59 

60 def encrypt_at_time(self, data: bytes, current_time: int) -> bytes: 

61 iv = os.urandom(16) 

62 return self._encrypt_from_parts(data, current_time, iv) 

63 

64 def _encrypt_from_parts( 

65 self, data: bytes, current_time: int, iv: bytes 

66 ) -> bytes: 

67 utils._check_bytes("data", data) 

68 

69 padder = _PKCS7_128.padder() 

70 padded_data = padder.update(data) + padder.finalize() 

71 encryptor = Cipher(self._aes, modes.CBC(iv)).encryptor() 

72 ciphertext = encryptor.update(padded_data) + encryptor.finalize() 

73 

74 basic_parts = ( 

75 b"\x80" 

76 + current_time.to_bytes(length=8, byteorder="big") 

77 + iv 

78 + ciphertext 

79 ) 

80 

81 h = HMAC(self._signing_key, _SHA256) 

82 h.update(basic_parts) 

83 hmac = h.finalize() 

84 return base64.urlsafe_b64encode(basic_parts + hmac) 

85 

86 def decrypt(self, token: bytes | str, ttl: int | None = None) -> bytes: 

87 timestamp, data = Fernet._get_unverified_token_data(token) 

88 if ttl is None: 

89 time_info = None 

90 else: 

91 time_info = (ttl, int(time.time())) 

92 return self._decrypt_data(data, timestamp, time_info) 

93 

94 def decrypt_at_time( 

95 self, token: bytes | str, ttl: int, current_time: int 

96 ) -> bytes: 

97 if ttl is None: 

98 raise ValueError( 

99 "decrypt_at_time() can only be used with a non-None ttl" 

100 ) 

101 timestamp, data = Fernet._get_unverified_token_data(token) 

102 return self._decrypt_data(data, timestamp, (ttl, current_time)) 

103 

104 def extract_timestamp(self, token: bytes | str) -> int: 

105 timestamp, data = Fernet._get_unverified_token_data(token) 

106 # Verify the token was not tampered with. 

107 self._verify_signature(data) 

108 return timestamp 

109 

110 @staticmethod 

111 def _get_unverified_token_data(token: bytes | str) -> tuple[int, bytes]: 

112 if not isinstance(token, (str, bytes)): 

113 raise TypeError("token must be bytes or str") 

114 

115 try: 

116 data = base64.urlsafe_b64decode(token) 

117 except (TypeError, binascii.Error): 

118 raise InvalidToken 

119 

120 if not data or data[0] != 0x80: 

121 raise InvalidToken 

122 

123 if len(data) < 9: 

124 raise InvalidToken 

125 

126 timestamp = int.from_bytes(data[1:9], byteorder="big") 

127 return timestamp, data 

128 

129 def _verify_signature(self, data: bytes) -> None: 

130 h = HMAC(self._signing_key, _SHA256) 

131 h.update(memoryview(data)[:-32]) 

132 try: 

133 h.verify(data[-32:]) 

134 except InvalidSignature: 

135 raise InvalidToken 

136 

137 def _decrypt_data( 

138 self, 

139 data: bytes, 

140 timestamp: int, 

141 time_info: tuple[int, int] | None, 

142 ) -> bytes: 

143 if time_info is not None: 

144 ttl, current_time = time_info 

145 if timestamp + ttl < current_time: 

146 raise InvalidToken 

147 

148 if current_time + _MAX_CLOCK_SKEW < timestamp: 

149 raise InvalidToken 

150 

151 self._verify_signature(data) 

152 

153 mv = memoryview(data) 

154 iv = data[9:25] 

155 ciphertext = mv[25:-32] 

156 decryptor = Cipher(self._aes, modes.CBC(iv)).decryptor() 

157 plaintext_padded = decryptor.update(ciphertext) 

158 try: 

159 plaintext_padded += decryptor.finalize() 

160 except ValueError: 

161 raise InvalidToken 

162 unpadder = _PKCS7_128.unpadder() 

163 

164 unpadded = unpadder.update(plaintext_padded) 

165 try: 

166 unpadded += unpadder.finalize() 

167 except ValueError: 

168 raise InvalidToken 

169 return unpadded 

170 

171 

172class MultiFernet: 

173 def __init__(self, fernets: Iterable[Fernet]): 

174 fernets = list(fernets) 

175 if not fernets: 

176 raise ValueError( 

177 "MultiFernet requires at least one Fernet instance" 

178 ) 

179 self._fernets = fernets 

180 

181 def encrypt(self, msg: bytes) -> bytes: 

182 return self.encrypt_at_time(msg, int(time.time())) 

183 

184 def encrypt_at_time(self, msg: bytes, current_time: int) -> bytes: 

185 return self._fernets[0].encrypt_at_time(msg, current_time) 

186 

187 def rotate(self, msg: bytes | str) -> bytes: 

188 timestamp, data = Fernet._get_unverified_token_data(msg) 

189 for f in self._fernets: 

190 try: 

191 p = f._decrypt_data(data, timestamp, None) 

192 break 

193 except InvalidToken: 

194 pass 

195 else: 

196 raise InvalidToken 

197 

198 iv = os.urandom(16) 

199 return self._fernets[0]._encrypt_from_parts(p, timestamp, iv) 

200 

201 def decrypt(self, msg: bytes | str, ttl: int | None = None) -> bytes: 

202 if ttl is None: 

203 time_info = None 

204 else: 

205 time_info = (ttl, int(time.time())) 

206 # Parse the token once rather than once per key. 

207 timestamp, data = Fernet._get_unverified_token_data(msg) 

208 for f in self._fernets: 

209 try: 

210 return f._decrypt_data(data, timestamp, time_info) 

211 except InvalidToken: 

212 pass 

213 raise InvalidToken 

214 

215 def decrypt_at_time( 

216 self, msg: bytes | str, ttl: int, current_time: int 

217 ) -> bytes: 

218 if ttl is None: 

219 raise ValueError( 

220 "decrypt_at_time() can only be used with a non-None ttl" 

221 ) 

222 # Parse the token once rather than once per key. 

223 timestamp, data = Fernet._get_unverified_token_data(msg) 

224 for f in self._fernets: 

225 try: 

226 return f._decrypt_data(data, timestamp, (ttl, current_time)) 

227 except InvalidToken: 

228 pass 

229 raise InvalidToken 

230 

231 def extract_timestamp(self, msg: bytes | str) -> int: 

232 # Parse the token once rather than once per key. 

233 timestamp, data = Fernet._get_unverified_token_data(msg) 

234 for f in self._fernets: 

235 try: 

236 f._verify_signature(data) 

237 return timestamp 

238 except InvalidToken: 

239 pass 

240 raise InvalidToken