Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/flask_jwt_extended/jwt_manager.py: 77%

1import datetime 

2from typing import Any 

3from typing import Callable 

4from typing import Optional 

5from typing import TypeVar 

6 

7import jwt 

8from flask import Flask 

9from jwt import DecodeError 

10from jwt import ExpiredSignatureError 

11from jwt import InvalidAudienceError 

12from jwt import InvalidIssuerError 

13from jwt import InvalidTokenError 

14from jwt import MissingRequiredClaimError 

15 

16from flask_jwt_extended.config import config 

17from flask_jwt_extended.default_callbacks import default_additional_claims_callback 

18from flask_jwt_extended.default_callbacks import default_blocklist_callback 

19from flask_jwt_extended.default_callbacks import default_decode_key_callback 

20from flask_jwt_extended.default_callbacks import default_encode_key_callback 

21from flask_jwt_extended.default_callbacks import default_expired_token_callback 

22from flask_jwt_extended.default_callbacks import default_invalid_token_callback 

23from flask_jwt_extended.default_callbacks import default_jwt_headers_callback 

24from flask_jwt_extended.default_callbacks import default_needs_fresh_token_callback 

25from flask_jwt_extended.default_callbacks import default_revoked_token_callback 

26from flask_jwt_extended.default_callbacks import default_token_verification_callback 

27from flask_jwt_extended.default_callbacks import ( 

28 default_token_verification_failed_callback, 

29) 

30from flask_jwt_extended.default_callbacks import default_unauthorized_callback 

31from flask_jwt_extended.default_callbacks import default_user_identity_callback 

32from flask_jwt_extended.default_callbacks import default_user_lookup_error_callback 

33from flask_jwt_extended.exceptions import CSRFError 

34from flask_jwt_extended.exceptions import FreshTokenRequired 

35from flask_jwt_extended.exceptions import InvalidHeaderError 

36from flask_jwt_extended.exceptions import InvalidQueryParamError 

37from flask_jwt_extended.exceptions import JWTDecodeError 

38from flask_jwt_extended.exceptions import NoAuthorizationError 

39from flask_jwt_extended.exceptions import RevokedTokenError 

40from flask_jwt_extended.exceptions import UserClaimsVerificationError 

41from flask_jwt_extended.exceptions import UserLookupError 

42from flask_jwt_extended.exceptions import WrongTokenError 

43from flask_jwt_extended.tokens import _decode_jwt 

44from flask_jwt_extended.tokens import _encode_jwt 

45from flask_jwt_extended.typing import ExpiresDelta 

46from flask_jwt_extended.typing import Fresh 

47from flask_jwt_extended.utils import current_user_context_processor 

48 

49_F = TypeVar("_F", bound=Callable[..., Any]) 

50 

51 

52class JWTManager(object): 

53 """ 

54 An object used to hold JWT settings and callback functions for the 

55 Flask-JWT-Extended extension. 

56 

57 Instances of :class:`JWTManager` are *not* bound to specific apps, so 

58 you can create one in the main body of your code and then bind it 

59 to your app in a factory function. 

60 """ 

61 

62 def __init__( 

63 self, app: Optional[Flask] = None, add_context_processor: bool = False 

64 ) -> None: 

65 """ 

66 Create the JWTManager instance. You can either pass a flask application 

67 in directly here to register this extension with the flask app, or 

68 call init_app after creating this object (in a factory pattern). 

69 

70 :param app: 

71 The Flask Application object 

72 :param add_context_processor: 

73 Controls if `current_user` is should be added to flasks template 

74 context (and thus be available for use in Jinja templates). Defaults 

75 to ``False``. 

76 """ 

77 # Register the default error handler callback methods. These can be 

78 # overridden with the appropriate loader decorators 

79 self._decode_key_callback = default_decode_key_callback 

80 self._encode_key_callback = default_encode_key_callback 

81 self._expired_token_callback = default_expired_token_callback 

82 self._invalid_token_callback = default_invalid_token_callback 

83 self._jwt_additional_header_callback = default_jwt_headers_callback 

