Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/jwcrypto/common.py: 43%

1# Copyright (C) 2015 JWCrypto Project Contributors - see LICENSE file 

2 

3import copy 

4import json 

5from base64 import urlsafe_b64decode, urlsafe_b64encode 

6from collections import namedtuple 

7from collections.abc import MutableMapping 

8 

9# Padding stripping versions as described in 

10# RFC 7515 Appendix C 

11 

12 

13def base64url_encode(payload): 

14 if not isinstance(payload, bytes): 

15 payload = payload.encode('utf-8') 

16 encode = urlsafe_b64encode(payload) 

17 return encode.decode('utf-8').rstrip('=') 

18 

19 

20def base64url_decode(payload): 

21 size = len(payload) % 4 

22 if size == 2: 

23 payload += '==' 

24 elif size == 3: 

25 payload += '=' 

26 elif size != 0: 

27 raise ValueError('Invalid base64 string') 

28 return urlsafe_b64decode(payload.encode('utf-8')) 

29 

30 

31# JSON encoding/decoding helpers with good defaults 

32 

33def json_encode(string): 

34 if isinstance(string, bytes): 

35 string = string.decode('utf-8') 

36 return json.dumps(string, separators=(',', ':'), sort_keys=True) 

37 

38 

39def json_decode(string): 

40 if isinstance(string, bytes): 

41 string = string.decode('utf-8') 

42 return json.loads(string) 

43 

44 

45class JWException(Exception): 

46 pass 

47 

48 

49class InvalidJWAAlgorithm(JWException): 

50 def __init__(self, message=None): 

51 msg = 'Invalid JWA Algorithm name' 

52 if message: 

53 msg += ' (%s)' % message 

54 super(InvalidJWAAlgorithm, self).__init__(msg) 

55 

56 

57class InvalidCEKeyLength(JWException): 

58 """Invalid CEK Key Length. 

59 

60 This exception is raised when a Content Encryption Key does not match 

61 the required length. 

62 """ 

63 

64 def __init__(self, expected, obtained): 

65 msg = 'Expected key of length %d bits, got %d' % (expected, obtained) 

66 super(InvalidCEKeyLength, self).__init__(msg) 

67 

68 

69class InvalidJWEData(JWException): 

70 """Invalid JWE Object. 

71 

72 This exception is raised when the JWE Object is invalid and/or 

73 improperly formatted. 

74 """ 

75 

76 def __init__(self, message=None, exception=None): 

77 msg = None 

78 if message: 

79 msg = message 

80 else: 

81 msg = 'Unknown Data Verification Failure' 

82 if exception: 

83 msg += ' {%s}' % str(exception) 

84 super(InvalidJWEData, self).__init__(msg) 

85 

86 

87class InvalidJWEOperation(JWException): 

88 """Invalid JWS Object. 

89 

90 This exception is raised when a requested operation cannot 

91 be execute due to unsatisfied conditions. 

92 """ 

93 

94 def __init__(self, message=None, exception=None): 

95 msg = None 

96 if message: 

97 msg = message 

98 else: 

99 msg = 'Unknown Operation Failure' 

100 if exception: 

101 msg += ' {%s}' % repr(exception) 

102 super(InvalidJWEOperation, self).__init__(msg) 

103 

104 

105class InvalidJWEKeyType(JWException): 

106 """Invalid JWE Key Type. 

107 

108 This exception is raised when the provided JWK Key does not match 

109 the type required by the specified algorithm. 

110 """ 

111 

112 def __init__(self, expected, obtained): 

113 msg = 'Expected key type %s, got %s' % (expected, obtained) 

114 super(InvalidJWEKeyType, self).__init__(msg) 

115 

116 

117class InvalidJWEKeyLength(JWException): 

118 """Invalid JWE Key Length. 

119 

120 This exception is raised when the provided JWK Key does not match 

121 the length required by the specified algorithm. 

122 """ 

123 

124 def __init__(self, expected, obtained): 

125 msg = 'Expected key of length %d, got %d' % (expected, obtained) 

126 super(InvalidJWEKeyLength, self).__init__(msg) 

127 

128 

129class InvalidJWSERegOperation(JWException): 

130 """Invalid JWSE Header Registry Operation. 

131 

132 This exception is raised when there is an error in trying to add a JW 

133 Signature or Encryption header to the Registry. 

134 """ 

135 

136 def __init__(self, message=None, exception=None): 

137 msg = None 

138 if message: 

139 msg = message 

140 else: 

141 msg = 'Unknown Operation Failure' 

142 if exception: 

143 msg += ' {%s}' % repr(exception) 

144 super(InvalidJWSERegOperation, self).__init__(msg) 

145 

146 

147class JWKeyNotFound(JWException): 

148 """The key needed to complete the operation was not found. 

149 

150 This exception is raised when a JWKSet is used to perform 

151 some operation and the key required to successfully complete 

152 the operation is not found. 

153 """ 

154 

155 def __init__(self, message=None): 

156 if message: 

157 msg = message 

158 else: 

159 msg = 'Key Not Found' 

160 super(JWKeyNotFound, self).__init__(msg) 

161 

162 

163# JWSE Header Registry definitions 

164 

165# RFC 7515 - 9.1: JSON Web Signature and Encryption Header Parameters Registry 

166# HeaderParameters are for both JWS and JWE 

167JWSEHeaderParameter = namedtuple('Parameter', 

168 'description mustprotect supported check_fn') 

169 

170 

171class JWSEHeaderRegistry(MutableMapping): 

172 def __init__(self, init_registry=None): 

173 if init_registry: 

174 if isinstance(init_registry, dict): 

175 self._registry = copy.deepcopy(init_registry) 

176 else: 

177 raise InvalidJWSERegOperation('Unknown input type') 

178 else: 

179 self._registry = {} 

180 

181 MutableMapping.__init__(self) 

182 

183 def check_header(self, h, value): 

184 if h not in self._registry: 

185 raise InvalidJWSERegOperation('No header "%s" found in registry' 

186 % h) 

187 

188 param = self._registry[h] 

189 if param.check_fn is None: 

190 return True 

191 else: 

192 return param.check_fn(value) 

193 

194 def __getitem__(self, key): 

195 return self._registry.__getitem__(key) 

196 

197 def __iter__(self): 

198 return self._registry.__iter__() 

199 

200 def __delitem__(self, key): 

201 if self._registry[key].mustprotect or \ 

202 self._registry[key].supported: 

203 raise InvalidJWSERegOperation('Unable to delete protected or ' 

204 'supported field') 

205 else: 

206 self._registry.__delitem__(key) 

207 

208 def __setitem__(self, h, jwse_header_param): 

209 # Check if a header is not supported 

210 if h in self._registry: 

211 p = self._registry[h] 

212 if p.supported: 

213 raise InvalidJWSERegOperation('Supported header already exists' 

214 ' in registry') 

215 elif p.mustprotect and not jwse_header_param.mustprotect: 

216 raise InvalidJWSERegOperation('Header specified should be' 

217 'a protected header') 

218 else: 

219 del self._registry[h] 

220 

221 self._registry[h] = jwse_header_param 

222 

223 def __len__(self): 

224 return self._registry.__len__()