Metadata-Version: 2.5
Name: liteauth
Version: 0.2.0
Summary: Python 异步权限认证框架，受 Sa-Token 启发 — 多账号体系、JWT、OAuth2、SSO
Project-URL: Homepage, https://gitee.com/YueXia_1/liteauth
Project-URL: Repository, https://gitee.com/YueXia_1/liteauth
Project-URL: Documentation, https://gitee.com/YueXia_1/liteauth
Author-email: YueJian <yuexia@example.com>
License: Apache-2.0
License-File: LICENSE
Keywords: auth,authentication,fastapi,jwt,oauth2,sa-token,sso
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: FastAPI
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Internet :: WWW/HTTP :: Session
Classifier: Topic :: Security
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: pydantic>=2.13.4
Requires-Dist: pyjwt>=2.8.0
Provides-Extra: all
Requires-Dist: cryptography>=41.0.0; extra == 'all'
Requires-Dist: fastapi>=0.100.0; extra == 'all'
Requires-Dist: hiredis>=3.4.0; extra == 'all'
Requires-Dist: orjson>=3.11.9; extra == 'all'
Requires-Dist: redis<6.0,>=5.0; extra == 'all'
Requires-Dist: uvicorn>=0.24.0; extra == 'all'
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.100.0; extra == 'fastapi'
Requires-Dist: uvicorn>=0.24.0; extra == 'fastapi'
Provides-Extra: jwt-rsa
Requires-Dist: cryptography>=41.0.0; extra == 'jwt-rsa'
Provides-Extra: orjson
Requires-Dist: orjson>=3.11.9; extra == 'orjson'
Provides-Extra: redis
Requires-Dist: hiredis>=3.4.0; extra == 'redis'
Requires-Dist: redis<6.0,>=5.0; extra == 'redis'
Description-Content-Type: text/markdown

# liteauth

