<gh_stars>0
import jwt, re, uuid, hmac
from jwt.algorithms import requires_cryptography, has_crypto
from datetime import datetime, timezone, timedelta
from typing import Optional, Dict, Union, Sequence
from types import GeneratorType
from fastapi import Request, Response
from fastapi_jwt_auth.auth_config import AuthConfig
from fastapi_jwt_auth.exceptions import (
    InvalidHeaderError,
    CSRFError,
    JWTDecodeError,
    RevokedTokenError,
    MissingHeaderError,
    MissingCookieError,
    AccessTokenRequired,
    RefreshTokenRequired,
    FreshTokenRequired
)

class AuthJWT(AuthConfig):
    def __init__(self,req: Request = None, res: Response = None):
        """
        Get jwt header from incoming request or get
        request and response object if jwt in the cookie

        :param req: all incoming request
        :param res: response from endpoint
        """
        if res and self.jwt_in_cookies:
            self._response = res

        if req:
            # get request object when cookies in token location
            if self.jwt_in_cookies:
                self._request = req
            # get jwt in headers when headers in token location
            if self.jwt_in_headers:
                auth = req.headers.get(self._header_name.lower())
                if auth: self._get_jwt_from_headers(auth)

    def _get_jwt_from_headers(self,auth: str) -> "AuthJWT":
        """
        Get token from the headers

        :param auth: value from HeaderName
        """
        header_name, header_type = self._header_name, self._header_type

        parts = auth.split()

        # Make sure the header is in a valid format that we are expecting, ie
        if not header_type:
            # <HeaderName>: <JWT>
            if len(parts) != 1:
                msg = "Bad {} header. Expected value '<JWT>'".format(header_name)
                raise InvalidHeaderError(status_code=422,message=msg)
            self._token = parts[0]
        else:
            # <HeaderName>: <HeaderType> <JWT>
            if not re.match(r"{}\s".format(header_type),auth) or len(parts) != 2:
                msg = "Bad {} header. Expected value '{} <JWT>'".format(header_name,header_type)
                raise InvalidHeaderError(status_code=422,message=msg)
            self._token = parts[1]

    def _get_jwt_identifier(self) -> str:
        return str(uuid.uuid4())

    def _get_int_from_datetime(self,value: datetime) -> int:
        """
        :param value: datetime with or without timezone, if don't contains timezone
                      it will managed as it is UTC
        :return: Seconds since the Epoch
        """
        if not isinstance(value, datetime):  # pragma: no cover
            raise TypeError('a datetime is required')
        return int(value.timestamp())

    def _get_secret_key(self, algorithm: str, process: str) -> str:
        """
        Get key with a different algorithm

        :param algorithm: algorithm for decode and encode token
        :param process: for indicating get key for encode or decode token

        :return: plain text or RSA depends on algorithm
        """
        symmetric_algorithms, asymmetric_algorithms = {"HS256","HS384","HS512"}, requires_cryptography

        if algorithm not in symmetric_algorithms and algorithm not in asymmetric_algorithms:
            raise ValueError("Algorithm {} could not be found".format(algorithm))

        if algorithm in symmetric_algorithms:
            if not self._secret_key:
                raise RuntimeError(
                    "authjwt_secret_key must be set when using symmetric algorithm {}".format(algorithm)
                )

            return self._secret_key

        if algorithm in asymmetric_algorithms and not has_crypto:
            raise RuntimeError(
                "Missing dependencies for using asymmetric algorithms. run 'pip install fastapi-jwt-auth[asymmetric]'"
            )

        if process == "encode":
            if not self._private_key:
                raise RuntimeError(
                    "authjwt_private_key must be set when using asymmetric algorithm {}".format(algorithm)
                )

            return self._private_key

        if process == "decode":
            if not self._public_key:
                raise RuntimeError(
                    "authjwt_public_key must be set when using asymmetric algorithm {}".format(algorithm)
                )

            return self._public_key

    def _create_token(
        self,
        subject: Union[str,int],
        type_token: str,
        exp_time: Optional[int],
        fresh: Optional[bool] = False,
        algorithm: Optional[str] = None,
        headers: Optional[Dict] = None,
        issuer: Optional[str] = None,
        audience: Optional[Union[str,Sequence[str]]] = None,
        user_claims: Optional[Dict] = {}
    ) -> str:
        """
        Create token for access_token and refresh_token (utf-8)

        :param subject: Identifier for who this token is for example id or username from database.
        :param type_token: indicate token is access_token or refresh_token
        :param exp_time: Set the duration of the JWT
        :param fresh: Optional when token is access_token this param required
        :param algorithm: algorithm allowed to encode the token
        :param headers: valid dict for specifying additional headers in JWT header section
        :param issuer: expected issuer in the JWT
        :param audience: expected audience in the JWT
        :param user_claims: Custom claims to include in this token. This data must be dictionary

        :return: Encoded token
        """
        # Validation type data
        if not isinstance(subject, (str,int)):
            raise TypeError("subject must be a string or integer")
        if not isinstance(fresh, (bool)):
            raise TypeError("fresh must be a boolean")
        if audience and not isinstance(audience, (str, list, tuple, set, frozenset, GeneratorType)):
            raise TypeError("audience must be a string or sequence")
        if algorithm and not isinstance(algorithm, str):
            raise TypeError("algorithm must be a string")
        if user_claims and not isinstance(user_claims, dict):
            raise TypeError("user_claims must be a dictionary")

        # Data section
        reserved_claims = {
            "sub": subject,
            "iat": self._get_int_from_datetime(datetime.now(timezone.utc)),
            "nbf": self._get_int_from_datetime(datetime.now(timezone.utc)),
            "jti": self._get_jwt_identifier()
        }

        custom_claims = {"type": type_token}

        # for access_token only fresh needed
        if type_token == 'access':
            custom_claims['fresh'] = fresh
        # if cookie in token location and csrf protection enabled
        if self.jwt_in_cookies and self._cookie_csrf_protect:
            custom_claims['csrf'] = self._get_jwt_identifier()

        if exp_time:
            reserved_claims['exp'] = exp_time
        if issuer:
            reserved_claims['iss'] = issuer
        if audience:
            reserved_claims['aud'] = audience

        algorithm = algorithm or self._algorithm

        try:
            secret_key = self._get_secret_key(algorithm,"encode")
        except Exception:
            raise

        return jwt.encode(
            {**reserved_claims, **custom_claims, **user_claims},
            secret_key,
            algorithm=algorithm,
            headers=headers
        ).decode('utf-8')

    def _has_token_in_denylist_callback(self) -> bool:
        """
        Return True if token denylist callback set
        """
        return self._token_in_denylist_callback is not None

    def _check_token_is_revoked(self, raw_token: Dict[str,Union[str,int,bool]]) -> None:
        """
        Ensure that AUTHJWT_DENYLIST_ENABLED is true and callback regulated, and then
        call function denylist callback with passing decode JWT, if true
        raise exception Token has been revoked
        """
        if not self._denylist_enabled:
            return

        if not self._has_token_in_denylist_callback():
            raise RuntimeError("A token_in_denylist_callback must be provided via "
                "the '@AuthJWT.token_in_denylist_loader' if "
                "authjwt_denylist_enabled is 'True'")

        if self._token_in_denylist_callback.__func__(raw_token):
            raise RevokedTokenError(status_code=401,message="Token has been revoked")

    def _get_expired_time(
        self,
        type_token: str,
        expires_time: Optional[Union[timedelta,int,bool]] = None
    ) -> Union[None,int]:
        """
        Dynamic token expired if expires_time is False exp claim not created

        :param type_token: indicate token is access_token or refresh_token
        :param expires_time: duration expired jwt

        :return: duration exp claim jwt
        """
        if expires_time and not isinstance(expires_time, (timedelta,int,bool)):
            raise TypeError("expires_time must be between timedelta, int, bool")

        if expires_time is not False:
            if type_token == 'access':
                expires_time = expires_time or self._access_token_expires
            if type_token == 'refresh':
                expires_time = expires_time or self._refresh_token_expires

        if expires_time is not False:
            if isinstance(expires_time, bool):
                if type_token == 'access':
                    expires_time = self._access_token_expires
                if type_token == 'refresh':
                    expires_time = self._refresh_token_expires
            if isinstance(expires_time, timedelta):
                expires_time = int(expires_time.total_seconds())

            return self._get_int_from_datetime(datetime.now(timezone.utc)) + expires_time
        else:
            return None

    def create_access_token(
        self,
        subject: Union[str,int],
        fresh: Optional[bool] = False,
        algorithm: Optional[str] = None,
        headers: Optional[Dict] = None,
        expires_time: Optional[Union[timedelta,int,bool]] = None,
        audience: Optional[Union[str,Sequence[str]]] = None,
        user_claims: Optional[Dict] = {}
    ) -> str:
        """
        Create a access token with 15 minutes for expired time (default),
        info for param and return check to function create token

        :return: hash token
        """
        return self._create_token(
            subject=subject,
            type_token="access",
            exp_time=self._get_expired_time("access",expires_time),
            fresh=fresh,
            algorithm=algorithm,
            headers=headers,
            audience=audience,
            user_claims=user_claims,
            issuer=self._encode_issuer
        )

    def create_refresh_token(
        self,
        subject: Union[str,int],
        algorithm: Optional[str] = None,
        headers: Optional[Dict] = None,
        expires_time: Optional[Union[timedelta,int,bool]] = None,
        audience: Optional[Union[str,Sequence[str]]] = None,
        user_claims: Optional[Dict] = {}
    ) -> str:
        """
        Create a refresh token with 30 days for expired time (default),
        info for param and return check to function create token

        :return: hash token
        """
        return self._create_token(
            subject=subject,
            type_token="refresh",
            exp_time=self._get_expired_time("refresh",expires_time),
            algorithm=algorithm,
            headers=headers,
            audience=audience,
            user_claims=user_claims
        )

    def _get_csrf_token(self,encoded_token: str) -> str:
        """
        Returns the CSRF double submit token from an encoded JWT.

        :param encoded_token: The encoded JWT
        :return: The CSRF double submit token
        """
        return self._verified_token(encoded_token)['csrf']

    def set_access_cookies(self,encoded_access_token: str, max_age: Optional[int] = None) -> None:
        """
        Configures the response to set access token in a cookie.
        this will also set the CSRF double submit values in a separate cookie

        :param encoded_access_token: The encoded access token to set in the cookies
        :param max_age: The max age of the cookie value should be the number of seconds (integer)
        """
        if not self.jwt_in_cookies:
            raise RuntimeWarning(
                "set_access_cookies() called without 'authjwt_token_location' configured to use cookies"
            )

        if max_age and not isinstance(max_age,int):
            raise TypeError("max_age must be a integer")

        # Set the access JWT in the cookie
        self._response.set_cookie(
            self._access_cookie_key,
            encoded_access_token,
            max_age=max_age or self._cookie_max_age,
            path=self._access_cookie_path,
            domain=self._cookie_domain,
            secure=self._cookie_secure,
            httponly=True,
            samesite=self._cookie_samesite
        )

        # If enabled, set the csrf double submit access cookie
        if self._cookie_csrf_protect:
            self._response.set_cookie(
                self._access_csrf_cookie_key,
                self._get_csrf_token(encoded_access_token),
                max_age=max_age or self._cookie_max_age,
                path=self._access_csrf_cookie_path,
                domain=self._cookie_domain,
                secure=self._cookie_secure,
                httponly=False,
                samesite=self._cookie_samesite
            )

    def set_refresh_cookies(self, encoded_refresh_token: str, max_age: Optional[int] = None) -> None:
        """
        Configures the response to set refresh token in a cookie.
        this will also set the CSRF double submit values in a separate cookie

        :param encoded_refresh_token: The encoded refresh token to set in the cookies
        :param max_age: The max age of the cookie value should be the number of seconds (integer)
        """
        if not self.jwt_in_cookies:
            raise RuntimeWarning(
                "set_refresh_cookies() called without 'authjwt_token_location' configured to use cookies"
            )

        if max_age and not isinstance(max_age,int):
            raise TypeError("max_age must be a integer")

        # Set the refresh JWT in the cookie
        self._response.set_cookie(
            self._refresh_cookie_key,
            encoded_refresh_token,
            max_age=max_age or self._cookie_max_age,
            path=self._refresh_cookie_path,
            domain=self._cookie_domain,
            secure=self._cookie_secure,
            httponly=True,
            samesite=self._cookie_samesite
        )

        # If enabled, set the csrf double submit refresh cookie
        if self._cookie_csrf_protect:
            self._response.set_cookie(
                self._refresh_csrf_cookie_key,
                self._get_csrf_token(encoded_refresh_token),
                max_age=max_age or self._cookie_max_age,
                path=self._refresh_csrf_cookie_path,
                domain=self._cookie_domain,
                secure=self._cookie_secure,
                httponly=False,
                samesite=self._cookie_samesite
            )

    def unset_jwt_cookies(self) -> None:
        """
        Unset (delete) all jwt stored in a cookie
        """
        self.unset_access_cookies()
        self.unset_refresh_cookies()

    def unset_access_cookies(self) -> None:
        """
        Remove access token and access CSRF double submit from the response cookies
        """
        if not self.jwt_in_cookies:
            raise RuntimeWarning(
                "unset_access_cookies() called without 'authjwt_token_location' configured to use cookies"
            )

        self._response.delete_cookie(
            self._access_cookie_key,
            path=self._access_cookie_path,
            domain=self._cookie_domain
        )

        if self._cookie_csrf_protect:
            self._response.delete_cookie(
                self._access_csrf_cookie_key,
                path=self._access_csrf_cookie_path,
                domain=self._cookie_domain
            )

    def unset_refresh_cookies(self) -> None:
        """
        Remove refresh token and refresh CSRF double submit from the response cookies
        """
        if not self.jwt_in_cookies:
            raise RuntimeWarning(
                "unset_refresh_cookies() called without 'authjwt_token_location' configured to use cookies"
            )

        self._response.delete_cookie(
            self._refresh_cookie_key,
            path=self._refresh_cookie_path,
            domain=self._cookie_domain
        )

        if self._cookie_csrf_protect:
            self._response.delete_cookie(
                self._refresh_csrf_cookie_key,
                path=self._refresh_csrf_cookie_path,
                domain=self._cookie_domain
            )

    def _verify_and_get_jwt_optional_in_cookies(self,issuer: Optional[str] = None) -> "AuthJWT":
        """
        Optionally check if cookies have a valid access token. if an access token present in
        cookies property _token will set. raises exception error when an access token is invalid
        and doesn't match with CSRF token double submit

        :param issuer: expected issuer in the JWT
        """
        cookie_key = self._access_cookie_key
        cookie = self._request.cookies.get(cookie_key)
        csrf_cookie = self._request.headers.get(self._access_csrf_header_name)

        if (
            cookie and
            self._cookie_csrf_protect and
            self._request.method in self._csrf_methods and
            not csrf_cookie
        ):
            raise CSRFError(status_code=401,message="Missing CSRF Token")

        # set token from cookie and verify jwt
        self._token = cookie
        self._verify_jwt_optional_in_request(self._token,issuer)

        decoded_token = self.get_raw_jwt()

        if (
            self._cookie_csrf_protect and
            self._request.method in self._csrf_methods and
            csrf_cookie and
            decoded_token
        ):
            if 'csrf' not in decoded_token:
                raise JWTDecodeError(status_code=422,message="Missing claim: csrf")
            if not hmac.compare_digest(csrf_cookie,decoded_token['csrf']):
                raise CSRFError(status_code=401,message="CSRF double submit tokens do not match")

    def _verify_and_get_jwt_in_cookies(
        self,
        type_token: str,
        issuer: Optional[str] = None,
        fresh: Optional[bool] = False
    ) -> "AuthJWT":
        """
        Check if cookies have a valid access or refresh token. if an token present in
        cookies property _token will set. raises exception error when an access or refresh token
        is invalid and doesn't match with CSRF token double submit

        :param type_token: indicate token is access or refresh token
        :param issuer: expected issuer in the JWT
        :param fresh: check freshness token if True
        """
        if type_token == 'access':
            cookie_key = self._access_cookie_key
            cookie = self._request.cookies.get(cookie_key)
            csrf_cookie = self._request.headers.get(self._access_csrf_header_name)
        if type_token == 'refresh':
            cookie_key = self._refresh_cookie_key
            cookie = self._request.cookies.get(cookie_key)
            csrf_cookie = self._request.headers.get(self._refresh_csrf_header_name)

        if not cookie:
            raise MissingCookieError(status_code=401,message="Missing cookie {}".format(cookie_key))

        if self._cookie_csrf_protect and self._request.method in self._csrf_methods and not csrf_cookie:
            raise CSRFError(status_code=401,message="Missing CSRF Token")

        # set token from cookie and verify jwt
        self._token = cookie
        self._verify_jwt_in_request(self._token,type_token,'cookies',issuer,fresh)

        decoded_token = self.get_raw_jwt()

        if self._cookie_csrf_protect and self._request.method in self._csrf_methods and csrf_cookie:
            if 'csrf' not in decoded_token:
                raise JWTDecodeError(status_code=422,message="Missing claim: csrf")
            if not hmac.compare_digest(csrf_cookie,decoded_token['csrf']):
                raise CSRFError(status_code=401,message="CSRF double submit tokens do not match")

    def _verify_jwt_optional_in_request(self,token: str, issuer: Optional[str] = None) -> None:
        """
        Optionally check if this request has a valid access token

        :param token: The encoded JWT
        :param issuer: expected issuer in the JWT
        """
        if token:
            self._verifying_token(token,issuer)

        if token and self._get_type_token() != 'access':
            raise AccessTokenRequired(status_code=422,message="Only access tokens are allowed")

    def _verify_jwt_in_request(
        self,
        token: str,
        type_token: str,
        token_from: str,
        issuer: Optional[str] = None,
        fresh: Optional[bool] = False
    ) -> None:
        """
        Ensure that the requester has a valid token. this also check the freshness of the access token

        :param token: The encoded JWT
        :param type_token: indicate token is access or refresh token
        :param token_from: indicate token from headers or cookies
        :param issuer: expected issuer in the JWT
        :param fresh: check freshness token if True
        """
        if token:
            self._verifying_token(token,issuer)

        if not token and token_from == 'headers':
            raise MissingHeaderError(status_code=401,message="Missing {} Header".format(self._header_name))

        if self._get_type_token() != type_token:
            msg = "Only {} tokens are allowed".format(type_token)
            if type_token == 'access':
                raise AccessTokenRequired(status_code=422,message=msg)
            if type_token == 'refresh':
                raise RefreshTokenRequired(status_code=422,message=msg)

        if fresh and not self._get_fresh_token():
            raise FreshTokenRequired(status_code=401,message="Fresh token required")

    def _verifying_token(self,encoded_token: str, issuer: Optional[str] = None) -> None:
        """
        Verified token and check if token is revoked

        :param encoded_token: token hash
        :param issuer: expected issuer in the JWT
        """
        raw_token = self._verified_token(encoded_token,issuer)
        if raw_token['type'] in self._denylist_token_checks:
            self._check_token_is_revoked(raw_token)

    def _verified_token(self,encoded_token: str, issuer: Optional[str] = None) -> Dict[str,Union[str,int,bool]]:
        """
        Verified token and catch all error from jwt package and return decode token

        :param encoded_token: token hash
        :param issuer: expected issuer in the JWT

        :return: raw data from the hash token in the form of a dictionary
        """
        algorithms = self._decode_algorithms or [self._algorithm]

        try:
            unverified_headers = self.get_unverified_jwt_headers(encoded_token)
        except Exception as err:
            raise InvalidHeaderError(status_code=422,message=str(err))

        try:
            secret_key = self._get_secret_key(unverified_headers['alg'],"decode")
        except Exception:
            raise

        try:
            return jwt.decode(
                encoded_token,
                secret_key,
                issuer=issuer,
                audience=self._decode_audience,
                leeway=self._decode_leeway,
                algorithms=algorithms
            )
        except Exception as err:
            raise JWTDecodeError(status_code=422,message=str(err))

    def _get_type_token(self) -> str:
        return self.get_raw_jwt()['type']

    def _get_fresh_token(self) -> bool:
        return self.get_raw_jwt()['fresh']

    def jwt_required(self) -> None:
        """
        Only access token can access this function
        """
        if len(self._token_location) == 2:
            if self._token and self.jwt_in_headers:
                self._verify_jwt_in_request(self._token,'access','headers',self._decode_issuer)
            if not self._token and self.jwt_in_cookies:
                self._verify_and_get_jwt_in_cookies('access',self._decode_issuer)
        else:
            if self.jwt_in_headers:
                self._verify_jwt_in_request(self._token,'access','headers',self._decode_issuer)
            if self.jwt_in_cookies:
                self._verify_and_get_jwt_in_cookies('access',self._decode_issuer)

    def jwt_optional(self) -> None:
        """
        If an access token in present in the request you can get data from get_raw_jwt() or get_jwt_subject(),
        If no access token is present in the request, this endpoint will still be called, but
        get_raw_jwt() or get_jwt_subject() will return None
        """
        if len(self._token_location) == 2:
            if self._token and self.jwt_in_headers:
                self._verify_jwt_optional_in_request(self._token,self._decode_issuer)
            if not self._token and self.jwt_in_cookies:
                self._verify_and_get_jwt_optional_in_cookies(self._decode_issuer)
        else:
            if self.jwt_in_headers:
                self._verify_jwt_optional_in_request(self._token,self._decode_issuer)
            if self.jwt_in_cookies:
                self._verify_and_get_jwt_optional_in_cookies(self._decode_issuer)

    def jwt_refresh_token_required(self) -> None:
        """
        This function will ensure that the requester has a valid refresh token
        """
        if len(self._token_location) == 2:
            if self._token and self.jwt_in_headers:
                self._verify_jwt_in_request(self._token,'refresh','headers')
            if not self._token and self.jwt_in_cookies:
                self._verify_and_get_jwt_in_cookies('refresh')
        else:
            if self.jwt_in_headers:
                self._verify_jwt_in_request(self._token,'refresh','headers')
            if self.jwt_in_cookies:
                self._verify_and_get_jwt_in_cookies('refresh')

    def fresh_jwt_required(self) -> None:
        """
        This function will ensure that the requester has a valid access token and fresh token
        """
        if len(self._token_location) == 2:
            if self._token and self.jwt_in_headers:
                self._verify_jwt_in_request(self._token,'access','headers',self._decode_issuer,True)
            if not self._token and self.jwt_in_cookies:
                self._verify_and_get_jwt_in_cookies('access',self._decode_issuer,True)
        else:
            if self.jwt_in_headers:
                self._verify_jwt_in_request(self._token,'access','headers',self._decode_issuer,True)
            if self.jwt_in_cookies:
                self._verify_and_get_jwt_in_cookies('access',self._decode_issuer,True)

    def get_raw_jwt(self) -> Optional[Dict[str,Union[str,int,bool]]]:
        """
        this will return the python dictionary which has all of the claims of the JWT that is accessing the endpoint.
        If no JWT is currently present, return None instead

        :return: claims of JWT
        """
        if self._token:
            return self._verified_token(self._token)
        return None

    def get_jti(self,encoded_token: str) -> str:
        """
        Returns the JTI (unique identifier) of an encoded JWT

        :return: string of JTI
        """
        return self._verified_token(encoded_token)['jti']

    def get_jwt_subject(self) -> Optional[Union[str,int]]:
        """
        this will return the subject of the JWT that is accessing this endpoint.
        If no JWT is present, `None` is returned instead.

        :return: sub of JWT
        """
        if self._token:
            return self._verified_token(self._token)['sub']
        return None

    def get_unverified_jwt_headers(self,encoded_token: Optional[str] = None) -> dict:
        """
        Returns the Headers of an encoded JWT without verifying the actual signature of JWT

        :param encoded_token: The encoded JWT to get the Header from
        :return: JWT header parameters as a dictionary
        """
        encoded_token = encoded_token or self._token

        return jwt.get_unverified_header(encoded_token)