84 self._needs_fresh_token_callback = default_needs_fresh_token_callback 

85 self._revoked_token_callback = default_revoked_token_callback 

86 self._token_in_blocklist_callback = default_blocklist_callback 

87 self._token_verification_callback = default_token_verification_callback 

88 self._unauthorized_callback = default_unauthorized_callback 

89 self._user_claims_callback = default_additional_claims_callback 

90 self._user_identity_callback = default_user_identity_callback 

91 self._user_lookup_callback: Optional[Callable] = None 

92 self._user_lookup_error_callback = default_user_lookup_error_callback 

93 self._token_verification_failed_callback = ( 

94 default_token_verification_failed_callback 

95 ) 

96 

97 # Register this extension with the flask app now (if it is provided) 

98 if app is not None: 

99 self.init_app(app, add_context_processor) 

100 

101 def init_app(self, app: Flask, add_context_processor: bool = False) -> None: 

102 """ 

103 Register this extension with the flask app. 

104 

105 :param app: 

106 The Flask Application object 

107 :param add_context_processor: 

108 Controls if `current_user` is should be added to flasks template 

109 context (and thus be available for use in Jinja templates). Defaults 

110 to ``False``. 

111 """ 

112 # Save this so we can use it later in the extension 

113 if not hasattr(app, "extensions"): # pragma: no cover 

114 app.extensions = {} 

115 app.extensions["flask-jwt-extended"] = self 

116 

117 if add_context_processor: 

118 app.context_processor(current_user_context_processor) 

119 

120 # Set all the default configurations for this extension 

121 self._set_default_configuration_options(app) 

122 self._set_error_handler_callbacks(app) 

123 

124 def _set_error_handler_callbacks(self, app: Flask) -> None: 

125 @app.errorhandler(CSRFError) 

126 def handle_csrf_error(e): 

127 return self._unauthorized_callback(str(e)) 

128 

129 @app.errorhandler(DecodeError) 

130 def handle_decode_error(e): 

131 return self._invalid_token_callback(str(e)) 

132 

133 @app.errorhandler(ExpiredSignatureError) 

134 def handle_expired_error(e): 

135 return self._expired_token_callback(e.jwt_header, e.jwt_data) 

136 

137 @app.errorhandler(FreshTokenRequired) 

138 def handle_fresh_token_required(e): 

139 return self._needs_fresh_token_callback(e.jwt_header, e.jwt_data) 

140 

141 @app.errorhandler(MissingRequiredClaimError) 

142 def handle_missing_required_claim_error(e): 

143 return self._invalid_token_callback(str(e)) 

144 

145 @app.errorhandler(InvalidAudienceError) 

146 def handle_invalid_audience_error(e): 

147 return self._invalid_token_callback(str(e)) 

148 

149 @app.errorhandler(InvalidIssuerError) 

150 def handle_invalid_issuer_error(e): 

151 return self._invalid_token_callback(str(e)) 

152 

153 @app.errorhandler(InvalidHeaderError) 

154 def handle_invalid_header_error(e): 

155 return self._invalid_token_callback(str(e)) 

156 

157 @app.errorhandler(InvalidTokenError) 

158 def handle_invalid_token_error(e): 

159 return self._invalid_token_callback(str(e)) 

160 

161 @app.errorhandler(JWTDecodeError) 

162 def handle_jwt_decode_error(e): 

163 return self._invalid_token_callback(str(e)) 

164 

165 @app.errorhandler(NoAuthorizationError) 

166 def handle_auth_error(e): 

167 return self._unauthorized_callback(str(e)) 

168 

169 @app.errorhandler(InvalidQueryParamError) 

170 def handle_invalid_query_param_error(e): 

171 return self._invalid_token_callback(str(e)) 

172 

173 @app.errorhandler(RevokedTokenError) 

174 def handle_revoked_token_error(e): 

175 return self._revoked_token_callback(e.jwt_header, e.jwt_data) 

176 

177 @app.errorhandler(UserClaimsVerificationError) 

178 def handle_failed_token_verification(e): 

