litefs.auth package

认证授权模块

提供完整的用户认证和授权解决方案： - JWT Token 认证 - OAuth2 社交登录 - 用户注册/登录 - 密码加密 - 权限控制 - Token 刷新 - 认证中间件

使用示例:

from litefs.auth import Auth, login_required, role_required

auth = Auth(app, secret_key='your-secret-key')

@route('/protected')
@login_required
def protected(request):
    return {'user': request.user.username}

@route('/admin')
@role_required('admin')
def admin_only(request):
    return {'message': 'Admin access'}

OAuth2 示例:

from litefs.auth import OAuth2

oauth = OAuth2(app)
oauth.register(
    name='github',
    client_id='your-client-id',
    client_secret='your-client-secret',
    redirect_uri='http://localhost:8080/auth/github/callback'
)

@route('/login/github')
def github_login(request):
    return oauth.authorize_redirect(request, 'github')

@route('/auth/github/callback')
def github_callback(request):
    user_info = oauth.authorize_user(request, 'github')
    return {'user': user_info.to_dict()}

class litefs.auth.Auth(app=None, secret_key: str | None = None, algorithm: str = 'HS256', access_token_expires: int = 3600, refresh_token_expires: int = 604800, token_location: str = 'header', token_name: str = 'Authorization', token_prefix: str = 'Bearer')[源代码]

基类：object

认证授权管理器

提供完整的用户认证和授权功能。

authenticate(username: str, password: str)[源代码]

验证用户

参数:

username -- 用户名
password -- 密码

返回:

User 实例或 None

create_user(username: str, password: str, **kwargs)[源代码]

创建用户

参数:

username -- 用户名
password -- 密码
**kwargs -- 其他属性（email, roles 等）

返回:

User 实例

init_app(app)[源代码]

初始化应用

参数:: app -- Litefs 应用实例

login_handler(request)[源代码]: 登录处理

logout_handler(request)[源代码]: 登出处理

refresh_handler(request)[源代码]: Token 刷新处理

register_handler(request)[源代码]: 注册处理

class litefs.auth.AuthMiddleware(app=None, jwt_manager=None, token_location: str = 'header', token_name: str = 'Authorization', token_prefix: str = 'Bearer', exempt_paths: list | None = None)[源代码]

基类：object

认证中间件

process_request(request)[源代码]

处理请求

参数:: request -- 请求对象
返回:: None 或响应

process_response(request, response)[源代码]

处理响应

参数:

request -- 请求对象
response -- 响应数据

返回:

响应数据

class litefs.auth.BaseUserStore[源代码]

基类：ABC

用户存储后端抽象基类

所有用户存储后端必须实现以下方法： - get_by_id(user_id): 根据 ID 获取用户 - get_by_username(username): 根据用户名获取用户 - create(username, password_hash, **kwargs): 创建用户 - delete(user_id): 删除用户 - list_all(): 获取所有用户

abstract create(username: str, password_hash: str, email: str | None = None, roles: List[Role] | None = None) → User[源代码]: 创建用户

abstract delete(user_id: int) → bool[源代码]: 删除用户

abstract get_by_id(user_id: int) → User | None[源代码]: 根据 ID 获取用户

abstract get_by_username(username: str) → User | None[源代码]: 根据用户名获取用户

abstract list_all() → List[User][源代码]: 获取所有用户

class litefs.auth.GitHubProvider(client_id: str, client_secret: str, redirect_uri: str = '', scope: str | None = None, **kwargs)[源代码]

基类：OAuth2Provider

GitHub OAuth2 提供商

access_token_url: str = 'https://github.com/login/oauth/access_token'

authorize_url: str = 'https://github.com/login/oauth/authorize'

get_access_token(code: str, redirect_uri: str | None = None) → Dict[str, Any][源代码]

使用授权码获取 Access Token

参数:

code -- 授权码
redirect_uri -- 回调地址

返回:

Token 信息