import jwt, re, uuid, hmac
from jwt.algorithms import requires_cryptography, has_crypto
from datetime import datetime, timezone, timedelta
from typing import Optional, Dict, Union, Sequence
from types import GeneratorType
from fastapi import Request, Response
from fastapi_jwt_auth.auth_config import AuthConfig
from fastapi_jwt_auth.exceptions import (
    InvalidHeaderError,
    CSRFError,
    JWTDecodeError,
    RevokedTokenError,
    MissingHeaderError,
    MissingCookieError,
    AccessTokenRequired,
    RefreshTokenRequired,
    FreshTokenRequired
)

class AuthJWT(AuthConfig):
    def __init__(self,req: Request = None, res: Response = None):
        """
        Get jwt header from incoming request or get
        request and response object if jwt in the cookie

        :param req: all incoming request
        :param res: response from endpoint
        """
        if res and self.jwt_in_cookies:
            self._response = res

        if req:
            # get request object when cookies in token location
            if self.jwt_in_cookies:
                self._request = req
            # get jwt in headers when headers in token location
            if self.jwt_in_headers:
                auth = req.headers.get(self._header_name.lower())
                if auth: self._get_jwt_from_headers(auth)

    def _get_jwt_from_headers(self,auth: str) -> "AuthJWT":
        """
        Get token from the headers

        :param auth: value from HeaderName
        """
        header_name, header_type = self._header_name, self._header_type

        parts = auth.split()

        # Make sure the header is in a valid format that we are expecting, ie
        if not header_type:
            # <HeaderName>: <JWT>
            if len(parts) != 1:
                msg = "Bad {} header. Expected value '<JWT>'".format(header_name)
                raise InvalidHeaderError(status_code=422,message=msg)
            self._token = parts[0]
        else:
            # <HeaderName>: <HeaderType> <JWT>
            if not re.match(r"{}\s".format(header_type),auth) or len(parts) != 2:
                msg = "Bad {} header. Expected value '{} <JWT>'".format(header_name,header_type)
                raise InvalidHeaderError(status_code=422,message=msg)
            self._token = parts[1]

    def _get_jwt_identifier(self) -> str:
        return str(uuid.uuid4())

    def _get_int_from_datetime(self,value: datetime) -> int:
        """
        :param value: datetime with or without timezone, if don't contains timezone
                      it will managed as it is UTC
        :return: Seconds since the Epoch
        """
        if not isinstance(value, datetime):  # pragma: no cover
            raise TypeError('a datetime is required')
        return int(value.timestamp())

    def _get_secret_key(self, algorithm: str, process: str) -> str:
        """
        Get key with a different algorithm

        :param algorithm: algorithm for decode and encode token
        :param process: for indicating get key for encode or decode token

        :return: plain text or RSA depends on algorithm
        """
        symmetric_algorithms, asymmetric_algorithms = {"HS256","HS384","HS512"}, requires_cryptography

        if algorithm not in symmetric_algorithms and algorithm not in asymmetric_algorithms:
            raise ValueError("Algorithm {} could not be found".format(algorithm))

        if algorithm in symmetric_algorithms:
            if not self._secret_key:
                raise RuntimeError(
                    "authjwt_secret_key must be set when using symmetric algorithm {}".format(algorithm)
                )

            return self._secret_key

        if algorithm in asymmetric_algorithms and not has_crypto:
            raise RuntimeError(
                "Missing dependencies for using asymmetric algorithms. run 'pip install fastapi-jwt-auth[asymmetric]'"
            )

        if process == "encode":
            if not self._private_key:
                raise RuntimeError(
                    "authjwt_private_key must be set when using asymmetric algorithm {}".format(algorithm)
                )

            return self._private_key

        if process == "decode":
            if not self._public_key:
                raise RuntimeError(
                    "authjwt_public_key must be set when using asymmetric algorithm {}".format(algorithm)
                )

            return self._public_key

    def _create_token(
        self,
        subject: Union[str,int],
        type_token: str,
        exp_time: Optional[int],
        fresh: Optional[bool] = False,
        algorithm: Optional[str] = None,
        headers: Optional[Dict] = None,
        issuer: Optional[str] = None,
        audience: Optional[Union[str,Sequence[str]]] = None,
        user_claims: Optional[Dict] = {}
    ) -> str:
        """
        Create token for access_token and refresh_token (utf-8)

        :param subject: Identifier for who this token is for example id or username from database.
        :param type_token: indicate token is access_token or refresh_token
        :param exp_time: Set the duration of the JWT
        :param fresh: Optional when token is access_token this param required
        :param algorithm: algorithm allowed to encode the token
        :param headers: valid dict for specifying additional headers in JWT header section
        :param issuer: expected issuer in the JWT
        :param audience: expected audience in the JWT
        :param user_claims: Custom claims to include in this token. This data must be dictionary

        :return: Encoded token
        """
        # Validation type data
        if not isinstance(subject, (str,int)):
            raise TypeError("subject must be a string or integer")
        if not isinstance(fresh, (bool)):
            raise TypeError("fresh must be a boolean")
        if audience and not isinstance(audience, (str, list, tuple, set, frozenset, GeneratorType)):
            raise TypeError("audience must be a string or sequence")
        if algorithm and not isinstance(algorithm, str):
            raise TypeError("algorithm must be a string")
        if user_claims and not isinstance(user_claims, dict):
            raise TypeError("user_claims must be a dictionary")

        # Data section
        reserved_claims = {
            "sub": subject,
            "iat": self._get_int_from_datetime(datetime.now(timezone.utc)),
            "nbf": self._get_int_from_datetime(datetime.now(timezone.utc)),
            "jti": self._get_jwt_identifier()
        }

        custom_claims = {"type": type_token}

        # for access_token only fresh needed
        if type_token == 'access':
            custom_claims['fresh'] = fresh
        # if cookie in token location and csrf protection enabled
        if self.jwt_in_cookies and self._cookie_csrf_protect:
            custom_claims['csrf'] = self._get_jwt_identifier()

        if exp_time:
            reserved_claims['exp'] = exp_time
        if issuer:
            reserved_claims['iss'] = issuer
        if audience:
            reserved_claims['aud'] = audience

        algorithm = algorithm or self._algorithm

        try:
            secret_key = self._get_secret_key(algorithm,"encode")
        except Exception:
            raise

        return jwt.encode(
            {**reserved_claims, **custom_claims, **user_claims},
            secret_key,
            algorithm=algorithm,
            headers=headers
        ).decode('utf-8')

    def _has_token_in_denylist_callback(self) -> bool:
        """
        Return True if token denylist callback set
        """
        return self._token_in_denylist_callback is not None

    def _check_token_is_revoked(self, raw_token: Dict[str,Union[str,int,bool]]) -> None:
        """
        Ensure that AUTHJWT_DENYLIST_ENABLED is true and callback regulated, and then
        call function denylist callback with passing decode JWT, if true
        raise exception Token has been revoked
        """
        if not self._denylist_enabled:
            return

        if not self._has_token_in_denylist_callback():
            raise RuntimeError("A token_in_denylist_callback must be provided via "
                "the '@AuthJWT.token_in_denylist_loader' if "
                "authjwt_denylist_enabled is 'True'")

        if self._token_in_denylist_callback.__func__(raw_token):
            raise RevokedTokenError(status_code=401,message="Token has been revoked")

    def _get_expired_time(
        self,
        type_token: str,
        expires_time: Optional[Union[timedelta,int,bool]] = None
    ) -> Union[None,int]:
        """
        Dynamic token expired if expires_time is False exp claim not created

        :param type_token: indicate token is access_token or refresh_token
        :param expires_time: duration expired jwt

        :return: duration exp claim jwt
        """
        if expires_time and not isinstance(expires_time, (timedelta,int,bool)):
            raise TypeError("expires_time must be between timedelta, int, bool")

        if expires_time is not False:
            if type_token == 'access':
                expires_time = expires_time or self._access_token_expires
            if type_token == 'refresh':
                expires_time = expires_time or self._refresh_token_expires

        if expires_time is not False:
            if isinstance(expires_time, bool):
                if type_token == 'access':
                    expires_time = self._access_token_expires
                if type_token == 'refresh':
                    expires_time = self._refresh_token_expires
            if isinstance(expires_time, timedelta):
                expires_time = int(expires_time.total_seconds())

            return self._get_int_from_datetime(datetime.now(timezone.utc)) + expires_time
        else:
            return None

    def create_access_token(
        self,
        subject: Union[str,int],
        fresh: Optional[bool] = False,
        algorithm: Optional[str] = None,
        headers: Optional[Dict] = None,
        expires_time: Optional[Union[timedelta,int,bool]] = None,
        audience: Optional[Union[str,Sequence[str]]] = None,
        user_claims: Optional[Dict] = {}
    ) -> str:
        """
        Create a access token with 15 minutes for expired time (default),
        info for param and return check to function create token

        :return: hash token
        """
        return self._create_token(
            subject=subject,
            type_token="access",
            exp_time=self._get_expired_time("access",expires_time),
            fresh=fresh,
            algorithm=algorithm,
            headers=headers,
            audience=audience,
            user_claims=user_claims,
            issuer=self._encode_issuer
        )

    def create_refresh_token(
        self,
        subject: Union[str,int],
        algorithm: Optional[str] = None,
        headers: Optional[Dict] = None,
        expires_time: Optional[Union[timedelta,int,bool]] = None,
        audience: Optional[Union[str,Sequence[str]]] = None,
        user_claims: Optional[Dict] = {}
    ) -> str:
        """
        Create a refresh token with 30 days for expired time (default),
        info for param and return check to function create token

        :return: hash token
        """
        return self._create_token(
            subject=subject,
            type_token="refresh",
            exp_time=self._get_expired_time("refresh",expires_time),
            algorithm=algorithm,
            headers=headers,
            audience=audience,
            user_claims=user_claims
        )

    def _get_csrf_token(self,encoded_token: str) -> str:
        """
        Returns the CSRF double submit token from an encoded JWT.

        :param encoded_token: The encoded JWT
        :return: The CSRF double submit token
        """
        return self._verified_token(encoded_token)['csrf']

    def set_access_cookies(self,encoded_access_token: str, max_age: Optional[int] = None) -> None:
        """
        Configures the response to set access token in a cookie.
        this will also set the CSRF double submit values in a separate cookie

        :param encoded_access_token: The encoded access token to set in the cookies
        :param max_age: The max age of the cookie value should be the number of seconds (integer)
        """
        if not self.jwt_in_cookies:
            raise RuntimeWarning(
                "set_access_cookies() called without 'authjwt_token_location' configured to use cookies"
            )

        if max_age and not isinstance(max_age,int):
            raise TypeError("max_age must be a integer")

        # Set the access JWT in the cookie
        self._response.set_cookie(
            self._access_cookie_key,
            encoded_access_token,
            max_age=max_age or self._cookie_max_age,
            path=self._access_cookie_path,
            domain=self._cookie_domain,
            secure=self._cookie_secure,
            httponly=True,
            samesite=self._cookie_samesite
        )

        # If enabled, set the csrf double submit access cookie
        if self._cookie_csrf_protect:
            self._response.set_cookie(
                self._access_csrf_cookie_key,
                self._get_csrf_token(encoded_access_token),
                max_age=max_age or self._cookie_max_age,
                path=self._access_csrf_cookie_path,
                domain=self._cookie_domain,
                secure=self._cookie_secure,
                httponly=False,
                samesite=self._cookie_samesite
            )

    def set_refresh_cookies(self, encoded_refresh_token: str, max_age: Optional[int] = None) -> None:
        """
        Configures the response to set refresh token in a cookie.
        this will also set the CSRF double submit values in a separate cookie

        :param encoded_refresh_token: The encoded refresh token to set in the cookies
        :param max_age: The max age of the cookie value should be the number of seconds (integer)
        """
        if not self.jwt_in_cookies:
            raise RuntimeWarning(
                "set_refresh_cookies() called without 'authjwt_token_location' configured to use cookies"
            )

        if max_age and not isinstance(max_age,int):
            raise TypeError("max_age must be a integer")

        # Set the refresh JWT in the cookie
        self._response.set_cookie(
            self._refresh_cookie_key,
            encoded_refresh_token,
            max_age=max_age or self._cookie_max_age,
            path=self._refresh_cookie_path,
            domain=self._cookie_domain,
            secure=self._cookie_secure,
            httponly=True,
            samesite=self._cookie_samesite
        )

        # If enabled, set the csrf double submit refresh cookie
        if self._cookie_csrf_protect:
            self._response.set_cookie(
                self._refresh_csrf_cookie_key,
                self._get_csrf_token(encoded_refresh_token),
                max_age=max_age or self._cookie_max_age,
                path=self._refresh_csrf_cookie_path,
                domain=self._cookie_domain,
                secure=self._cookie_secure,
                httponly=False,
                samesite=self._cookie_samesite
            )

    def unset_jwt_cookies(self) -> None:
        """
        Unset (delete) all jwt stored in a cookie
        """
        self.unset_access_cookies()
        self.unset_refresh_cookies()

    def unset_access_cookies(self) -> None:
        """
        Remove access token and access CSRF double submit from the response cookies
        """
        if not self.jwt_in_cookies:
            raise RuntimeWarning(
                "unset_access_cookies() called without 'authjwt_token_location' configured to use cookies"
            )

        self._response.delete_cookie(
            self._access_cookie_key,
            path=self._access_cookie_path,
            domain=self._cookie_domain
        )

        if self._cookie_csrf_protect:
            self._response.delete_cookie(
                self._access_csrf_cookie_key,
                path=self._access_csrf_cookie_path,
                domain=self._cookie_domain
            )

    def unset_refresh_cookies(self) -> None:
        """
        Remove refresh token and refresh CSRF double submit from the response cookies
        """
        if not self.jwt_in_cookies:
            raise RuntimeWarning(
                "unset_refresh_cookies() called without 'authjwt_token_location' configured to use cookies"
            )

        self._response.delete_cookie(
            self._refresh_cookie_key,
            path=self._refresh_cookie_path,
            domain=self._cookie_domain
        )

        if self._cookie_csrf_protect:
            self._response.delete_cookie(
                self._refresh_csrf_cookie_key,
                path=self._refresh_csrf_cookie_path,
                domain=self._cookie_domain
            )

    def _verify_and_get_jwt_optional_in_cookies(self,issuer: Optional[str] = None) -> "AuthJWT":
        """
        Optionally check if cookies have a valid access token. if an access token present in
        cookies property _token will set. raises exception error when an access token is invalid
        and doesn't match with CSRF token double submit

        :param issuer: expected issuer in the JWT
        """
        cookie_key = self._access_cookie_key
        cookie = self._request.cookies.get(cookie_key)
        csrf_cookie = self._request.headers.get(self._access_csrf_header_name)

        if (
            cookie and
            self._cookie_csrf_protect and
            self._request.method in self._csrf_methods and
            not csrf_cookie
        ):
            raise CSRFError(status_code=401,message="Missing CSRF Token")

        # set token from cookie and verify jwt
        self._token = cookie
        self._verify_jwt_optional_in_request(self._token,issuer)

        decoded_token = self.get_raw_jwt()

        if (
            self._cookie_csrf_protect and
            self._request.method in self._csrf_methods and
            csrf_cookie and
            decoded_token
        ):
            if 'csrf' not in decoded_token:
                raise JWTDecodeError(status_code=422,message="Missing claim: csrf")
            if not hmac.compare_digest(csrf_cookie,decoded_token['csrf']):
                raise CSRFError(status_code=401,message="CSRF double submit tokens do not match")

    def _verify_and_get_jwt_in_cookies(
        self,
        type_token: str,
        issuer: Optional[str] = None,
        fresh: Optional[bool] = False
    ) -> "AuthJWT":
        """
        Check if cookies have a valid access or refresh token. if an token present in
        cookies property _token will set. raises exception error when an access or refresh token
        is invalid and doesn't match with CSRF token double submit

        :param type_token: indicate token is access or refresh token
        :param issuer: expected issuer in the JWT
        :param fresh: check freshness token if True
        """
        if type_token == 'access':
            cookie_key = self._access_cookie_key
            cookie = self._request.cookies.get(cookie_key)
            csrf_cookie = self._request.headers.get(self._access_csrf_header_name)
        if type_token == 'refresh':
            cookie_key = self._refresh_cookie_key
            cookie = self._request.cookies.get(cookie_key)
            csrf_cookie = self._request.headers.get(self._refresh_csrf_header_name)

        if not cookie:
            raise MissingCookieError(status_code=401,message="Missing cookie {}".format(cookie_key))

        if self._cookie_csrf_protect and self._request.method in self._csrf_methods and not csrf_cookie:
            raise CSRFError(status_code=401,message="Missing CSRF Token")

        # set token from cookie and verify jwt
        self._token = cookie
        self._verify_jwt_in_request(self._token,type_token,'cookies',issuer,fresh)

        decoded_token = self.get_raw_jwt()

        if self._cookie_csrf_protect and self._request.method in self._csrf_methods and csrf_cookie:
            if 'csrf' not in decoded_token:
                raise JWTDecodeError(status_code=422,message="Missing claim: csrf")
            if not hmac.compare_digest(csrf_cookie,decoded_token['csrf']):
                raise CSRFError(status_code=401,message="CSRF double submit tokens do not match")

    def _verify_jwt_optional_in_request(self,token: str, issuer: Optional[str] = None) -> None:
        """
        Optionally check if this request has a valid access token

        :param token: The encoded JWT
        :param issuer: expected issuer in the JWT
        """
        if token:
            self._verifying_token(token,issuer)

        if token and self._get_type_token() != 'access':
            raise AccessTokenRequired(status_code=422,message="Only access tokens are allowed")

    def _verify_jwt_in_request(
        self,
        token: str,
        type_token: str,
        token_from: str,
        issuer: Optional[str] = None,
        fresh: Optional[bool] = False
    ) -> None:
        """
        Ensure that the requester has a valid token. this also check the freshness of the access token

        :param token: The encoded JWT
        :param type_token: indicate token is access or refresh token
        :param token_from: indicate token from headers or cookies
        :param issuer: expected issuer in the JWT
        :param fresh: check freshness token if True
        """
        if token:
            self._verifying_token(token,issuer)

        if not token and token_from == 'headers':
            raise MissingHeaderError(status_code=401,message="Missing {} Header".format(self._header_name))

        if self._get_type_token() != type_token:
            msg = "Only {} tokens are allowed".format(type_token)
            if type_token == 'access':
                raise AccessTokenRequired(status_code=422,message=msg)
            if type_token == 'refresh':
                raise RefreshTokenRequired(status_code=422,message=msg)

        if fresh and not self._get_fresh_token():
            raise FreshTokenRequired(status_code=401,message="Fresh token required")

    def _verifying_token(self,encoded_token: str, issuer: Optional[str] = None) -> None:
        """
        Verified token and check if token is revoked

        :param encoded_token: token hash
        :param issuer: expected issuer in the JWT
        """
        raw_token = self._verified_token(encoded_token,issuer)
        if raw_token['type'] in self._denylist_token_checks:
            self._check_token_is_revoked(raw_token)

    def _verified_token(self,encoded_token: str, issuer: Optional[str] = None) -> Dict[str,Union[str,int,bool]]:
        """
        Verified token and catch all error from jwt package and return decode token

        :param encoded_token: token hash
        :param issuer: expected issuer in the JWT

        :return: raw data from the hash token in the form of a dictionary
        """
        algorithms = self._decode_algorithms or [self._algorithm]

        try:
            unverified_headers = self.get_unverified_jwt_headers(encoded_token)
        except Exception as err:
            raise InvalidHeaderError(status_code=422,message=str(err))

        try:
            secret_key = self._get_secret_key(unverified_headers['alg'],"decode")
        except Exception:
            raise

        try:
            return jwt.decode(
                encoded_token,
                secret_key,
                issuer=issuer,
                audience=self._decode_audience,
                leeway=self._decode_leeway,
                algorithms=algorithms
            )
        except Exception as err:
            raise JWTDecodeError(status_code=422,message=str(err))

    def _get_type_token(self) -> str:
        return self.get_raw_jwt()['type']

    def _get_fresh_token(self) -> bool:
        return self.get_raw_jwt()['fresh']

    def jwt_required(self) -> None:
        """
        Only access token can access this function
        """
        if len(self._token_location) == 2:
            if self._token and self.jwt_in_headers:
                self._verify_jwt_in_request(self._token,'access','headers',self._decode_issuer)
            if not self._token and self.jwt_in_cookies:
                self._verify_and_get_jwt_in_cookies('access',self._decode_issuer)
        else:
            if self.jwt_in_headers:
                self._verify_jwt_in_request(self._token,'access','headers',self._decode_issuer)
            if self.jwt_in_cookies:
                self._verify_and_get_jwt_in_cookies('access',self._decode_issuer)

    def jwt_optional(self) -> None:
        """
        If an access token in present in the request you can get data from get_raw_jwt() or get_jwt_subject(),
        If no access token is present in the request, this endpoint will still be called, but
        get_raw_jwt() or get_jwt_subject() will return None
        """
        if len(self._token_location) == 2:
            if self._token and self.jwt_in_headers:
                self._verify_jwt_optional_in_request(self._token,self._decode_issuer)
            if not self._token and self.jwt_in_cookies:
                self._verify_and_get_jwt_optional_in_cookies(self._decode_issuer)
        else:
            if self.jwt_in_headers:
                self._verify_jwt_optional_in_request(self._token,self._decode_issuer)
            if self.jwt_in_cookies:
                self._verify_and_get_jwt_optional_in_cookies(self._decode_issuer)

    def jwt_refresh_token_required(self) -> None:
        """
        This function will ensure that the requester has a valid refresh token
        """
        if len(self._token_location) == 2:
            if self._token and self.jwt_in_headers:
                self._verify_jwt_in_request(self._token,'refresh','headers')
            if not self._token and self.jwt_in_cookies:
                self._verify_and_get_jwt_in_cookies('refresh')
        else:
            if self.jwt_in_headers:
                self._verify_jwt_in_request(self._token,'refresh','headers')
            if self.jwt_in_cookies:
                self._verify_and_get_jwt_in_cookies('refresh')

    def fresh_jwt_required(self) -> None:
        """
        This function will ensure that the requester has a valid access token and fresh token
        """
        if len(self._token_location) == 2:
            if self._token and self.jwt_in_headers:
                self._verify_jwt_in_request(self._token,'access','headers',self._decode_issuer,True)
            if not self._token and self.jwt_in_cookies:
                self._verify_and_get_jwt_in_cookies('access',self._decode_issuer,True)
        else:
            if self.jwt_in_headers:
                self._verify_jwt_in_request(self._token,'access','headers',self._decode_issuer,True)
            if self.jwt_in_cookies:
                self._verify_and_get_jwt_in_cookies('access',self._decode_issuer,True)

    def get_raw_jwt(self) -> Optional[Dict[str,Union[str,int,bool]]]:
        """
        this will return the python dictionary which has all of the claims of the JWT that is accessing the endpoint.
        If no JWT is currently present, return None instead

        :return: claims of JWT
        """
        if self._token:
            return self._verified_token(self._token)
        return None

    def get_jti(self,encoded_token: str) -> str:
        """
        Returns the JTI (unique identifier) of an encoded JWT

        :return: string of JTI
        """
        return self._verified_token(encoded_token)['jti']

    def get_jwt_subject(self) -> Optional[Union[str,int]]:
        """
        this will return the subject of the JWT that is accessing this endpoint.
        If no JWT is present, `None` is returned instead.

        :return: sub of JWT
        """
        if self._token:
            return self._verified_token(self._token)['sub']
        return None

    def get_unverified_jwt_headers(self,encoded_token: Optional[str] = None) -> dict:
        """
        Returns the Headers of an encoded JWT without verifying the actual signature of JWT

        :param encoded_token: The encoded JWT to get the Header from
        :return: JWT header parameters as a dictionary
        """
        encoded_token = encoded_token or self._token

        return jwt.get_unverified_header(encoded_token)