> Python 异步权限认证框架，受 [Sa-Token](https://github.com/dromara/sa-token) 启发。

支持多账号体系、JWT（Simple / Mixin / Stateless）、OAuth2 服务端、SSO、Session 管理、角色/权限校验、踢人/顶号/封禁等能力。

## 设计原则

- **Core 无框架依赖** — 核心认证逻辑与 FastAPI / Flask / Django 解耦
- **Async first** — 全异步设计，原生适配 FastAPI / Starlette
- **多账号体系** — 一个系统多套用户表（`user` / `admin` / `merchant`），各自独立认证
- **可插拔存储** — Memory / Redis 自由切换，也可实现自定义 Store
- **JWT 可选** — 三种模式（Simple / Mixin / Stateless）按需选用

## 快速开始

```python
from liteauth import LiteAuthManager, LiteAuthConfig
from liteauth.store import MemoryStore

sa = LiteAuthManager(config=LiteAuthConfig(), store=MemoryStore())
user_auth = sa.create_logic("user")

pair = await user_auth.login("10001")
token = pair.access_token                     # 统一 TokenPair（基础模式 refresh_token 为 None）
login_id = await user_auth.get_login_id_by_token(token)  # "10001"
```

### 多账号体系与独立配置

一个 `LiteAuthManager` 可承载多套账号体系，`login_type` 是唯一标识（同 `login_type`
重复调用返回同一实例）。每套体系可以有自己的配置与存储，不传的项回落管理器的全局值：

```python
sa = LiteAuthManager(config=LiteAuthConfig(key_prefix="liteauth"), store=MemoryStore())

# user：全局 config + 全局 store
user_auth = sa.create_logic("user")

# admin：独立配置（独立 key 前缀 + 独立 token 风格），store 仍用全局
admin_auth = sa.create_logic(
    "admin",
    config=LiteAuthConfig(key_prefix="liteauth:admin", token_style="random-64"),
)

# merchant：独立存储（例如另一个 Redis 实例/库）
merchant_auth = sa.create_logic("merchant", store=other_store)
```

- `config` 里的 `key_prefix` / `token_style` 会派生到该体系自己的 KeyBuilder 与
  token 生成器，登录态、Session、封禁等全部落在该前缀下，与其它体系互不干扰。
- JWT 模式同理：`sa.create_logic("admin", logic_class=JwtAuthLogic, config=...)`
  可以让两套体系使用不同的 `jwt.secret_key`。
- 需要完全自建实例（含自定义 KeyBuilder）时用 `sa.register_logic(my_auth)`。
- 同名 `login_type` 已存在时，显式传入不一致的 `config` / `store` / `logic_class`
  会抛 `ValueError`（不会静默忽略你的参数）。

## 模式功能对比

liteauth 提供四种认证模式：基础模式（UUID + 全状态）与三种 JWT 模式（Simple / Mixin / Stateless），按需选用。

| 能力 | 基础模式 | JWT Simple | JWT Mixin | JWT Stateless |
| --- | --- | --- | --- | --- |
| 对应类 | `AuthLogic` | `JwtAuthLogic` | `JwtMixinAuthLogic` | `JwtStatelessAuthLogic` |
| Token 格式 | UUID | JWT | JWT | JWT |
| 存储依赖 | Store | Store | Store | ❌ 无（完全无状态） |
| login_id 来源 | Store（token→id 映射） | Store | JWT payload | JWT payload |
| 双 Token（access + refresh） | ❌ | ❌ | ✅ | ✅ |
| 服务端作废 token | ✅ | ✅ | ⚠️ 仅当前 token，refresh 无法作废 | ❌（logout 仅清客户端缓存） |
| Session（Account / Token） | ✅ | ✅ | ✅ | ❌ |
| 踢人 / 顶号 | ✅ | ✅ | ❌ | ❌ |
| 全部下线（logout_by_login_id） | ✅ | ✅ | ❌ | ❌ |
| 封禁（disable / disable_service） | ✅ | ✅ | ✅ | ❌ |
| 角色 / 权限校验 | ✅ | ✅ | ✅ | ✅ |
| 二级认证（safe） | ✅ Store 记录 | ✅ Store 记录 | ✅ Store 记录 | ✅ fresh claim |
| JWT payload 解析 | ❌ | ✅ | ✅ | ✅ |

**如何选择**：
- **基础模式** — 不需要 JWT 时的全功能兜底
- **JWT Simple** — 想要 JWT 格式（便于跨端解析 / 调试），同时保留 Redis 全状态能力
- **JWT Mixin** — 减少 Redis 查询（login_id 直接从 payload 读取），接受放弃踢人 / 顶号 / 全部下线
- **JWT Stateless** — 完全无状态，适合分布式 / 微服务；放弃 Session、封禁、服务端作废，内置 fresh-claim 二级认证

## JWT Stateless 双 Token（access + refresh）

Stateless 模式完全无状态（不依赖任何 Store），登录返回 `TokenPair`：

```python
from liteauth.config import JwtConfig, LiteAuthConfig
from liteauth.plugin.jwt import JwtStatelessAuthLogic

auth = JwtStatelessAuthLogic(
    "user",
    LiteAuthConfig(jwt=JwtConfig(
        secret_key="your-secret",
        jwt_access_token_timeout=3600,    # access 短效（秒）
        jwt_refresh_token_timeout=604800, # refresh 有限（秒）
        enable_refresh_token=True,        # False 时只签发 access
    )),
)

# 登录 → TokenPair(access_token, refresh_token, token_type, expires_in, ...)
pair = await auth.login("10001")
login_id = await auth.get_login_id_by_token(pair.access_token)  # "10001"

# 刷新 → 新 token 对（纯无状态，不依赖存储）
new_pair = await auth.refresh(pair.refresh_token)
```

安全模型说明（纯无状态取舍）：
- access 泄漏危害窗口 = `jwt_access_token_timeout`（短）
- refresh 泄漏危害窗口 = `jwt_refresh_token_timeout`（有限）
- 不依赖存储 ⇒ 无法作废旧 token / 检测重放，`logout` 仅清客户端缓存
- `get_login_id_by_token` 会拒绝 refresh token（校验 `token_use` claim）

## FastAPI 集成

```python
from fastapi import FastAPI, Depends
from liteauth.integration.fastapi.dependency import require_login, require_role

app = FastAPI()
sa.init_app(app)

@app.get("/me", dependencies=[Depends(require_login("user"))])
async def me():
    return {"msg": "已登录"}

@app.get("/admin", dependencies=[Depends(require_role("admin", login_type="user"))])
async def admin():
    return {"msg": "管理员"}
```

### 路由中间件方式

```python
from liteauth.core.router import GuardRule
from liteauth.integration.fastapi.middleware import GuardRuleMiddleware

router = GuardRule()
router.match("/api/admin/**").check(lambda ctx: admin_auth.check_login(ctx))
app.add_middleware(GuardRuleMiddleware, guard_rule=router)
```

## 插件：OAuth2 / SSO 端点

插件核心（`liteauth/plugin/**`）零 Web 框架依赖，只提供协议方法；HTTP 端点由
`liteauth/integration/**` 以 `APIRouter` 工厂交付，路由归属由你决定。

### OAuth2 服务端

```python
from liteauth.integration.fastapi import oauth2_router
from liteauth.plugin.oauth2 import (
    MemoryOAuth2DataLoader, OAuth2Client, OAuth2Config, OAuth2LoginResult,
    OAuth2Server, OAuth2Strategy,
)

server = OAuth2Server(
    config=OAuth2Config(),                      # 也可写进 LiteAuthConfig(oauth2=...)
    store=sa.store,
    data_loader=MemoryOAuth2DataLoader([        # 生产可自行实现 OAuth2DataLoader 查库
        OAuth2Client(
            client_id="c1",
            client_secret="s3cret",
            redirect_uris=["https://app.example/cb"],
            allowed_scopes=["read", "write"],
            allowed_grant_types=["authorization_code", "refresh_token"],
        )
    ]),
    key_builder=sa.key_builder,
)

async def check_password(name: str, pwd: str) -> OAuth2LoginResult:
    user = await load_user(name)                 # 你的用户表
    if user is None or not verify(pwd, user.pwd_hash):
        return OAuth2LoginResult(message="账号或密码错误")
    return OAuth2LoginResult(login_id=str(user.id))

app.include_router(oauth2_router(
    server, auth,                                # auth = sa.create_logic("user")
    strategy=OAuth2Strategy(do_login_handle=check_password),
))
```

端点：`GET /oauth2/authorize`、`POST /oauth2/doConfirm`、`POST /oauth2/token`、
`POST /oauth2/doLogin`、`GET /oauth2/userinfo`（路径前缀可配）。

协议要点：支持 PKCE（S256/plain）、`state` 透传、授权码与 refresh token
**原子消费**（并发下不可双花）、refresh 轮转（`enable_refresh_rotation`）、
`redirect_uri` 白名单默认拒绝、客户端维度 grant 限制、密钥常量时间比较；
错误响应按 RFC 6749 §5.2 输出 `{"error", "error_description"}`。

### SSO

```python
from liteauth.integration.fastapi import sso_client_router, sso_router
from liteauth.plugin.sso import (
    HttpSsoExchange, InProcessSsoExchange, SsoClient, SsoConfig, SsoServer, SsoStrategy,
)

# 认证中心
sso_server = SsoServer(
    config=SsoConfig(sign_key="<强随机密钥>", allow_url=["https://app.example"]),
    store=sa.store,
    key_builder=sa.key_builder,
)
app.include_router(sso_router(sso_server, admin_auth, strategy=SsoStrategy(
    do_login_handle=check_admin_password,
)))

# 接入方
sso_client = SsoClient(
    config=SsoConfig(sign_key="<同一个密钥>", server_url="https://sso.example",
                     client_id="app1", sso_login_path="/sso/auth"),
    exchange=InProcessSsoExchange(sso_server),   # 跨进程用 HttpSsoExchange(...)
)
app.include_router(sso_client_router(sso_client, user_auth, prefix="/sso"))
```

端点：认证中心 `GET /sso/auth`、`POST /sso/doLogin`、`POST /sso/exchange`、
`POST /sso/signout`；接入方 `GET /sso/callback`。

协议要点：ticket **一次性原子消费**、回调白名单按 `scheme+host+port` 精确匹配
（不再用 `startswith`）、签名范围含 `client_id` 与 `state`（防跨客户端重放与篡改）、
`sign_key` 必填（无弱默认值）。

> 登录页 / 账号校验 / 授权确认页由策略对象注入（`OAuth2Strategy` / `SsoStrategy`
> 的 frozen dataclass），框架不预设实现；不配置时使用内置极简页面。

## 安装

```bash
pip install liteauth

# 带 FastAPI 集成
pip install liteauth[fastapi]

# 带 Redis 存储
pip install liteauth[redis]

# 全量
pip install liteauth[all]
```

## 项目结构

```
liteauth/
├── core/           # 核心认证逻辑（无框架依赖）
│   ├── logic/      # AuthLogic —— 按领域拆分的 Mixin 组装包
│   │   ├── base.py         # 依赖注入 + 共享基础设施
│   │   ├── login.py        # 登录生命周期
│   │   ├── logout.py       # 注销 / 踢人 / 顶人
│   │   ├── token_query.py  # Token 查询校验
│   │   ├── context_auth.py # 上下文身份操作
│   │   ├── disable.py      # 封禁
│   │   ├── permission.py   # 角色 / 权限
│   │   ├── safe_auth.py    # 二级认证
│   │   └── session.py      # 会话操作
│   ├── session.py  # AuthSession — 会话管理
│   ├── router.py   # GuardRule — 路由鉴权器
│   └── ...
├── config/         # 配置模型包（唯一入口，依赖图叶子节点）
│   ├── root.py     # LiteAuthConfig —— 根配置（组合各能力配置）
│   ├── jwt.py      # JwtConfig
│   ├── oauth2.py   # OAuth2Config
│   └── sso.py      # SsoConfig
├── store/          # 存储层抽象
│   ├── base.py     # Store 协议（含 ttl 三态语义与 lock() 锁原语）
│   ├── memory.py   # MemoryStore（纯 dict，零依赖）
│   └── redis_store.py  # RedisStore（含分布式锁）
├── integration/    # Web 框架集成（端点与依赖工厂都在这里，插件核心不含 Web 框架）
│   └── fastapi/    # Depends / Middleware / oauth2_router / sso_router / sso_client_router
├── plugin/         # 插件（协议核心：只依赖 Store / KeyBuilder / 注入的 SPI）
│   ├── jwt/        # JWT Simple / Mixin / Stateless
│   ├── oauth2/     # OAuth2 服务端（loader.py 数据加载 SPI、strategy.py 业务钩子）
│   ├── sso/        # SSO（exchange.py ticket 交换、sign.py 共用签名）
│   └── temp_token/ # 临时 Token（邮件验证链接 / 一次性令牌）
├── strategy/       # 可替换策略
│   ├── key_builder.py  # Redis key 命名规则（所有存储 key 的唯一来源）
│   └── token.py        # Token 生成策略
└── manager.py      # LiteAuthManager 全局管理器
```

### 存储层说明

- `ttl()` 返回三态：`-2` key 不存在（或已过期）/ `-1` 永不过期 / `>=1` 剩余秒数。
  存在且会过期的 key **不返回 0**（`set(ttl<=0)` 的语义是永不过期，而踢人标记、
  活跃时间等会按剩余 TTL 回写，返回 0 会把"即将过期"变成"永久 key"）
- `keys()` 基于游标扫描（Redis SCAN），不使用阻塞的 KEYS 命令，生产安全
- 连接级故障的重试**交给 redis-py 自身**（框架不额外包一层，避免与它的重试叠加、把故障延迟翻倍）：
  redis-py 的 `execute_command` 已内建命令级重试，但**默认 `retry_on_error=[]` 即不重试**，
  需要显式配置：
  ```python
  from redis.backoff import ExponentialBackoff
  from redis.exceptions import ConnectionError as RedisConnectionError
  from redis.retry import Retry

  RedisStore.from_url(                      # kwargs 会透传给 redis-py 客户端
      "redis://127.0.0.1:6379/1",
      retry=Retry(ExponentialBackoff(), 3),          # 重试策略
      retry_on_error=[RedisConnectionError],         # 允许重试的异常
      health_check_interval=30,                      # 空闲连接先探活
  )
  ```
  说明（实测）：redis-py 的重试覆盖"连接已建立后、命令收发中"的故障；连接获取阶段的
  失败（如 `Connection closed by server.`）不在其重试范围内，仍会直接抛出。
- 排障：Windows / 双栈机器上若持续出现 `Error while reading from localhost:6379 : (10054, ...)`
  或 `(22, '指定的网络名不再可用。', None, 64, None)`，通常是客户端用 `localhost` 而它**优先
  解析到 IPv6 `::1`**，Redis 又**监听了 IPv6 但拒绝该连接**（如 `protected-mode` 拒绝 IPv6 回环）
  —— TCP 能连上却被对端立刻关闭，客户端**没有回落机会**（与瞬时抖动无关，重试无效）。
  两种修法：客户端改用 `127.0.0.1`；或让 Redis 根本不监听 IPv6（`bind 127.0.0.1`）——
  此时 `::1` 直接被拒，客户端会自动回落到 IPv4（实测）
- `lock(key, timeout, lease_time)` 提供 per-key 互斥锁：RedisStore 基于 `SET NX EX` + Lua
  释放（分布式锁），MemoryStore 基于 `asyncio.Lock`；同一 asyncio Task 内可重入，
  跨 Task / 跨进程真正互斥。`timeout` 是获取锁的**等待上限**（10 秒），`lease_time` 是
  锁**租约**（持有者崩溃后自动释放，默认 30 秒，见 `LiteAuthConfig.lock_lease_time`）——
  两者独立：临界区可能变慢时应调大租约，而不是动等待上限。内存实现无租约概念，忽略该参数；
  Account-Session 的读-改-写临界区（并发登录 / 踢人 / 顶号）均在锁保护下执行，
  自定义 Store 未实现 `lock()` 时自动退化为无锁（向后兼容）
- `mget()` / `set_many()` 批量原语：Token 续签等热路径用批量读写减少网络往返
  （Redis 下续签从 7 次往返降至 3 次）；自定义 Store 未实现时自动降级为逐个读写
- 自定义 Store 只需实现 `Store` Protocol 的所有成员即可接入

## 许可证

Apache-2.0
