ApoJWT

创建该apojwt包的目的是为 In10t 的 Apogee 微服务提供 JWT 支持。这些服务需要在所有端点之间变化的权限层次结构。因此，这个包旨在提供可以附加路由声明的装饰器，以确保在请求标头中发送具有适当权限的有效 JWT。该软件包旨在与 Flask 或 FastAPI 等 API 框架一起使用。

ApoJWT 类

ApoJWT 类具有以下构造函数：

(self, secret: str, exp_period: int=900, iss: str="", asynch: bool=False, server_audience: str="default", algorithm: str="HS256", token_finder=None, permission_formatter=None, exception_handler=None)
"""
Keyword Arguments (those with asterisks are functions):

JWT Validation
    secret: Secret string used to encode and decode the access JWT
    exp_period: Length of time in seconds access tokens should be valid for. Default 900 (15 minutes) 
    iss: Issuer string used for additional security. Default ""
    server_audience: Audience name of the server hosting the HTTP framework. 
        Audience names are typically base address URLs. Ex: https://example.com
    algorithm: The algorithm to use when encoding/decoding. Default HS256
    admin_permission: Optional permission allowing full access to JWTs carrying this permission. USE CAREFULLY
    * token_finder: Function used to retrieve access JWT from http Authorization header. Default None
        Expected Function Structure: (*args, **kwargs) -> str
        NOTE: args and kwargs are the same arguments given to the http request handler 

Framework Configuration
    asynch: Tells ApoJWT to use async decorators instead of the normal. Default False
        (FastAPI needs this True)
    * exception_handler: HTTP error handling function given an HTTP code and message as arguments
        Expected Function Structure: (code: int, exception_type: str, msg: str) -> None

Request Data Utility
    * permission_formatter: String formatting function that is given permission_name as an argument
        Can be used to format request data in the permission name
        Expected Function Structure: (permission: str, *args, **kwargs) -> str
"""

ApoJWT 中的高阶函数

令牌查找器

token_finder 函数必须传递给更高阶的构造函数，以使修饰的令牌验证成功。该函数必须返回 JWT 字符串，该字符串通常可以在带有“授权”键的 HTTP 请求标头中找到。JWT 的标准做法是使用“Bearer”作为前缀。由这个函数来删除这个子字符串。

预期功能结构：(*args, **kwargs) -> str

注意：args和kwargs是给 HTTP 请求处理程序的相同参数

例子

request.headers["Authorization"]
>>> 'Bearer <token>'
request.headers["Authorization"].replace("Bearer ", "")
>>> '<token>'

"""Token Finder: used to locate and return the JWT"""
# FastAPI
token_finder = lambda **kwargs: str(kwargs["Authorization"]).replace("Bearer ", "")
ajwt = ("secret", iss="issuer", asynch=True, token_finder=token_finder)
## NOTE: asynch is True for FastAPI

# Flask
token_finder = lambda: request.headers["Authorization"].replace("Bearer ", "")
ajwt = ("secret", iss="issuer", token_finder=token_finder)
## NOTE: asynch defaults to False for Flask

异常处理程序

异常处理程序是可选的，但允许使用正在使用的 HTTP 框架提供的 HTTP 错误响应正确处理修饰验证。

预期功能结构：(code: int, exception_type: str, msg: str, *args, **kwargs) -> None

例子

"""Exception Handler"""
# FastAPI
def exception_handler(code: int, msg: str):
    raise HTTPException(status_code=code, detail=msg)
ajwt = ("secret", iss="issuer", asynch=True, token_finder=..., exception_handler=exception_handler)

# Flask
def exception_handler(code: int, msg: str):
    abort(code, msg)
ajwt = ("secret", iss="issuer", token_finder=..., exception_handler=exception_handler)

权限格式化程序

permission_formatter 是完全可选的，但可用于向只能在请求正文中找到的权限添加附加信息。

预期功能结构：(permission: str, *args, **kwargs) -> str

示例：附加请求中找到的 resource_id

"""Permission Formatter: used to apply additional formatting to permission"""
# FastAPI
def fastapi_permission_formatter(permission_name, *args, **kwargs):
    if "resource_id" in kwargs.keys():
        return f"{permission_name}:{kwargs['resource_id']}"
ajwt = ("secret", iss="issuer", asynch=True, token_finder=..., permission_formatter=fastapi_permission_formatter)

# Flask
def flask_permission_formatter(permission_name):
    if "resource_id" in request.args.keys():
        return f"{permission_name}:{request.args['resource_id']}"
    elif "resource_id" in request.json.keys():
        return f"{permission_name}:{request.json['resource_id']}"
ajwt = ("secret", iss="issuer", asynch=False, token_finder=..., permission_formatter=flask_permission_formatter)

装饰器

装饰器是初始化后 ApoJWT 包的主要用例。它们允许使用一行简单的代码来保护任何端点。

ajwt = ApoJWT(secret, iss, token_finder=lambda: ..., ...)


@ajwt.token_required
"""Validates JWT

Returns 'token_data' and 'token_sub' as kwargs to HTTP handler
"""


@ajwt.permission_required(permission_name: str)
"""Validates JWT and ensures permission_name is among the token permissions

permission_name: a permission string

Returns 'token_data' and 'token_sub' as kwargs to HTTP handler
"""