179 return self._token_verification_failed_callback(e.jwt_header, e.jwt_data) 

180 

181 @app.errorhandler(UserLookupError) 

182 def handler_user_lookup_error(e): 

183 return self._user_lookup_error_callback(e.jwt_header, e.jwt_data) 

184 

185 @app.errorhandler(WrongTokenError) 

186 def handle_wrong_token_error(e): 

187 return self._invalid_token_callback(str(e)) 

188 

189 @staticmethod 

190 def _set_default_configuration_options(app: Flask) -> None: 

191 app.config.setdefault( 

192 "JWT_ACCESS_TOKEN_EXPIRES", datetime.timedelta(minutes=15) 

193 ) 

194 app.config.setdefault("JWT_ACCESS_COOKIE_NAME", "access_token_cookie") 

195 app.config.setdefault("JWT_ACCESS_COOKIE_PATH", "/") 

196 app.config.setdefault("JWT_ACCESS_CSRF_COOKIE_NAME", "csrf_access_token") 

197 app.config.setdefault("JWT_ACCESS_CSRF_COOKIE_PATH", "/") 

198 app.config.setdefault("JWT_ACCESS_CSRF_FIELD_NAME", "csrf_token") 

199 app.config.setdefault("JWT_ACCESS_CSRF_HEADER_NAME", "X-CSRF-TOKEN") 

200 app.config.setdefault("JWT_ALGORITHM", "HS256") 

201 app.config.setdefault("JWT_COOKIE_CSRF_PROTECT", True) 

202 app.config.setdefault("JWT_COOKIE_DOMAIN", None) 

203 app.config.setdefault("JWT_COOKIE_SAMESITE", None) 

204 app.config.setdefault("JWT_COOKIE_SECURE", False) 

205 app.config.setdefault("JWT_CSRF_CHECK_FORM", False) 

206 app.config.setdefault("JWT_CSRF_IN_COOKIES", True) 

207 app.config.setdefault("JWT_CSRF_METHODS", ["POST", "PUT", "PATCH", "DELETE"]) 

208 app.config.setdefault("JWT_DECODE_ALGORITHMS", None) 

209 app.config.setdefault("JWT_DECODE_AUDIENCE", None) 

210 app.config.setdefault("JWT_DECODE_ISSUER", None) 

211 app.config.setdefault("JWT_DECODE_LEEWAY", 0) 

212 app.config.setdefault("JWT_ENCODE_AUDIENCE", None) 

213 app.config.setdefault("JWT_ENCODE_ISSUER", None) 

214 app.config.setdefault("JWT_ERROR_MESSAGE_KEY", "msg") 

215 app.config.setdefault("JWT_HEADER_NAME", "Authorization") 

216 app.config.setdefault("JWT_HEADER_TYPE", "Bearer") 

217 app.config.setdefault("JWT_IDENTITY_CLAIM", "sub") 

218 app.config.setdefault("JWT_JSON_KEY", "access_token") 

219 app.config.setdefault("JWT_PRIVATE_KEY", None) 

220 app.config.setdefault("JWT_PUBLIC_KEY", None) 

221 app.config.setdefault("JWT_QUERY_STRING_NAME", "jwt") 

222 app.config.setdefault("JWT_QUERY_STRING_VALUE_PREFIX", "") 

223 app.config.setdefault("JWT_REFRESH_COOKIE_NAME", "refresh_token_cookie") 

224 app.config.setdefault("JWT_REFRESH_COOKIE_PATH", "/") 

225 app.config.setdefault("JWT_REFRESH_CSRF_COOKIE_NAME", "csrf_refresh_token") 

226 app.config.setdefault("JWT_REFRESH_CSRF_COOKIE_PATH", "/") 

227 app.config.setdefault("JWT_REFRESH_CSRF_FIELD_NAME", "csrf_token") 

228 app.config.setdefault("JWT_REFRESH_CSRF_HEADER_NAME", "X-CSRF-TOKEN") 

229 app.config.setdefault("JWT_REFRESH_JSON_KEY", "refresh_token") 