Get jwt header from incoming request or get         request and response object if jwt in the cookie          :param req: all incoming request         :param res: response from endpoint # get request object when cookies in token location # get jwt in headers when headers in token location Get token from the headers          :param auth: value from HeaderName # Make sure the header is in a valid format that we are expecting, ie # <HeaderName>: <JWT> # <HeaderName>: <HeaderType> <JWT> :param value: datetime with or without timezone, if don't contains timezone                       it will managed as it is UTC         :return: Seconds since the Epoch # pragma: no cover Get key with a different algorithm          :param algorithm: algorithm for decode and encode token         :param process: for indicating get key for encode or decode token          :return: plain text or RSA depends on algorithm Create token for access_token and refresh_token (utf-8)          :param subject: Identifier for who this token is for example id or username from database.         :param type_token: indicate token is access_token or refresh_token         :param exp_time: Set the duration of the JWT         :param fresh: Optional when token is access_token this param required         :param algorithm: algorithm allowed to encode the token         :param headers: valid dict for specifying additional headers in JWT header section         :param issuer: expected issuer in the JWT         :param audience: expected audience in the JWT         :param user_claims: Custom claims to include in this token. This data must be dictionary          :return: Encoded token # Validation type data # Data section # for access_token only fresh needed # if cookie in token location and csrf protection enabled Return True if token denylist callback set Ensure that AUTHJWT_DENYLIST_ENABLED is true and callback regulated, and then         call function denylist callback with passing decode JWT, if true         raise exception Token has been revoked Dynamic token expired if expires_time is False exp claim not created          :param type_token: indicate token is access_token or refresh_token         :param expires_time: duration expired jwt          :return: duration exp claim jwt Create a access token with 15 minutes for expired time (default),         info for param and return check to function create token          :return: hash token Create a refresh token with 30 days for expired time (default),         info for param and return check to function create token          :return: hash token Returns the CSRF double submit token from an encoded JWT.          :param encoded_token: The encoded JWT         :return: The CSRF double submit token Configures the response to set access token in a cookie.         this will also set the CSRF double submit values in a separate cookie          :param encoded_access_token: The encoded access token to set in the cookies         :param max_age: The max age of the cookie value should be the number of seconds (integer) # Set the access JWT in the cookie # If enabled, set the csrf double submit access cookie Configures the response to set refresh token in a cookie.         this will also set the CSRF double submit values in a separate cookie          :param encoded_refresh_token: The encoded refresh token to set in the cookies         :param max_age: The max age of the cookie value should be the number of seconds (integer) # Set the refresh JWT in the cookie # If enabled, set the csrf double submit refresh cookie Unset (delete) all jwt stored in a cookie Remove access token and access CSRF double submit from the response cookies Remove refresh token and refresh CSRF double submit from the response cookies Optionally check if cookies have a valid access token. if an access token present in         cookies property _token will set. raises exception error when an access token is invalid         and doesn't match with CSRF token double submit          :param issuer: expected issuer in the JWT # set token from cookie and verify jwt Check if cookies have a valid access or refresh token. if an token present in         cookies property _token will set. raises exception error when an access or refresh token         is invalid and doesn't match with CSRF token double submit          :param type_token: indicate token is access or refresh token         :param issuer: expected issuer in the JWT         :param fresh: check freshness token if True # set token from cookie and verify jwt Optionally check if this request has a valid access token          :param token: The encoded JWT         :param issuer: expected issuer in the JWT Ensure that the requester has a valid token. this also check the freshness of the access token          :param token: The encoded JWT         :param type_token: indicate token is access or refresh token         :param token_from: indicate token from headers or cookies         :param issuer: expected issuer in the JWT         :param fresh: check freshness token if True Verified token and check if token is revoked          :param encoded_token: token hash         :param issuer: expected issuer in the JWT Verified token and catch all error from jwt package and return decode token          :param encoded_token: token hash         :param issuer: expected issuer in the JWT          :return: raw data from the hash token in the form of a dictionary Only access token can access this function If an access token in present in the request you can get data from get_raw_jwt() or get_jwt_subject(),         If no access token is present in the request, this endpoint will still be called, but         get_raw_jwt() or get_jwt_subject() will return None This function will ensure that the requester has a valid refresh token This function will ensure that the requester has a valid access token and fresh token this will return the python dictionary which has all of the claims of the JWT that is accessing the endpoint.         If no JWT is currently present, return None instead          :return: claims of JWT Returns the JTI (unique identifier) of an encoded JWT          :return: string of JTI this will return the subject of the JWT that is accessing this endpoint.         If no JWT is present, `None` is returned instead.          :return: sub of JWT Returns the Headers of an encoded JWT without verifying the actual signature of JWT          :param encoded_token: The encoded JWT to get the Header from         :return: JWT header parameters as a dictionary