两个装饰器都返回token_data和token_sub作为被装饰的 HTTP 处理程序的关键字参数。使用这些参数，存储在 JWT 和 JWT 主题中的附加数据都可以访问。

功能

ajwt = ApoJWT(secret, iss, asynch=..., token_finder=...)

ajwt.create_token(self, sub: str="", permissions: list[str]=[], aud: list[str]=[], data: dict=dict(), refresh_data: dict=dict()):
        """Encodes and returns an access JWT and optionally a refresh JWT

        sub: Subject of the JWT (typically some reference to the user of JWT)
        permissions: List of permissions to assign to token
        aud: List of audiences token should be accepted by
        data: Any additional information that is needed
        refresh_data: If refresh is configured, this additional data is stored with the refresh token

        JWT will contain the following claims:
            - exp: Expiration Time
            - nbf: Not Before Time
            - iss: Issuer
            - aud: Audience
            - iat: Issued At
        """

ajwt.token_data():
"""Retrieves the additional data stored in the JWT payload"""

刷新令牌

ApoJWT 1.5.0 引入了刷新令牌功能。强烈建议使用此功能为应用程序提供额外的安全层。要了解刷新令牌及其好处，请查看这篇 Auth0 文章了解更多信息。ApoJWT 中的刷新功能通过以下函数激活：

ajwt.config_refresh(refresh_secret: str, refresh_exp_period: int=86400, refresh_finder=None):
        """Configures ApoJWT for use with refresh tokens

        refresh_secret: Secret string used to encode and decode the refresh JWT

        refresh_exp_period: Number of seconds for the refresh token to be valid. Default 86400 (1 day)

        refresh_finder: Function used to retrieve the refresh JWT from an http-only cookie. Default None
        """

该函数与必须返回刷新令牌的函数refresh_finder类似。token_finder主要区别在于refresh_finder，在大多数情况下，应该在仅 http 的安全 cookie 而不是 HTTP 授权标头中找到刷新令牌。refresh_finder如果只需要创建刷新令牌而不由 ApoJWT 使用，则也是可选的。

预期refresh_finder功能结构：(*args, **kwargs) -> str

一旦调用并初始化此函数，ApoJWT 就可以处理刷新令牌。

刷新功能

该create_token函数现在将返回一个包含访问令牌和刷新令牌的元组

access, refresh = ajwt.create_token(...)

通常，此刷新令牌可以存储在仅 HTTP 的 cookie 中。

从那里，@ajwt.refresh装饰器可以放置在应该发生刷新的任何端点上。装饰器将新的访问令牌作为关键字参数access_token传递给它正在装饰的函数。

# Fast Api
@app.get("/some/endpoint")
@ajwt.refresh
def refresh(access_token):
    return access_token

使用示例

构建 ApoJWT

# FastAPI
def fastapi_permission_formatter(permission_name, *args, **kwargs):
    if "resource_id" in kwargs.keys():
        return f"{permission_name}:{kwargs['resource_id']}"
        
def fastapi_exception_handler(code, msg):
    raise HTTPException(status_code=code, detail=msg)

fastapi_token_finder = lambda **kwargs: str(kwargs["authorization"]).replace("Bearer ", "")
ajwt = (
    "secret", 
    iss="issuer", 
    asynch=True
    token_finder=fastapi_token_finder
    permission_formatter=fastapi_permission_formatter
    exception_handler=fastapi_exception_handler
)
## NOTE: asynch must be True for FastAPI

# Flask
def flask_permission_formatter(permission_name):
    if "resource_id" in request.args.keys():
        return f"{permission_name}:{request.args['resource_id']}"
    elif "resource_id" in request.json.keys():
        return f"{permission_name}:{request.json['resource_id']}"

def flask_exception_handler(code, msg):
    abort(code, msg)
    
flask_token_finder = lambda: request.headers["authorization"].replace("Bearer ", "")
ajwt = (
    "secret", 
    iss="issuer",
    token_finder=flask_token_finder
    permission_formatter=flask_permission_formatter
    exception_handler=flask_exception_handler
)
## NOTE: asynch defaults to False

使用装饰器验证 JWT

# fast api
@app.get("/some/endpoint")
@ajwt.permission_required("some:permission:name"):
...

# flask
@app.route("/some/endpoint", methods=["GET"])
@ajwt.permission_required("some:permission:name"):
...

刷新配置

# fast api
def refresh_finder(refresh_token: Union[str, None] = Cookie(default=None)):
    return refresh_token
ajwt.config_refresh("refresh_secret", refresh_finder=refresh_finder)

# flask
ajwt.config_refresh("refresh_secret", refresh_finder=lambda: request.cookies.get('refresh_token'))
(refresh_secret: str, refresh_exp_period: int=86400, refresh_finder=None)

创建新的 JWT

"""Permissions will be assigned to the new token"""

sub = "user_id_1"
permissions = ["some:permission:name", ...]
data = dict(...=...)
aud = [...]

# NOTE: all arguments are optional
# If refresh is not configured
token = ajwt.create_token(sub=sub, permissions=permissions, aud=aud, data=data)

# If refresh is configured
refresh_data = dict(...=...)
access, refresh = ajwt.create_token(
    sub=sub, 
    permissions=permissions, 
    aud=aud, 
    data=data, 
    refresh_data=refresh_data
)

从 JWT 获取令牌数据和主题

@app....
@ajwt.token_required
def route(token_data: dict, token_subject: str):
    print(token_subject)
    return token_data