230 app.config.setdefault("JWT_REFRESH_TOKEN_EXPIRES", datetime.timedelta(days=30)) 

231 app.config.setdefault("JWT_SECRET_KEY", None) 

232 app.config.setdefault("JWT_SESSION_COOKIE", True) 

233 app.config.setdefault("JWT_TOKEN_LOCATION", ("headers",)) 

234 app.config.setdefault("JWT_VERIFY_SUB", True) 

235 app.config.setdefault("JWT_ENCODE_NBF", True) 

236 

237 def additional_claims_loader(self, callback: _F) -> _F: 

238 """ 

239 This decorator sets the callback function used to add additional claims 

240 when creating a JWT. The claims returned by this function will be merged 

241 with any claims passed in via the ``additional_claims`` argument to 

242 :func:`~flask_jwt_extended.create_access_token` or 

243 :func:`~flask_jwt_extended.create_refresh_token`. 

244 

245 The decorated function must take **one** argument. 

246 

247 The argument is the identity that was used when creating a JWT. 

248 

249 The decorated function must return a dictionary of claims to add to the JWT. 

250 """ 

251 self._user_claims_callback = callback 

252 return callback 

253 

254 def additional_headers_loader(self, callback: _F) -> _F: 

255 """ 

256 This decorator sets the callback function used to add additional headers 

257 when creating a JWT. The headers returned by this function will be merged 

258 with any headers passed in via the ``additional_headers`` argument to 

259 :func:`~flask_jwt_extended.create_access_token` or 

260 :func:`~flask_jwt_extended.create_refresh_token`. 

261 

262 The decorated function must take **one** argument. 

263 

264 The argument is the identity that was used when creating a JWT. 

265 

266 The decorated function must return a dictionary of headers to add to the JWT. 

267 """ 

268 self._jwt_additional_header_callback = callback 

269 return callback 

270 

271 def decode_key_loader(self, callback: _F) -> _F: 

272 """ 

273 This decorator sets the callback function for dynamically setting the JWT 

274 decode key based on the **UNVERIFIED** contents of the token. Think 

275 carefully before using this functionality, in most cases you probably 

276 don't need it. 

277 

278 The decorated function must take **two** arguments. 

279 

280 The first argument is a dictionary containing the header data of the 

281 unverified JWT. 

282 

283 The second argument is a dictionary containing the payload data of the 

284 unverified JWT. 

285 

286 The decorated function must return a *string* that is used to decode and 

287 verify the token. 

288 """ 

289 self._decode_key_callback = callback 

290 return callback 

291 

292 def encode_key_loader(self, callback: _F) -> _F: 

293 """ 

294 This decorator sets the callback function for dynamically setting the JWT 

295 encode key based on the tokens identity. Think carefully before using this 

296 functionality, in most cases you probably don't need it. 

297 

298 The decorated function must take **one** argument. 

299 

300 The argument is the identity used to create this JWT. 

301 

302 The decorated function must return a *string* which is the secrete key used to 

303 encode the JWT. 

304 """ 

305 self._encode_key_callback = callback 

306 return callback 

307 

308 def expired_token_loader(self, callback: _F) -> _F: 

309 """ 

310 This decorator sets the callback function for returning a custom 

311 response when an expired JWT is encountered. 

312 

313 The decorated function must take **two** arguments. 

314 

315 The first argument is a dictionary containing the header data of the JWT. 

316 

317 The second argument is a dictionary containing the payload data of the JWT. 

318 

319 The decorated function must return a Flask Response. 

320 """ 

321 self._expired_token_callback = callback 

322 return callback 

323 

324 def invalid_token_loader(self, callback: _F) -> _F: 

325 """ 

326 This decorator sets the callback function for returning a custom 

327 response when an invalid JWT is encountered. 

328 

329 This decorator sets the callback function that will be used if an 

330 invalid JWT attempts to access a protected endpoint. 

331 

332 The decorated function must take **one** argument. 

333 

334 The argument is a string which contains the reason why a token is invalid. 

335 

336 The decorated function must return a Flask Response. 

337 """ 

