Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/jose/jws.py: 38%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

124 statements  

1import binascii 

2import json 

3 

4try: 

5 from collections.abc import Iterable, Mapping 

6except ImportError: 

7 from collections import Mapping, Iterable 

8 

9from jose import jwk 

10from jose.backends.base import Key 

11from jose.constants import ALGORITHMS 

12from jose.exceptions import JWSError, JWSSignatureError 

13from jose.utils import base64url_decode, base64url_encode 

14 

15 

16def sign(payload, key, headers=None, algorithm=ALGORITHMS.HS256): 

17 """Signs a claims set and returns a JWS string. 

18 

19 Args: 

20 payload (str or dict): A string to sign 

21 key (str or dict): The key to use for signing the claim set. Can be 

22 individual JWK or JWK set. 

23 headers (dict, optional): A set of headers that will be added to 

24 the default headers. Any headers that are added as additional 

25 headers will override the default headers. 

26 algorithm (str, optional): The algorithm to use for signing the 

27 the claims. Defaults to HS256. 

28 

29 Returns: 

30 str: The string representation of the header, claims, and signature. 

31 

32 Raises: 

33 JWSError: If there is an error signing the token. 

34 

35 Examples: 

36 

37 >>> jws.sign({'a': 'b'}, 'secret', algorithm='HS256') 

38 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhIjoiYiJ9.jiMyrsmD8AoHWeQgmxZ5yq8z0lXS67_QGs52AzC8Ru8' 

39 

40 """ 

41 

42 if algorithm not in ALGORITHMS.SUPPORTED: 

43 raise JWSError("Algorithm %s not supported." % algorithm) 

44 

45 encoded_header = _encode_header(algorithm, additional_headers=headers) 

46 encoded_payload = _encode_payload(payload) 

47 signed_output = _sign_header_and_claims(encoded_header, encoded_payload, algorithm, key) 

48 

49 return signed_output 

50 

51 

52def verify(token, key, algorithms, verify=True): 

53 """Verifies a JWS string's signature. 

54 

55 Args: 

56 token (str): A signed JWS to be verified. 

57 key (str or dict): A key to attempt to verify the payload with. Can be 

58 individual JWK or JWK set. 

59 algorithms (str or list): Valid algorithms that should be used to verify the JWS. 

60 

61 Returns: 

62 str: The str representation of the payload, assuming the signature is valid. 

63 

64 Raises: 

65 JWSError: If there is an exception verifying a token. 

66 

67 Examples: 

68 

69 >>> token = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhIjoiYiJ9.jiMyrsmD8AoHWeQgmxZ5yq8z0lXS67_QGs52AzC8Ru8' 

70 >>> jws.verify(token, 'secret', algorithms='HS256') 

71 

72 """ 

73 

74 header, payload, signing_input, signature = _load(token) 

75 

76 if verify: 

77 _verify_signature(signing_input, header, signature, key, algorithms) 

78 

79 return payload 

80 

81 

82def get_unverified_header(token): 

83 """Returns the decoded headers without verification of any kind. 

84 

85 Args: 

86 token (str): A signed JWS to decode the headers from. 

87 

88 Returns: 

89 dict: The dict representation of the token headers. 

90 

91 Raises: 

92 JWSError: If there is an exception decoding the token. 

93 """ 

94 header, claims, signing_input, signature = _load(token) 

95 return header 

96 

97 

98def get_unverified_headers(token): 

99 """Returns the decoded headers without verification of any kind. 

100 

101 This is simply a wrapper of get_unverified_header() for backwards 

102 compatibility. 

103 

104 Args: 

105 token (str): A signed JWS to decode the headers from. 

106 

107 Returns: 

108 dict: The dict representation of the token headers. 

109 

110 Raises: 

111 JWSError: If there is an exception decoding the token. 

112 """ 

113 return get_unverified_header(token) 

114 

115 

116def get_unverified_claims(token): 

117 """Returns the decoded claims without verification of any kind. 

118 

119 Args: 

120 token (str): A signed JWS to decode the headers from. 

121 

122 Returns: 

123 str: The str representation of the token claims. 

124 

125 Raises: 

126 JWSError: If there is an exception decoding the token. 

127 """ 

128 header, claims, signing_input, signature = _load(token) 

129 return claims 

130 

131 

132def _encode_header(algorithm, additional_headers=None): 

133 header = {"typ": "JWT", "alg": algorithm} 

