HEX
Server: LiteSpeed
System: Linux atali.colombiahosting.com.co 5.14.0-570.12.1.el9_6.x86_64 #1 SMP PREEMPT_DYNAMIC Tue May 13 06:11:55 EDT 2025 x86_64
User: coopserp (1713)
PHP: 8.2.29
Disabled: dl,exec,passthru,proc_open,proc_close,shell_exec,memory_limit,system,popen,curl_multi_exec,show_source,symlink,link,leak,listen,diskfreespace,tmpfile,ignore_user_abord,highlight_file,source,show_source,fpaththru,virtual,posix_ctermid,posix_getcwd,posix_getegid,posix_geteuid,posix_getgid,posix_getgrgid,posix_getgrnam,posix_getgroups,posix_getlogin,posix_getpgid,posix_getpgrp,posix_getpid,posix,posix_getppid,posix_getpwnam,posix_getpwuid,posix_getrlimit,posix_getsid,posix_getuid,posix_isatty,posix_kill,posix_mkfifo,posix_setegid,posix_seteuid,posix_setgid,posix_setpgid,posix_setsid,posix_setid,posix_times,posix_ttyname,posix_uname,proc_get_status,proc_nice,proc_terminate
Upload Files
File: //opt/cloudlinux/venv/lib64/python3.11/site-packages/aiohttp_security/jwt_identity.py
"""Identity policy for storing info in the jwt token.

"""

from .abc import AbstractIdentityPolicy

try:
    import jwt
except ImportError:  # pragma: no cover
    jwt = None


AUTH_HEADER_NAME = 'Authorization'
AUTH_SCHEME = 'Bearer '


class JWTIdentityPolicy(AbstractIdentityPolicy):
    def __init__(self, secret, algorithm='HS256'):
        if jwt is None:
            raise RuntimeError('Please install `PyJWT`')
        self.secret = secret
        self.algorithm = algorithm

    async def identify(self, request):
        header_identity = request.headers.get(AUTH_HEADER_NAME)

        if header_identity is None:
            return

        if not header_identity.startswith(AUTH_SCHEME):
            raise ValueError('Invalid authorization scheme. ' +
                             'Should be `Bearer <token>`')

        token = header_identity.split(' ')[1].strip()

        identity = jwt.decode(token,
                              self.secret,
                              algorithms=[self.algorithm])
        return identity

    async def remember(self, *args, **kwargs):  # pragma: no cover
        pass

    async def forget(self, request, response):  # pragma: no cover
        pass