get_user_info(access_token: str) → OAuth2UserInfo[源代码]

获取用户信息

参数:: access_token -- Access Token
返回:: 用户信息

name: str = 'github'

scope: str = 'user:email'

userinfo_url: str = 'https://api.github.com/user'

class litefs.auth.GoogleProvider(client_id: str, client_secret: str, redirect_uri: str = '', scope: str | None = None, **kwargs)[源代码]

基类：OAuth2Provider

Google OAuth2 提供商

access_token_url: str = 'https://oauth2.googleapis.com/token'

authorize_url: str = 'https://accounts.google.com/o/oauth2/v2/auth'

get_access_token(code: str, redirect_uri: str | None = None) → Dict[str, Any][源代码]

使用授权码获取 Access Token

参数:

code -- 授权码
redirect_uri -- 回调地址

返回:

Token 信息

get_user_info(access_token: str) → OAuth2UserInfo[源代码]

获取用户信息

参数:: access_token -- Access Token
返回:: 用户信息

name: str = 'google'

scope: str = 'openid email profile'

userinfo_url: str = 'https://www.googleapis.com/oauth2/v3/userinfo'

class litefs.auth.JWTManager(secret_key: str, algorithm: str = 'HS256', access_token_expires: int = 3600, refresh_token_expires: int = 604800, blacklist_backend=None)[源代码]

基类：object

JWT Token 管理器

cleanup_blacklist()[源代码]

清理黑名单中的过期条目

仅对内存后端有效，缓存后端依赖自身 TTL 机制。

create_access_token(payload: Dict[str, Any]) → str[源代码]

创建 Access Token

参数:: payload -- 载荷数据
返回:: Access Token

create_refresh_token(payload: Dict[str, Any]) → str[源代码]

创建 Refresh Token

参数:: payload -- 载荷数据
返回:: Refresh Token

create_token(payload: Dict[str, Any], expires_in: int | None = None) → str[源代码]

创建 JWT Token

参数:

payload -- 载荷数据
expires_in -- 过期时间（秒）

返回:

JWT Token 字符串

decode_token(token: str) → Dict[str, Any] | None[源代码]

解码 JWT Token

参数:: token -- JWT Token 字符串
返回:: 载荷数据或 None

is_token_revoked(token: str) → bool[源代码]

检查 Token 是否已撤销

参数:: token -- JWT Token
返回:: 是否已撤销

revoke_token(token: str)[源代码]

撤销 Token（加入黑名单）

自动计算黑名单条目的 TTL，与 token 自身的 exp 对齐， token 过期后黑名单条目自动清理。

参数:: token -- JWT Token

class litefs.auth.MemoryUserStore[源代码]

基类：BaseUserStore

内存用户存储（默认后端）

适合开发和测试环境。重启后数据丢失，多进程/多实例不共享。生产环境应使用数据库后端。

create(username: str, password_hash: str, email: str | None = None, roles: List[Role] | None = None) → User[源代码]: 创建用户

delete(user_id: int) → bool[源代码]: 删除用户

get_by_id(user_id: int) → User | None[源代码]: 根据 ID 获取用户

get_by_username(username: str) → User | None[源代码]: 根据用户名获取用户

list_all() → List[User][源代码]: 获取所有用户

class litefs.auth.OAuth2(app=None, state_ttl: int = 600, user_mapper: Callable[[OAuth2UserInfo], Any] | None = None)[源代码]

基类：object

OAuth2 认证管理器

提供完整的 OAuth2 认证流程管理。

使用示例:

oauth = OAuth2(app)

oauth.register(
    name='github',
    client_id='your-client-id',
    client_secret='your-client-secret',
    redirect_uri='http://localhost:8080/auth/github/callback'
)

@route('/login/github')
def github_login(request):
    return oauth.authorize_redirect(request, 'github')

@route('/auth/github/callback')
def github_callback(request):
    user_info = oauth.authorize_user(request, 'github')
    if user_info:
        return {'user': user_info.to_dict()}
    return {'error': '认证失败'}