134 

135 if additional_headers: 

136 header.update(additional_headers) 

137 

138 json_header = json.dumps( 

139 header, 

140 separators=(",", ":"), 

141 sort_keys=True, 

142 ).encode("utf-8") 

143 

144 return base64url_encode(json_header) 

145 

146 

147def _encode_payload(payload): 

148 if isinstance(payload, Mapping): 

149 try: 

150 payload = json.dumps( 

151 payload, 

152 separators=(",", ":"), 

153 ).encode("utf-8") 

154 except ValueError: 

155 pass 

156 

157 return base64url_encode(payload) 

158 

159 

160def _sign_header_and_claims(encoded_header, encoded_claims, algorithm, key): 

161 signing_input = b".".join([encoded_header, encoded_claims]) 

162 try: 

163 if not isinstance(key, Key): 

164 key = jwk.construct(key, algorithm) 

165 signature = key.sign(signing_input) 

166 except Exception as e: 

167 raise JWSError(e) 

168 

169 encoded_signature = base64url_encode(signature) 

170 

171 encoded_string = b".".join([encoded_header, encoded_claims, encoded_signature]) 

172 

173 return encoded_string.decode("utf-8") 

174 

175 

176def _load(jwt): 

177 if isinstance(jwt, str): 

178 jwt = jwt.encode("utf-8") 

179 try: 

180 signing_input, crypto_segment = jwt.rsplit(b".", 1) 

181 header_segment, claims_segment = signing_input.split(b".", 1) 

182 header_data = base64url_decode(header_segment) 

183 except ValueError: 

184 raise JWSError("Not enough segments") 

185 except (TypeError, binascii.Error): 

186 raise JWSError("Invalid header padding") 

187 

188 try: 

189 header = json.loads(header_data.decode("utf-8")) 

190 except ValueError as e: 

191 raise JWSError("Invalid header string: %s" % e) 

192 

193 if not isinstance(header, Mapping): 

194 raise JWSError("Invalid header string: must be a json object") 

195 

196 try: 

197 payload = base64url_decode(claims_segment) 

198 except (TypeError, binascii.Error): 

199 raise JWSError("Invalid payload padding") 

200 

201 try: 

202 signature = base64url_decode(crypto_segment) 

203 except (TypeError, binascii.Error): 

204 raise JWSError("Invalid crypto padding") 

205 

206 return (header, payload, signing_input, signature) 

207 

208 

209def _sig_matches_keys(keys, signing_input, signature, alg): 

210 for key in keys: 

211 if not isinstance(key, Key): 

212 key = jwk.construct(key, alg) 

213 try: 

214 if key.verify(signing_input, signature): 

215 return True 

216 except Exception: 

217 pass 

218 return False 

219 

220 

221def _get_keys(key): 

222 if isinstance(key, Key): 

223 return (key,) 

224 

225 try: 

226 key = json.loads(key, parse_int=str, parse_float=str) 

227 except Exception: 

228 pass 

229 

230 if isinstance(key, Mapping): 

231 if "keys" in key: 

232 # JWK Set per RFC 7517 

233 return key["keys"] 

234 elif "kty" in key: 

235 # Individual JWK per RFC 7517 

236 return (key,) 

237 else: 

238 # Some other mapping. Firebase uses just dict of kid, cert pairs 

239 values = key.values() 

240 if values: 

241 return values 

242 return (key,) 

243 

244 # Iterable but not text or mapping => list- or tuple-like 

245 elif isinstance(key, Iterable) and not (isinstance(key, str) or isinstance(key, bytes)): 

246 return key 

247 

248 # Scalar value, wrap in tuple. 

249 else: 

250 return (key,) 

251 

252 

253def _verify_signature(signing_input, header, signature, key="", algorithms=None): 

254 alg = header.get("alg") 

255 if not alg: 

256 raise JWSError("No algorithm was specified in the JWS header.") 

257 

258 if algorithms is not None and alg not in algorithms: 

259 raise JWSError("The specified alg value is not allowed") 

260 

261 keys = _get_keys(key) 

262 try: 

263 if not _sig_matches_keys(keys, signing_input, signature, alg): 

264 raise JWSSignatureError() 

265 except JWSSignatureError: 

266 raise JWSError("Signature verification failed.") 

267 except JWSError: 

268 raise JWSError("Invalid or unsupported algorithm: %s" % alg)