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
1from typing import Any
2from typing import Optional
4import jwt
5from flask import g
6from flask import Response
7from werkzeug.local import LocalProxy
9from flask_jwt_extended.config import config
10from flask_jwt_extended.internal_utils import get_jwt_manager
11from flask_jwt_extended.typing import ExpiresDelta
12from flask_jwt_extended.typing import Fresh
14# Proxy to access the current user
15current_user: Any = LocalProxy(lambda: get_current_user())
18def get_jwt() -> dict:
19 """
20 In a protected endpoint, this will return the python dictionary which has
21 the payload of the JWT that is accessing the endpoint. If no JWT is present
22 due to ``jwt_required(optional=True)``, an empty dictionary is returned.
24 :return:
25 The payload (claims) of the JWT in the current request
26 """
27 decoded_jwt = g.get("_jwt_extended_jwt", None)
28 if decoded_jwt is None:
29 raise RuntimeError(
30 "You must call `@jwt_required()` or `verify_jwt_in_request()` "
31 "before using this method"
32 )
33 return decoded_jwt
36def get_jwt_header() -> dict:
37 """
38 In a protected endpoint, this will return the python dictionary which has
39 the header of the JWT that is accessing the endpoint. If no JWT is present
40 due to ``jwt_required(optional=True)``, an empty dictionary is returned.
42 :return:
43 The headers of the JWT in the current request
44 """
45 decoded_header = g.get("_jwt_extended_jwt_header", None)
46 if decoded_header is None:
47 raise RuntimeError(
48 "You must call `@jwt_required()` or `verify_jwt_in_request()` "
49 "before using this method"
50 )
51 return decoded_header
54def get_jwt_identity() -> Any:
55 """
56 In a protected endpoint, this will return the identity of the JWT that is
57 accessing the endpoint. If no JWT is present due to
58 ``jwt_required(optional=True)``, ``None`` is returned.
60 :return:
61 The identity of the JWT in the current request
62 """
63 return get_jwt().get(config.identity_claim_key, None)
66def get_jwt_request_location() -> Optional[str]:
67 """
68 In a protected endpoint, this will return the "location" at which the JWT
69 that is accessing the endpoint was found--e.g., "cookies", "query-string",
70 "headers", or "json". If no JWT is present due to ``jwt_required(optional=True)``,
71 None is returned.
73 :return:
74 The location of the JWT in the current request; e.g., "cookies",
75 "query-string", "headers", or "json"
76 """
77 return g.get("_jwt_extended_jwt_location", None)
80def get_current_user() -> Any:
81 """
82 In a protected endpoint, this will return the user object for the JWT that
83 is accessing the endpoint.
85 This is only usable if :meth:`~flask_jwt_extended.JWTManager.user_lookup_loader`
86 is configured. If the user loader callback is not being used, this will
87 raise an error.
89 If no JWT is present due to ``jwt_required(optional=True)``, ``None`` is returned.
91 :return:
92 The current user object for the JWT in the current request
93 """
94 get_jwt() # Raise an error if not in a decorated context
95 jwt_user_dict = g.get("_jwt_extended_jwt_user", None)
96 if jwt_user_dict is None:
97 raise RuntimeError(
98 "You must provide a `@jwt.user_lookup_loader` callback to use "
99 "this method"
100 )
101 return jwt_user_dict["loaded_user"]
104def decode_token(
105 encoded_token: str, csrf_value: Optional[str] = None, allow_expired: bool = False
106) -> dict:
107 """
108 Returns the decoded token (python dict) from an encoded JWT. This does all
109 the checks to ensure that the decoded token is valid before returning it.
111 This will not fire the user loader callbacks, save the token for access
112 in protected endpoints, checked if a token is revoked, etc. This is puerly
113 used to ensure that a JWT is valid.
115 :param encoded_token:
116 The encoded JWT to decode.
118 :param csrf_value:
119 Expected CSRF double submit value (optional).
121 :param allow_expired:
122 If ``True``, do not raise an error if the JWT is expired. Defaults to ``False``
124 :return:
125 Dictionary containing the payload of the JWT decoded JWT.
126 """
127 jwt_manager = get_jwt_manager()
128 return jwt_manager._decode_jwt_from_config(encoded_token, csrf_value, allow_expired)
131def create_access_token(
132 identity: Any,
133 fresh: Fresh = False,
134 expires_delta: Optional[ExpiresDelta] = None,
135 additional_claims=None,
136 additional_headers=None,
137):
138 """
139 Create a new access token.
141 :param identity:
142 The identity of this token. This must either be a string, or you must have
143 defined :meth:`~flask_jwt_extended.JWTManager.user_identity_loader` in order
144 to convert the object you passed in into a string.
146 :param fresh:
147 If this token should be marked as fresh, and can thus access endpoints
148 protected with ``@jwt_required(fresh=True)``. Defaults to ``False``.
150 This value can also be a ``datetime.timedelta``, which indicate
151 how long this token will be considered fresh.
153 :param expires_delta:
154 A ``datetime.timedelta`` for how long this token should last before it
155 expires. Set to False to disable expiration. If this is None, it will use
156 the ``JWT_ACCESS_TOKEN_EXPIRES`` config value (see :ref:`Configuration Options`)
158 :param additional_claims:
159 Optional. A hash of claims to include in the access token. These claims are
160 merged into the default claims (exp, iat, etc) and claims returned from the
161 :meth:`~flask_jwt_extended.JWTManager.additional_claims_loader` callback.
162 On conflict, these claims take precedence.
164 :param headers:
165 Optional. A hash of headers to include in the access token. These headers
166 are merged into the default headers (alg, typ) and headers returned from
167 the :meth:`~flask_jwt_extended.JWTManager.additional_headers_loader`
168 callback. On conflict, these headers take precedence.
170 :return:
171 An encoded access token
172 """
173 jwt_manager = get_jwt_manager()
174 return jwt_manager._encode_jwt_from_config(
175 claims=additional_claims,
176 expires_delta=expires_delta,
177 fresh=fresh,
178 headers=additional_headers,
179 identity=identity,
180 token_type="access",
181 )
184def create_refresh_token(
185 identity: Any,
186 expires_delta: Optional[ExpiresDelta] = None,
187 additional_claims=None,
188 additional_headers=None,
189):
190 """
191 Create a new refresh token.
193 :param identity:
194 The identity of this token. This must either be a string, or you must have
195 defined :meth:`~flask_jwt_extended.JWTManager.user_identity_loader` in order
196 to convert the object you passed in into a string.
198 :param expires_delta:
199 A ``datetime.timedelta`` for how long this token should last before it expires.
200 Set to False to disable expiration. If this is None, it will use the
201 ``JWT_REFRESH_TOKEN_EXPIRES`` config value (see :ref:`Configuration Options`)
203 :param additional_claims:
204 Optional. A hash of claims to include in the refresh token. These claims are
205 merged into the default claims (exp, iat, etc) and claims returned from the
206 :meth:`~flask_jwt_extended.JWTManager.additional_claims_loader` callback.
207 On conflict, these claims take precedence.
209 :param headers:
210 Optional. A hash of headers to include in the refresh token. These headers
211 are merged into the default headers (alg, typ) and headers returned from the
212 :meth:`~flask_jwt_extended.JWTManager.additional_headers_loader` callback.
213 On conflict, these headers take precedence.
215 :return:
216 An encoded refresh token
217 """
218 jwt_manager = get_jwt_manager()
219 return jwt_manager._encode_jwt_from_config(
220 claims=additional_claims,
221 expires_delta=expires_delta,
222 fresh=False,
223 headers=additional_headers,
224 identity=identity,
225 token_type="refresh",
226 )
229def get_unverified_jwt_headers(encoded_token: str) -> dict:
230 """
231 Returns the Headers of an encoded JWT without verifying the signature of the JWT.
233 :param encoded_token:
234 The encoded JWT to get the Header from.
236 :return:
237 JWT header parameters as python dict()
238 """
239 return jwt.get_unverified_header(encoded_token)
242def get_jti(encoded_token: str) -> Optional[str]:
243 """
244 Returns the JTI (unique identifier) of an encoded JWT
246 :param encoded_token:
247 The encoded JWT to get the JTI from.
249 :return:
250 The JTI (unique identifier) of a JWT, if it is present.
251 """
252 return decode_token(encoded_token).get("jti")
255def get_csrf_token(encoded_token: str) -> str:
256 """
257 Returns the CSRF double submit token from an encoded JWT.
259 :param encoded_token:
260 The encoded JWT
262 :return:
263 The CSRF double submit token (string)
264 """
265 token = decode_token(encoded_token)
266 return token["csrf"]
269def set_access_cookies(
270 response: Response, encoded_access_token: str, max_age=None, domain=None
271) -> None:
272 """
273 Modifiy a Flask Response to set a cookie containing the access JWT.
274 Also sets the corresponding CSRF cookies if ``JWT_CSRF_IN_COOKIES`` is ``True``
275 (see :ref:`Configuration Options`)
277 :param response:
278 A Flask Response object.
280 :param encoded_access_token:
281 The encoded access token to set in the cookies.
283 :param max_age:
284 The max age of the cookie. If this is None, it will use the
285 ``JWT_SESSION_COOKIE`` option (see :ref:`Configuration Options`). Otherwise,
286 it will use this as the cookies ``max-age`` and the JWT_SESSION_COOKIE option
287 will be ignored. Values should be the number of seconds (as an integer).
289 :param domain:
290 The domain of the cookie. If this is None, it will use the
291 ``JWT_COOKIE_DOMAIN`` option (see :ref:`Configuration Options`). Otherwise,
292 it will use this as the cookies ``domain`` and the JWT_COOKIE_DOMAIN option
293 will be ignored.
294 """
295 response.set_cookie(
296 config.access_cookie_name,
297 value=encoded_access_token,
298 max_age=max_age or config.cookie_max_age,
299 secure=config.cookie_secure,
300 httponly=True,
301 domain=domain or config.cookie_domain,
302 path=config.access_cookie_path,
303 samesite=config.cookie_samesite,
304 )
306 if config.cookie_csrf_protect and config.csrf_in_cookies:
307 response.set_cookie(
308 config.access_csrf_cookie_name,
309 value=get_csrf_token(encoded_access_token),
310 max_age=max_age or config.cookie_max_age,
311 secure=config.cookie_secure,
312 httponly=False,
313 domain=domain or config.cookie_domain,
314 path=config.access_csrf_cookie_path,
315 samesite=config.cookie_samesite,
316 )
319def set_refresh_cookies(
320 response: Response,
321 encoded_refresh_token: str,
322 max_age: Optional[int] = None,
323 domain: Optional[str] = None,
324) -> None:
325 """
326 Modifiy a Flask Response to set a cookie containing the refresh JWT.
327 Also sets the corresponding CSRF cookies if ``JWT_CSRF_IN_COOKIES`` is ``True``
328 (see :ref:`Configuration Options`)
330 :param response:
331 A Flask Response object.
333 :param encoded_refresh_token:
334 The encoded refresh token to set in the cookies.
336 :param max_age:
337 The max age of the cookie. If this is None, it will use the
338 ``JWT_SESSION_COOKIE`` option (see :ref:`Configuration Options`). Otherwise,
339 it will use this as the cookies ``max-age`` and the JWT_SESSION_COOKIE option
340 will be ignored. Values should be the number of seconds (as an integer).
342 :param domain:
343 The domain of the cookie. If this is None, it will use the
344 ``JWT_COOKIE_DOMAIN`` option (see :ref:`Configuration Options`). Otherwise,
345 it will use this as the cookies ``domain`` and the JWT_COOKIE_DOMAIN option
346 will be ignored.
347 """
348 response.set_cookie(
349 config.refresh_cookie_name,
350 value=encoded_refresh_token,
351 max_age=max_age or config.cookie_max_age,
352 secure=config.cookie_secure,
353 httponly=True,
354 domain=domain or config.cookie_domain,
355 path=config.refresh_cookie_path,
356 samesite=config.cookie_samesite,
357 )
359 if config.cookie_csrf_protect and config.csrf_in_cookies:
360 response.set_cookie(
361 config.refresh_csrf_cookie_name,
362 value=get_csrf_token(encoded_refresh_token),
363 max_age=max_age or config.cookie_max_age,
364 secure=config.cookie_secure,
365 httponly=False,
366 domain=domain or config.cookie_domain,
367 path=config.refresh_csrf_cookie_path,
368 samesite=config.cookie_samesite,
369 )
372def unset_jwt_cookies(response: Response, domain: Optional[str] = None) -> None:
373 """
374 Modifiy a Flask Response to delete the cookies containing access or refresh
375 JWTs. Also deletes the corresponding CSRF cookies if applicable.
377 :param response:
378 A Flask Response object
379 """
380 unset_access_cookies(response, domain)
381 unset_refresh_cookies(response, domain)
384def unset_access_cookies(response: Response, domain: Optional[str] = None) -> None:
385 """
386 Modifiy a Flask Response to delete the cookie containing an access JWT.
387 Also deletes the corresponding CSRF cookie if applicable.
389 :param response:
390 A Flask Response object
392 :param domain:
393 The domain of the cookie. If this is None, it will use the
394 ``JWT_COOKIE_DOMAIN`` option (see :ref:`Configuration Options`). Otherwise,
395 it will use this as the cookies ``domain`` and the JWT_COOKIE_DOMAIN option
396 will be ignored.
397 """
398 response.set_cookie(
399 config.access_cookie_name,
400 value="",
401 expires=0,
402 secure=config.cookie_secure,
403 httponly=True,
404 domain=domain or config.cookie_domain,
405 path=config.access_cookie_path,
406 samesite=config.cookie_samesite,
407 )
409 if config.cookie_csrf_protect and config.csrf_in_cookies:
410 response.set_cookie(
411 config.access_csrf_cookie_name,
412 value="",
413 expires=0,
414 secure=config.cookie_secure,
415 httponly=False,
416 domain=domain or config.cookie_domain,
417 path=config.access_csrf_cookie_path,
418 samesite=config.cookie_samesite,
419 )
422def unset_refresh_cookies(response: Response, domain: Optional[str] = None) -> None:
423 """
424 Modifiy a Flask Response to delete the cookie containing a refresh JWT.
425 Also deletes the corresponding CSRF cookie if applicable.
427 :param response:
428 A Flask Response object
430 :param domain:
431 The domain of the cookie. If this is None, it will use the
432 ``JWT_COOKIE_DOMAIN`` option (see :ref:`Configuration Options`). Otherwise,
433 it will use this as the cookies ``domain`` and the JWT_COOKIE_DOMAIN option
434 will be ignored.
435 """
436 response.set_cookie(
437 config.refresh_cookie_name,
438 value="",
439 expires=0,
440 secure=config.cookie_secure,
441 httponly=True,
442 domain=domain or config.cookie_domain,
443 path=config.refresh_cookie_path,
444 samesite=config.cookie_samesite,
445 )
447 if config.cookie_csrf_protect and config.csrf_in_cookies:
448 response.set_cookie(
449 config.refresh_csrf_cookie_name,
450 value="",
451 expires=0,
452 secure=config.cookie_secure,
453 httponly=False,
454 domain=domain or config.cookie_domain,
455 path=config.refresh_csrf_cookie_path,
456 samesite=config.cookie_samesite,
457 )
460def current_user_context_processor() -> Any:
461 return {"current_user": get_current_user()}