authorize_redirect(request, provider_name: str, callback_url: str | None = None, **kwargs)[源代码]

生成授权重定向响应

参数:

request -- 请求对象
provider_name -- 提供商名称
callback_url -- 回调地址
**kwargs -- 额外参数

返回:

重定向响应

authorize_user(request, provider_name: str | None = None, **kwargs) → OAuth2UserInfo | None[源代码]

完成授权并获取用户信息

参数:

request -- 请求对象
provider_name -- 提供商名称（可选，从状态中获取）
**kwargs -- 额外参数

返回:

用户信息或 None

async authorize_user_async(request, provider_name: str | None = None, **kwargs) → OAuth2UserInfo | None[源代码]

异步完成授权并获取用户信息

使用 aiohttp 进行异步 HTTP 请求，不会阻塞请求处理线程。如果未安装 aiohttp，回退到同步方式。

参数:

request -- 请求对象
provider_name -- 提供商名称（可选，从状态中获取）
**kwargs -- 额外参数

返回:

用户信息或 None

get_authorize_url(provider_name: str, callback_url: str | None = None, state: str | None = None) → str[源代码]

获取授权 URL

参数:

provider_name -- 提供商名称
callback_url -- 回调地址
state -- 状态参数

返回:

授权 URL

get_provider(name: str) → OAuth2Provider | None[源代码]

获取提供商

参数:: name -- 提供商名称
返回:: OAuth2Provider 实例

init_app(app)[源代码]

初始化应用

参数:: app -- Litefs 应用实例

register(name: str, client_id: str, client_secret: str, redirect_uri: str = '', scope: str | None = None, provider_class: type | None = None, **kwargs)[源代码]

注册 OAuth2 提供商

参数:

name -- 提供商名称
client_id -- 客户端 ID
client_secret -- 客户端密钥
redirect_uri -- 回调地址
scope -- 权限范围
provider_class -- 提供商类
**kwargs -- 额外参数

set_user_mapper(mapper: Callable[[OAuth2UserInfo], Any])[源代码]

设置用户映射函数

参数:: mapper -- 映射函数，接收 OAuth2UserInfo 返回用户对象

class litefs.auth.OAuth2Provider(client_id: str, client_secret: str, redirect_uri: str = '', scope: str | None = None, **kwargs)[源代码]

基类：ABC

OAuth2 提供商基类

access_token_url: str = ''

authorize_url: str = ''

generate_pkce_challenge(verifier: str, method: str = 'S256') → str[源代码]

生成 PKCE code_challenge

参数:

verifier -- code_verifier
method -- 挑战方法 ('S256' 或 'plain')

返回:

code_challenge

generate_pkce_verifier() → str[源代码]: 生成 PKCE code_verifier

generate_state() → str[源代码]: 生成随机状态参数

abstract get_access_token(code: str, redirect_uri: str | None = None) → Dict[str, Any][源代码]

使用授权码获取 Access Token

参数:

code -- 授权码
redirect_uri -- 回调地址

返回:

Token 信息

get_authorize_url(state: str, redirect_uri: str | None = None, **kwargs) → str[源代码]

生成授权 URL

参数:

state -- 状态参数，用于防止 CSRF 攻击
redirect_uri -- 回调地址
**kwargs -- 额外参数

返回:

授权 URL

abstract get_user_info(access_token: str) → OAuth2UserInfo[源代码]

获取用户信息

参数:: access_token -- Access Token
返回:: 用户信息

name: str = 'base'

scope: str = ''

userinfo_url: str = ''

class litefs.auth.OAuth2State(ttl: int = 600)[源代码]

基类：object

OAuth2 状态管理

cleanup()[源代码]: 清理过期状态

create(provider: str, redirect_uri: str | None = None, **kwargs) → str[源代码]

创建状态

参数:

provider -- 提供商名称
redirect_uri -- 回调地址
**kwargs -- 额外数据

返回:

状态字符串

verify(state: str) → Dict[str, Any] | None[源代码]

验证状态

参数:: state -- 状态字符串
返回:: 状态数据或 None

class litefs.auth.OAuth2UserInfo(provider: str, provider_user_id: str, username: str, email: str | None = None, name: str | None = None, avatar_url: str | None = None, raw_data: Dict[str, Any] | None = None)[源代码]

基类：object

OAuth2 用户信息

avatar_url: str | None = None

email: str | None = None

name: str | None = None

provider: str

provider_user_id: str

raw_data: Dict[str, Any] | None = None

to_dict() → Dict[str, Any][源代码]

username: str

class litefs.auth.Permission(name: str, description: str = '')[源代码]

基类：object

权限

classmethod create(name: str, description: str = '') → Permission[源代码]: 创建权限

description: str = ''

classmethod get_by_name(name: str) → Permission | None[源代码]: 根据名称获取权限

name: str

class litefs.auth.Role(name: str, description: str = '', permissions: ~typing.List[~litefs.auth.models.Permission] = <factory>)[源代码]

基类：object

角色

add_permission(permission: Permission)[源代码]: 添加权限

classmethod create(name: str, description: str = '', permissions: List[Permission] | None = None) → Role[源代码]: 创建角色

description: str = ''

classmethod get_by_name(name: str) → Role | None[源代码]: 根据名称获取角色

has_permission(permission_name: str) → bool[源代码]: 检查是否有权限

name: str

permissions: List[Permission]

remove_permission(permission: Permission)[源代码]: 移除权限

class litefs.auth.User(id: int, username: str, password_hash: str, email: str | None = None, is_active: bool = True, roles: ~typing.List[~litefs.auth.models.Role] = <factory>)[源代码]

基类：object

用户

add_role(role: Role)[源代码]: 添加角色

email: str | None = None

has_permission(permission_name: str) → bool[源代码]: 检查是否有权限

has_role(role_name: str) → bool[源代码]: 检查是否有角色

id: int

is_active: bool = True

password_hash: str

remove_role(role: Role)[源代码]: 移除角色

roles: List[Role]

username: str

class litefs.auth.WeChatProvider(client_id: str, client_secret: str, redirect_uri: str = '', scope: str | None = None, **kwargs)[源代码]

基类：OAuth2Provider

微信 OAuth2 提供商

access_token_url: str = 'https://api.weixin.qq.com/sns/oauth2/access_token'

authorize_url: str = 'https://open.weixin.qq.com/connect/qrconnect'

get_access_token(code: str, redirect_uri: str | None = None) → Dict[str, Any][源代码]

使用授权码获取 Access Token

参数:

code -- 授权码
redirect_uri -- 回调地址

返回:

Token 信息

get_authorize_url(state: str, redirect_uri: str | None = None, **kwargs) → str[源代码]

生成授权 URL

参数:

state -- 状态参数，用于防止 CSRF 攻击
redirect_uri -- 回调地址
**kwargs -- 额外参数

返回:

授权 URL

get_user_info(access_token: str, openid: str | None = None) → OAuth2UserInfo[源代码]

获取用户信息

参数:: access_token -- Access Token
返回:: 用户信息

name: str = 'wechat'

scope: str = 'snsapi_login'

userinfo_url: str = 'https://api.weixin.qq.com/sns/userinfo'

class litefs.auth.WeComProvider(client_id: str, client_secret: str, agent_id: str = '', redirect_uri: str = '', **kwargs)[源代码]

基类：OAuth2Provider

企业微信 OAuth2 提供商

access_token_url: str = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken'

authorize_url: str = 'https://open.work.weixin.qq.com/wwopen/sso/qrConnect'

get_access_token(code: str, redirect_uri: str | None = None) → Dict[str, Any][源代码]

使用授权码获取 Access Token

参数:

code -- 授权码
redirect_uri -- 回调地址

返回:

Token 信息

get_authorize_url(state: str, redirect_uri: str | None = None, **kwargs) → str[源代码]

生成授权 URL

参数:

state -- 状态参数，用于防止 CSRF 攻击
redirect_uri -- 回调地址
**kwargs -- 额外参数

返回:

授权 URL

get_user_info(access_token: str, code: str | None = None) → OAuth2UserInfo[源代码]

获取用户信息

参数:: access_token -- Access Token
返回:: 用户信息

name: str = 'wecom'

scope: str = 'snsapi_base'

userinfo_url: str = 'https://qyapi.weixin.qq.com/cgi-bin/user/getuserinfo'

litefs.auth.create_oauth2_blueprint(oauth: OAuth2, prefix: str = '/auth')[源代码]

创建 OAuth2 蓝图

参数:

oauth -- OAuth2 管理器
prefix -- URL 前缀

返回:

路由列表

litefs.auth.create_token(payload: Dict[str, Any], secret_key: str, expires_in: int = 3600, algorithm: str = 'HS256') → str[源代码]

创建 JWT Token（便捷函数）

参数:

payload -- 载荷数据
secret_key -- 密钥
expires_in -- 过期时间（秒）
algorithm -- 算法

返回:

JWT Token

litefs.auth.current_user(func: Callable) → Callable[源代码]

当前用户装饰器

将当前用户作为第一个参数传递给处理函数。

参数:: func -- 处理函数
返回:: 装饰后的函数

litefs.auth.decode_token(token: str, secret_key: str, algorithm: str = 'HS256') → Dict[str, Any] | None[源代码]

解码 JWT Token（便捷函数）

参数:

token -- JWT Token
secret_key -- 密钥
algorithm -- 算法

返回:

载荷数据或 None

litefs.auth.get_provider(name: str) → type | None[源代码]: 获取提供商类

litefs.auth.get_user_store() → BaseUserStore[源代码]: 获取全局用户存储实例

litefs.auth.hash_password(password: str, rounds: int = 12) → str[源代码]

哈希密码

参数:

password -- 明文密码
rounds -- 哈希轮数（仅 bcrypt）

返回:

哈希后的密码

litefs.auth.init_default_roles_and_permissions()[源代码]: 初始化默认角色和权限

litefs.auth.login_required(func: Callable) → Callable[源代码]

登录验证装饰器

参数:: func -- 处理函数
返回:: 装饰后的函数

litefs.auth.permission_required(*permissions: str) → Callable[源代码]

权限验证装饰器

参数:: *permissions -- 需要的权限列表
返回:: 装饰器函数

litefs.auth.role_required(*roles: str) → Callable[源代码]

角色验证装饰器

参数:: *roles -- 允许的角色列表
返回:: 装饰器函数

litefs.auth.set_user_store(store: BaseUserStore)[源代码]

设置全局用户存储实例

用于切换到数据库后端等持久化存储。

参数:: store -- BaseUserStore 实例

使用示例:

from litefs.auth.models import set_user_store, BaseUserStore

class DatabaseUserStore(BaseUserStore):
    def __init__(self, db_session):
        self.db = db_session
    # ... 实现抽象方法

set_user_store(DatabaseUserStore(db_session))

litefs.auth.verify_password(password: str, password_hash: str) → bool[源代码]

验证密码

参数:

password -- 明文密码
password_hash -- 哈希后的密码

返回:

是否匹配

Submodules

litefs.auth.jwt module

JWT Token 管理

提供 JWT Token 的生成、验证和刷新功能。

class litefs.auth.jwt.JWTManager(secret_key: str, algorithm: str = 'HS256', access_token_expires: int = 3600, refresh_token_expires: int = 604800, blacklist_backend=None)[源代码]

基类：object

JWT Token 管理器

cleanup_blacklist()[源代码]

清理黑名单中的过期条目