338 self._invalid_token_callback = callback 

339 return callback 

340 

341 def needs_fresh_token_loader(self, callback: _F) -> _F: 

342 """ 

343 This decorator sets the callback function for returning a custom 

344 response when a valid and non-fresh token is used on an endpoint 

345 that is marked as ``fresh=True``. 

346 

347 The decorated function must take **two** arguments. 

348 

349 The first argument is a dictionary containing the header data of the JWT. 

350 

351 The second argument is a dictionary containing the payload data of the JWT. 

352 

353 The decorated function must return a Flask Response. 

354 """ 

355 self._needs_fresh_token_callback = callback 

356 return callback 

357 

358 def revoked_token_loader(self, callback: _F) -> _F: 

359 """ 

360 This decorator sets the callback function for returning a custom 

361 response when a revoked token is encountered. 

362 

363 The decorated function must take **two** arguments. 

364 

365 The first argument is a dictionary containing the header data of the JWT. 

366 

367 The second argument is a dictionary containing the payload data of the JWT. 

368 

369 The decorated function must return a Flask Response. 

370 """ 

371 self._revoked_token_callback = callback 

372 return callback 

373 

374 def token_in_blocklist_loader(self, callback: _F) -> _F: 

375 """ 

376 This decorator sets the callback function used to check if a JWT has 

377 been revoked. 

378 

379 The decorated function must take **two** arguments. 

380 

381 The first argument is a dictionary containing the header data of the JWT. 

382 

383 The second argument is a dictionary containing the payload data of the JWT. 

384 

385 The decorated function must be return ``True`` if the token has been 

386 revoked, ``False`` otherwise. 

387 """ 

388 self._token_in_blocklist_callback = callback 

389 return callback 

390 

391 def token_verification_failed_loader(self, callback: _F) -> _F: 

392 """ 

393 This decorator sets the callback function used to return a custom 

394 response when the claims verification check fails. 

395 

396 The decorated function must take **two** arguments. 

397 

398 The first argument is a dictionary containing the header data of the JWT. 

399 

400 The second argument is a dictionary containing the payload data of the JWT. 

401 

402 The decorated function must return a Flask Response. 

403 """ 

404 self._token_verification_failed_callback = callback 

405 return callback 

406 

407 def token_verification_loader(self, callback: _F) -> _F: 

408 """ 

409 This decorator sets the callback function used for custom verification 

410 of a valid JWT. 

411 

412 The decorated function must take **two** arguments. 

413 

414 The first argument is a dictionary containing the header data of the JWT. 

415 

416 The second argument is a dictionary containing the payload data of the JWT. 

417 

418 The decorated function must return ``True`` if the token is valid, or 

419 ``False`` otherwise. 

420 """ 

421 self._token_verification_callback = callback 

422 return callback 

423 

424 def unauthorized_loader(self, callback: _F) -> _F: 

425 """ 

426 This decorator sets the callback function used to return a custom 

427 response when no JWT is present. 

428 

429 The decorated function must take **one** argument. 

430 

431 The argument is a string that explains why the JWT could not be found. 

432 

433 The decorated function must return a Flask Response. 

434 """ 

435 self._unauthorized_callback = callback 

436 return callback 

437 

438 def user_identity_loader(self, callback: _F) -> _F: 

439 """ 

440 This decorator sets the callback function used to convert an identity to 

441 a string when creating JWTs. This is useful for using objects (such as 

442 SQLAlchemy instances) as the identity when creating your tokens. 

443 

444 The decorated function must take **one** argument. 

445 

446 The argument is the identity that was used when creating a JWT. 

447 

448 The decorated function must return a string. 

449 """ 

450 self._user_identity_callback = callback 

451 return callback 

452 

453 def user_lookup_loader(self, callback: _F) -> _F: 