仅对内存后端有效，缓存后端依赖自身 TTL 机制。

create_access_token(payload: Dict[str, Any]) → str[源代码]

创建 Access Token

参数:: payload -- 载荷数据
返回:: Access Token

create_refresh_token(payload: Dict[str, Any]) → str[源代码]

创建 Refresh Token

参数:: payload -- 载荷数据
返回:: Refresh Token

create_token(payload: Dict[str, Any], expires_in: int | None = None) → str[源代码]

创建 JWT Token

参数:

payload -- 载荷数据
expires_in -- 过期时间（秒）

返回:

JWT Token 字符串

decode_token(token: str) → Dict[str, Any] | None[源代码]

解码 JWT Token

参数:: token -- JWT Token 字符串
返回:: 载荷数据或 None

is_token_revoked(token: str) → bool[源代码]

检查 Token 是否已撤销

参数:: token -- JWT Token
返回:: 是否已撤销

revoke_token(token: str)[源代码]

撤销 Token（加入黑名单）

自动计算黑名单条目的 TTL，与 token 自身的 exp 对齐， token 过期后黑名单条目自动清理。

参数:: token -- JWT Token

class litefs.auth.jwt.TokenBlacklist(backend=None)[源代码]

基类：object

Token 黑名单存储

支持多种后端：内存、缓存（Redis 等）。黑名单条目自动过期清理，不会无限增长。

add(token: str, expires_in: int = 3600)[源代码]

将 token 加入黑名单

参数:

token -- JWT Token
expires_in -- 黑名单条目存活时间（秒），应 >= token 自身的 exp

cleanup()[源代码]

清理内存后端中的过期条目

仅对内存后端有效，缓存后端依赖自身的 TTL 机制自动清理。

clear()[源代码]: 清空黑名单

is_blacklisted(token: str) → bool[源代码]

检查 token 是否在黑名单中

参数:: token -- JWT Token
返回:: 是否已撤销

litefs.auth.jwt.create_token(payload: Dict[str, Any], secret_key: str, expires_in: int = 3600, algorithm: str = 'HS256') → str[源代码]

创建 JWT Token（便捷函数）

参数:

payload -- 载荷数据
secret_key -- 密钥
expires_in -- 过期时间（秒）
algorithm -- 算法

返回:

JWT Token

litefs.auth.jwt.decode_token(token: str, secret_key: str, algorithm: str = 'HS256') → Dict[str, Any] | None[源代码]

解码 JWT Token（便捷函数）

参数:

token -- JWT Token
secret_key -- 密钥
algorithm -- 算法

返回:

载荷数据或 None

litefs.auth.password module

密码哈希和验证

提供安全的密码哈希和验证功能。

litefs.auth.password.check_password_breach(password: str, use_cache: bool = True, timeout: int = 5) → bool[源代码]

检查密码是否在已知泄露数据库中

使用 Have I Been Pwned (HIBP) API 的 k-anonymity 模型： 1. 计算密码的 SHA-1 哈希 2. 只发送哈希的前5个字符到 API 3. API 返回所有以这5个字符开头的哈希后缀和出现次数 4. 在本地检查完整的哈希是否在返回的列表中

这种方式既保护了密码隐私，又能有效检查泄露情况。

参数:

password -- 密码
use_cache -- 是否使用缓存（默认 True）
timeout -- API 请求超时时间（秒，默认 5）

返回:

True 表示已泄露，False 表示安全

抛出:

无异常抛出，网络错误时返回 False 并记录警告 --

示例

>>> is_breached = check_password_breach("password123")
>>> if is_breached:
...     print("密码已泄露，请更换")

async litefs.auth.password.check_password_breach_async(password: str, use_cache: bool = True, timeout: int = 5) → bool[源代码]

异步检查密码是否在已知泄露数据库中

使用 Have I Been Pwned (HIBP) API 的 k-anonymity 模型，异步版本不会阻塞请求处理线程。

参数:

password -- 密码
use_cache -- 是否使用缓存（默认 True）
timeout -- API 请求超时时间（秒，默认 5）

返回:

True 表示已泄露，False 表示安全

抛出:

无异常抛出，网络错误时返回 False 并记录警告 --

示例

>>> is_breached = await check_password_breach_async("password123")
>>> if is_breached:
...     print("密码已泄露，请更换")

litefs.auth.password.clear_breach_cache() → None[源代码]

清空密码泄露检查缓存

在需要强制刷新缓存数据时调用此函数。

litefs.auth.password.generate_password(length: int = 16) → str[源代码]

生成随机密码

参数:: length -- 密码长度
返回:: 随机密码

litefs.auth.password.get_breach_count(password: str, timeout: int = 5) → int[源代码]

获取密码在泄露数据库中的出现次数

参数:

password -- 密码
timeout -- API 请求超时时间（秒，默认 5）

返回:

泄露次数，0 表示未泄露，-1 表示查询失败

示例

>>> count = get_breach_count("password123")
>>> if count > 0:
...     print(f"密码已泄露 {count} 次")

async litefs.auth.password.get_breach_count_async(password: str, timeout: int = 5) → int[源代码]

异步获取密码在泄露数据库中的出现次数

参数:

password -- 密码
timeout -- API 请求超时时间（秒，默认 5）

返回:

泄露次数，0 表示未泄露，-1 表示查询失败

示例

>>> count = await get_breach_count_async("password123")
>>> if count > 0:
...     print(f"密码已泄露 {count} 次")

litefs.auth.password.hash_password(password: str, rounds: int = 12) → str[源代码]

哈希密码

参数:

password -- 明文密码
rounds -- 哈希轮数（仅 bcrypt）

返回:

哈希后的密码

litefs.auth.password.validate_password_strength(password: str) → Tuple[bool, list][源代码]

验证密码强度

参数:: password -- 密码
返回:: (是否通过, 错误消息列表)

litefs.auth.password.verify_password(password: str, password_hash: str) → bool[源代码]

验证密码

参数:

password -- 明文密码
password_hash -- 哈希后的密码

返回:

是否匹配

litefs.auth.middleware module

认证中间件

提供 JWT Token 验证和用户身份注入功能。

class litefs.auth.middleware.AuthMiddleware(app=None, jwt_manager=None, token_location: str = 'header', token_name: str = 'Authorization', token_prefix: str = 'Bearer', exempt_paths: list | None = None)[源代码]

基类：object

认证中间件

process_request(request)[源代码]

处理请求

参数:: request -- 请求对象
返回:: None 或响应

process_response(request, response)[源代码]

处理响应

参数:

request -- 请求对象
response -- 响应数据

返回:

响应数据

litefs.auth.middleware.login_required(func: Callable) → Callable[源代码]

登录验证装饰器

参数:: func -- 处理函数
返回:: 装饰后的函数

litefs.auth.middleware.role_required(*roles: str) → Callable[源代码]

角色验证装饰器

参数:: *roles -- 允许的角色列表
返回:: 装饰器函数

litefs.auth.decorators module

权限装饰器

提供权限验证和用户信息获取的装饰器。

litefs.auth.decorators.admin_required(func: Callable) → Callable[源代码]

管理员权限装饰器

参数:: func -- 处理函数
返回:: 装饰后的函数

litefs.auth.decorators.current_user(func: Callable) → Callable[源代码]

当前用户装饰器

将当前用户作为第一个参数传递给处理函数。

参数:: func -- 处理函数
返回:: 装饰后的函数

litefs.auth.decorators.owner_or_admin_required(get_owner_id: Callable) → Callable[源代码]

所有者或管理员权限装饰器

参数:: get_owner_id -- 获取所有者 ID 的函数
返回:: 装饰器函数

litefs.auth.decorators.permission_required(*permissions: str) → Callable[源代码]

权限验证装饰器

参数:: *permissions -- 需要的权限列表
返回:: 装饰器函数

litefs.auth.models module

用户模型

提供用户、角色和权限的基础模型，以及可替换的用户存储后端。

默认使用内存存储（MemoryUserStore），适合开发和测试。生产环境可实现 BaseUserStore 接口对接数据库。

class litefs.auth.models.BaseUserStore[源代码]

基类：ABC

用户存储后端抽象基类

所有用户存储后端必须实现以下方法： - get_by_id(user_id): 根据 ID 获取用户 - get_by_username(username): 根据用户名获取用户 - create(username, password_hash, **kwargs): 创建用户 - delete(user_id): 删除用户 - list_all(): 获取所有用户

abstract create(username: str, password_hash: str, email: str | None = None, roles: List[Role] | None = None) → User[源代码]: 创建用户

abstract delete(user_id: int) → bool[源代码]: 删除用户

abstract get_by_id(user_id: int) → User | None[源代码]: 根据 ID 获取用户

abstract get_by_username(username: str) → User | None[源代码]: 根据用户名获取用户

abstract list_all() → List[User][源代码]: 获取所有用户

class litefs.auth.models.MemoryUserStore[源代码]

基类：BaseUserStore

内存用户存储（默认后端）

适合开发和测试环境。重启后数据丢失，多进程/多实例不共享。生产环境应使用数据库后端。

create(username: str, password_hash: str, email: str | None = None, roles: List[Role] | None = None) → User[源代码]: 创建用户

delete(user_id: int) → bool[源代码]: 删除用户

get_by_id(user_id: int) → User | None[源代码]: 根据 ID 获取用户

get_by_username(username: str) → User | None[源代码]: 根据用户名获取用户

list_all() → List[User][源代码]: 获取所有用户

class litefs.auth.models.Permission(name: str, description: str = '')[源代码]

基类：object

权限

classmethod create(name: str, description: str = '') → Permission[源代码]: 创建权限

description: str = ''

classmethod get_by_name(name: str) → Permission | None[源代码]: 根据名称获取权限

name: str

class litefs.auth.models.Role(name: str, description: str = '', permissions: ~typing.List[~litefs.auth.models.Permission] = <factory>)[源代码]

基类：object

角色

add_permission(permission: Permission)[源代码]: 添加权限

classmethod create(name: str, description: str = '', permissions: List[Permission] | None = None) → Role[源代码]: 创建角色

description: str = ''

classmethod get_by_name(name: str) → Role | None[源代码]: 根据名称获取角色

has_permission(permission_name: str) → bool[源代码]: 检查是否有权限

name: str

permissions: List[Permission]

remove_permission(permission: Permission)[源代码]: 移除权限

class litefs.auth.models.User(id: int, username: str, password_hash: str, email: str | None = None, is_active: bool = True, roles: ~typing.List[~litefs.auth.models.Role] = <factory>)[源代码]

基类：object

用户

add_role(role: Role)[源代码]: 添加角色

email: str | None = None

has_permission(permission_name: str) → bool[源代码]: 检查是否有权限

has_role(role_name: str) → bool[源代码]: 检查是否有角色

id: int

is_active: bool = True

password_hash: str

remove_role(role: Role)[源代码]: 移除角色

roles: List[Role]

username: str

litefs.auth.models.get_user_store() → BaseUserStore[源代码]: 获取全局用户存储实例

litefs.auth.models.init_default_roles_and_permissions()[源代码]: 初始化默认角色和权限

litefs.auth.models.set_user_store(store: BaseUserStore)[源代码]

设置全局用户存储实例

用于切换到数据库后端等持久化存储。

参数:: store -- BaseUserStore 实例

使用示例:

from litefs.auth.models import set_user_store, BaseUserStore

class DatabaseUserStore(BaseUserStore):
    def __init__(self, db_session):
        self.db = db_session
    # ... 实现抽象方法

set_user_store(DatabaseUserStore(db_session))