454 """ 

455 This decorator sets the callback function used to convert a JWT into 

456 a python object that can be used in a protected endpoint. This is useful 

457 for automatically loading a SQLAlchemy instance based on the contents 

458 of the JWT. 

459 

460 The object returned from this function can be accessed via 

461 :attr:`~flask_jwt_extended.current_user` or 

462 :meth:`~flask_jwt_extended.get_current_user` 

463 

464 The decorated function must take **two** arguments. 

465 

466 The first argument is a dictionary containing the header data of the JWT. 

467 

468 The second argument is a dictionary containing the payload data of the JWT. 

469 

470 The decorated function can return any python object, which can then be 

471 accessed in a protected endpoint. If an object cannot be loaded, for 

472 example if a user has been deleted from your database, ``None`` must be 

473 returned to indicate that an error occurred loading the user. 

474 """ 

475 self._user_lookup_callback = callback 

476 return callback 

477 

478 def user_lookup_error_loader(self, callback: _F) -> _F: 

479 """ 

480 This decorator sets the callback function used to return a custom 

481 response when loading a user via 

482 :meth:`~flask_jwt_extended.JWTManager.user_lookup_loader` fails. 

483 

484 The decorated function must take **two** arguments. 

485 

486 The first argument is a dictionary containing the header data of the JWT. 

487 

488 The second argument is a dictionary containing the payload data of the JWT. 

489 

490 The decorated function must return a Flask Response. 

491 """ 

492 self._user_lookup_error_callback = callback 

493 return callback 

494 

495 def _encode_jwt_from_config( 

496 self, 

497 identity: Any, 

498 token_type: str, 

499 claims=None, 

500 fresh: Fresh = False, 

501 expires_delta: Optional[ExpiresDelta] = None, 

502 headers=None, 

503 ) -> str: 

504 header_overrides = self._jwt_additional_header_callback(identity) 

505 if headers is not None: 

506 header_overrides.update(headers) 

507 

508 claim_overrides = self._user_claims_callback(identity) 

509 if claims is not None: 

510 claim_overrides.update(claims) 

511 

512 if expires_delta is None: 

513 if token_type == "access": 

514 expires_delta = config.access_expires 

515 else: 

516 expires_delta = config.refresh_expires 

517 

518 return _encode_jwt( 

519 algorithm=config.algorithm, 

520 audience=config.encode_audience, 

521 claim_overrides=claim_overrides, 

522 csrf=config.cookie_csrf_protect, 

523 expires_delta=expires_delta, 

524 fresh=fresh, 

525 header_overrides=header_overrides, 

526 identity=self._user_identity_callback(identity), 

527 identity_claim_key=config.identity_claim_key, 

528 issuer=config.encode_issuer, 

529 json_encoder=config.json_encoder, 

530 secret=self._encode_key_callback(identity), 

531 token_type=token_type, 

532 nbf=config.encode_nbf, 

533 ) 

534 

535 def _decode_jwt_from_config( 

536 self, encoded_token: str, csrf_value=None, allow_expired: bool = False 

537 ) -> dict: 

538 unverified_claims = jwt.decode( 

539 encoded_token, 

540 algorithms=config.decode_algorithms, 

541 options={"verify_signature": False}, 

542 ) 

543 unverified_headers = jwt.get_unverified_header(encoded_token) 

544 secret = self._decode_key_callback(unverified_headers, unverified_claims) 

545 

546 kwargs = { 

547 "algorithms": config.decode_algorithms, 

548 "audience": config.decode_audience, 

549 "csrf_value": csrf_value, 

550 "encoded_token": encoded_token, 

551 "identity_claim_key": config.identity_claim_key, 

552 "issuer": config.decode_issuer, 

553 "leeway": config.leeway, 

554 "secret": secret, 

555 "verify_aud": config.decode_audience is not None, 

556 "verify_sub": config.verify_sub, 

557 } 

558 

559 try: 

560 return _decode_jwt(**kwargs, allow_expired=allow_expired) 

561 except ExpiredSignatureError as e: 

562 # TODO: If we ever do another breaking change, don't raise this pyjwt 

563 # error directly, instead raise a custom error of ours from this 

564 # error. 

565 e.jwt_header = unverified_headers # type: ignore 

566 e.jwt_data = _decode_jwt(**kwargs, allow_expired=True) # type: ignore 

567 raise