Metadata-Version: 2.4
Name: micosauth
Version: 0.1.1
Summary: Redis-based auth and session package for Python applications
Author-email: CharlieZhang <charlie@example.com>
License-Expression: MIT
Project-URL: Homepage, https://github.com/jiangbyte/micosauth
Project-URL: Documentation, https://github.com/jiangbyte/micosauth
Project-URL: Repository, https://github.com/jiangbyte/micosauth
Project-URL: Issues, https://github.com/jiangbyte/micosauth/issues
Keywords: auth,authentication,authorization,redis,fastapi,session,token
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
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: Framework :: FastAPI
Classifier: Topic :: Security
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Requires-Dist: fastapi>=0.136.1
Requires-Dist: redis>=7.4.0
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.136.1; extra == "fastapi"
Requires-Dist: starlette>=0.37.2; extra == "fastapi"
Provides-Extra: dev
Requires-Dist: pytest>=8.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.25.0; extra == "dev"
Requires-Dist: httpx>=0.28.0; extra == "dev"

# micosauth 使用文档

`micosauth` 是一个基于 Redis 的通用认证鉴权包，当前版本核心模型是：

- 多 realm
- 基于 `login_id` 登录
- 默认 64 位字符串 token
- 支持多设备、多 token 策略控制
- 普通 token 与临时 token
- `MicosAuthUtil` 负责认证、鉴权、撤销、反查
- `MicosSessionUtil` 负责查询、统计、分布分析

## 1. 核心对象

- `MicosSetting`
  启动配置对象
- `MicosService`
  启动装配对象
- `MicosAuthUtil`
  主认证鉴权工具
- `MicosSessionUtil`
  会话查询统计工具
- `MicosTokenUtil`
  token 生成工具

## 2. 安装

```bash
pip install micosauth
```

如需 FastAPI 适配层：

```bash
pip install "micosauth[fastapi]"
```

## 3. 启动接入

```python
from micosauth import (
    MicosAccessSetting,
    build_runtime,
    MicosRealmSetting,
    MicosRedisSetting,
    MicosSecuritySetting,
    MicosService,
    MicosSetting,
)

micos_setting = MicosSetting(
    redis=MicosRedisSetting(url="redis://:123456@127.0.0.1:6379/1"),
    access=MicosAccessSetting(
        provider_timeout_seconds=3.0,
        acl_cache_ttl_seconds=5,
        acl_cache_maxsize=10000,
    ),
    security=MicosSecuritySetting(
        temp_token_one_time=True,
    ),
)

micos_service = MicosService(micos_setting)
micos_service.register_realm(MicosRealmSetting(realm_id="admin"))
micos_service.register_realm(MicosRealmSetting(realm_id="user"))

runtime = build_runtime(micos_service)
```

推荐把 `runtime` 作为唯一运行时入口，在 Web、任务进程、脚本里统一复用。  
不建议业务代码自己重复创建 `MicosAuthUtil` / `MicosSessionUtil`。

## 3.1 Redis 配置

`MicosRedisSetting` 既支持 `url`，也支持代码字段配置。

最简单的 URL 方式：

```python
MicosRedisSetting(
    url="redis://:123456@127.0.0.1:6379/1",
)
```

代码字段方式：

```python
MicosRedisSetting(
    host="127.0.0.1",
    port=6379,
    db=1,
    username="default",
    password="123456",
    ssl=False,
    max_connections=500,
    socket_connect_timeout=5,
    socket_timeout=10,
    socket_keepalive=True,
    retry_on_timeout=True,
    health_check_interval=30,
    client_name="micosauth-admin",
)
```

当前主要连接参数：

- `url`
  Redis 连接串。设置后优先使用
- `host` / `port` / `db`
  代码方式指定连接地址
- `username` / `password`
  ACL 认证信息
- `ssl`
  是否启用 TLS
- `max_connections`
  连接池大小
- `socket_connect_timeout`
  建连超时秒数
- `socket_timeout`
  读写超时秒数
- `socket_keepalive`
  是否启用 TCP keepalive
- `retry_on_timeout`
  超时后是否重试
- `health_check_interval`
  连接健康检查间隔秒数
- `client_name`
  Redis 客户端名，便于服务端排查
- `client_kwargs`
  透传给 `redis.asyncio.Redis` / `redis.from_url` 的额外参数

高并发建议：

- 在线业务建议明确设置 `max_connections`，不要完全依赖默认值
- API 峰值较高时，可先从 `200` 提升到 `500` 或 `1000`，但要和 Redis 服务端 `maxclients` 一起评估
- `socket_connect_timeout` 建议控制在 `3-5s`
- `socket_timeout` 建议控制在 `5-15s`
- 跨机房或云 Redis 建议开启 `socket_keepalive=True`
- 生产环境建议设置 `client_name`，方便在 Redis 侧追踪连接来源
- 如果有更细粒度连接池需求，可以通过 `client_kwargs` 继续传递底层 redis-py 参数

## 3.2 配置总表

`MicosSetting` 由四部分组成：

- `redis`
  Redis 连接与连接池参数
- `defaults`
  token 长度、临时 token 长度、Redis key 前缀
- `security`
  安全相关开关
- `access`
  ACL/provider 调用超时与本地缓存参数
- `resilience`
  并发锁与 Redis 故障快速失败参数

完整示例：

```python
from micosauth import (
    MicosAccessSetting,
    MicosDefaultsSetting,
    MicosRedisSetting,
    MicosResilienceSetting,
    MicosSecuritySetting,
    MicosSetting,
)

setting = MicosSetting(
    redis=MicosRedisSetting(
        url="redis://:123456@127.0.0.1:6379/1",
        max_connections=500,
        socket_connect_timeout=5,
        socket_timeout=10,
        client_name="micosauth-admin",
    ),
    defaults=MicosDefaultsSetting(
        token_length=64,
        temp_token_length=64,
        redis_prefix="micosauth",
    ),
    security=MicosSecuritySetting(
        temp_token_one_time=True,
    ),
    access=MicosAccessSetting(
        provider_timeout_seconds=3.0,
        acl_cache_ttl_seconds=5,
        acl_cache_maxsize=10000,
    ),
    resilience=MicosResilienceSetting(
        login_lock_seconds=5,
        login_lock_retry_interval_ms=50,
        login_lock_wait_timeout_seconds=5,
        redis_circuit_breaker_failures=5,
        redis_circuit_breaker_reset_seconds=30,
    ),
)
```

配置建议：

- `defaults.token_length`
  建议不低于 `64`
- `defaults.temp_token_length`
  建议不低于 `64`
- `defaults.redis_prefix`
  多项目共享 Redis 时建议按业务隔离
- `security.temp_token_one_time`
  下载、回调确认、一次性授权建议开启
- `access.provider_timeout_seconds`
  建议 `1-5s`，避免下游 ACL 提供器拖垮鉴权链路
- `access.acl_cache_ttl_seconds`
  建议 `1-30s`，高并发场景通常需要短 TTL 抗抖
- `access.acl_cache_maxsize`
  建议按单进程并发主体数评估，常见可从 `10000` 起步
- `resilience.login_lock_seconds`
  建议 `3-10s`
- `resilience.login_lock_retry_interval_ms`
  建议 `20-100ms`
- `resilience.login_lock_wait_timeout_seconds`
  建议 `3-10s`
- `resilience.redis_circuit_breaker_failures`
  建议 `3-10`
- `resilience.redis_circuit_breaker_reset_seconds`
  建议 `10-60s`

## 4. Realm 配置

`MicosRealmSetting` 当前支持的核心策略项：

```python
MicosRealmSetting(
    realm_id="admin",
    token_name="Authorization",
    token_ttl_seconds=2592000,
    temp_token_ttl_seconds=300,
    allow_multi_device_login=True,
    keep_old_token_on_same_device_login=False,
    keep_old_token_on_new_device_login=True,
    max_devices_per_login_id=2,
    max_tokens_per_login_id=5,
    max_tokens_per_device=1,
)
```

字段含义：

- `allow_multi_device_login`
  是否允许多设备同时登录
- `keep_old_token_on_same_device_login`
  同设备再次登录时是否保留旧 token
- `keep_old_token_on_new_device_login`
  新设备登录时是否保留其他设备旧 token
- `max_devices_per_login_id`
  每个 `login_id` 允许的最大设备数，`0` 表示不限制
- `max_tokens_per_login_id`
  每个 `login_id` 允许的最大 token 数，`0` 表示不限制
- `max_tokens_per_device`
  每个设备允许的最大 token 数，`0` 表示不限制

## 5. 注册 AccessProvider

使用方需要自己实现权限、角色、数据范围获取逻辑。

```python
from micosauth import MicosAccessProvider


class AdminAccessProvider:
    async def get_roles(self, realm_id: str, login_id: str) -> list[str]:
        return ["ADMIN"]

    async def get_permissions(self, realm_id: str, login_id: str) -> list[str]:
        return ["sys:user:view", "sys:session:page"]

    async def get_data_scopes(self, realm_id: str, login_id: str) -> list[str]:
        return ["ALL"]

    async def get_extra(self, realm_id: str, login_id: str) -> dict:
        return {"nickname": "管理员"}
```

注册方式：

```python
micos_service.register_access_provider("admin", AdminAccessProvider())
```

生产建议：

- `get_roles/get_permissions/get_data_scopes/get_extra` 应避免直接访问慢 SQL，建议走本地缓存或上游聚合服务
- `micosauth` 会并行拉取 ACL 四个维度，并使用 `access.provider_timeout_seconds` 做统一超时保护
- `micosauth` 默认会做进程内 ACL 短 TTL 缓存；如需权限立刻生效，可调用 `service.invalidate_acl_cache(realm_id, login_id)`

## 6. 登录与普通 token

登录只需要 `login_id`，推荐同时传入 `device_id`：

```python
result = await micos_auth.login(
    "admin",
    "1001",
    device_id="pc",
    extra={"nickname": "管理员"},
)
```

返回示例：

```python
{
    "token": "...64位字符串...",
    "login_id": "1001",
    "realm_id": "admin",
    "device_id": "pc",
    "expires_at": "2026-06-14T12:00:00+00:00",
}
```

说明：

- 当前模型中不再暴露独立 `session_id`
- 会话主体围绕 `realm_id + login_id` 组织
- token 是登录明细单元

## 7. 鉴权与反查

```python
is_valid = await micos_auth.is_token_valid(token, "admin")
inspect_result = await micos_auth.inspect_token(token, "admin")
login_id = await micos_auth.get_login_id_by_token(token, "admin")
roles = await micos_auth.get_roles("admin", "1001")
permissions = await micos_auth.get_permissions("admin", "1001")
data_scopes = await micos_auth.get_data_scopes_by_login_id("admin", "1001")
```

## 8. 临时 token

创建：

```python
temp = await micos_auth.create_temp_token(
    "admin",
    temp_id="file-1001",
    type="download",
    time=300,
    extra={"filename": "report.xlsx"},
)
```

校验：

```python
ok = await micos_auth.verify_temp_token(
    temp["token"],
    "admin",
    type="download",
)
```

如果开启 `MicosSecuritySetting(temp_token_one_time=True)`，则 `verify_temp_token(...)` 成功后会立即消费该 token，后续不可重复使用。

## 9. 会话查询与统计

当前查询模型：

- `session` 是 `login_id` 级聚合
- `token` 是具体登录明细
- `device` 是 token 分组维度

```python
sessions = await micos_session.list_sessions("admin")
page = await micos_session.page_sessions("admin", current=1, size=20)
login_session = await micos_session.get_session("admin", "1001")
tokens = await micos_session.list_tokens("admin", "1001")
device_tokens = await micos_session.list_device_sessions("admin", "1001", "pc")
device_summaries = await micos_session.list_device_summaries("admin", "1001")

online_count = await micos_session.count_online_sessions("admin")
login_token_count = await micos_session.count_login_id_sessions("admin", "1001")
device_token_count = await micos_session.count_device_tokens("admin", "1001", "pc")

analysis = await micos_session.get_analysis()
chart = await micos_session.get_chart_data(days=7)
distribution = await micos_session.get_realm_distribution()
```

统计对象说明：

- `get_analysis()`
  返回总登录数、总 token 数、总设备数、最近一小时新增 token 数
- `get_chart_data(days=7)`
  返回按天统计的 token 创建趋势
- `get_realm_distribution()`
  返回各 realm 的登录数、token 数、设备数分布
- `list_device_summaries("admin", "1001")`
  直接返回某个 `login_id` 下各设备摘要，包括 token 数、最近登录时间、最近访问时间

## 10. 撤销与下线

```python
await micos_auth.revoke_token("admin", token)
await micos_auth.revoke_login("admin", "1001")
await micos_auth.kickout_login_id("admin", "1001")
await micos_auth.kickout_device("admin", "1001", "pc")
```

兼容接口：

```python
await micos_auth.logout_current("admin", token)
await micos_auth.logout_login_id("admin", "1001")
await micos_auth.logout_device("admin", "1001", "pc")
```

## 11. Redis 存储模型

当前 Redis 主结构围绕 `realm_id + login_id`：

- `micosauth:{realm_id}:{login_id}:session`
  登录主体聚合数据
- `micosauth:{realm_id}:{login_id}:tokens`
  token 明细 `HASH`
- `micosauth:{realm_id}:{login_id}:device_tokens:{device_id}`
  设备维度 token 明细 `HASH`
- `micosauth:{realm_id}:token_lookup:{token}`
  token 反查 `login_id`
- `micosauth:{realm_id}:login_index`
  realm 级 login 分布索引
- `micosauth:{realm_id}:token_index`
  token 过期索引
- `micosauth:{realm_id}:created_index`
  token 创建时间索引
- `micosauth:{realm_id}:temp:{token}`
  临时 token

说明：

- 统计查询已改为批量 `mget` / pipeline，避免明显的 N+1 Redis 往返
- 登录过程会对 `realm_id + login_id` 加短时 Redis 锁，降低并发登录下的配额穿透风险
- 新 token 先落库，再执行旧 token 撤销计划，避免“旧 token 先删但新 token 写失败”的空窗
- token 默认按明文存储与索引，便于人工管理和运维排查

## 12. FastAPI 接入

```python
from fastapi import FastAPI
from micosauth import build_runtime
from micosauth.adapters.fastapi import install_fastapi_auth

runtime = build_runtime(micos_service)
app = FastAPI()
install_fastapi_auth(app, runtime)
```

说明：

- `install_fastapi_auth(app, runtime)` 默认会自动接管并合并 FastAPI lifespan
- 不要求你改 `FastAPI(...)` 的构造方式
- 不会覆盖用户已有 lifespan，而是在其外层组合执行
- `MicosService.init()` / `close()` 是幂等的，重复调用不会重复初始化或重复关闭
- FastAPI dependency 默认推荐显式传 `realm`
- 如需减少重复配置，可以在 `app.state` 或请求上下文绑定默认 `realm`
- decorator 仍建议显式传 `realm`
- middleware 不做默认自动认证
- `micosauth` 不做隐式猜测，只读取你显式绑定的默认 `realm`

当前 FastAPI 状态字段：

- `app.state.micos_runtime`
- `app.state.micos_service`
- `app.state.micos_auth`
- `app.state.micos_session`

接入建议：

- 整个服务进程只构建一次 `runtime`
- HTTP、任务消费、管理脚本共用同一套 `service` / `runtime` 装配代码
- 把 `realm` 固定在 router 级或模块级，减少散落在 handler 里的重复配置
- 如果是更敏感的生产环境，建议至少开启：
  - `temp_token_one_time=True`

示例：

```python
from fastapi import Depends
from micosauth.adapters.fastapi import require_login_dep


@app.get("/admin/me", dependencies=[Depends(require_login_dep(realm="admin"))])
async def admin_me():
    return {"ok": True}
```

如需模块级默认 realm 绑定：

```python
from fastapi import APIRouter, Depends
from micosauth.adapters.fastapi import require_login_dep, set_default_realm

admin_router = APIRouter(prefix="/admin")
set_default_realm(admin_router, "admin")


@admin_router.get("/me", dependencies=[Depends(require_login_dep())])
async def admin_me(request):
    return {
        "realm_id": request.state.micos_realm_id,
        "login_id": request.state.micos_login_id,
    }
```

如果你是整站单 realm，也可以在安装时直接指定：

```python
install_fastapi_auth(app, runtime, default_realm_id="admin")
```

## 13. 非 Web 进程接入

脚本、任务消费者、定时任务不需要 adapter，直接使用 `runtime`：

```python
runtime = build_runtime(micos_service)
await runtime.startup()
try:
    result = await runtime.auth.login("admin", "1001", device_id="worker")
    sessions = await runtime.session.list_sessions("admin")
finally:
    await runtime.shutdown()
```

推荐做法：

- Web 进程与任务进程共用同一套 `build_service()` / `build_runtime()` 装配代码
- 把 `runtime.startup()` / `shutdown()` 放在进程入口，不要放在业务函数内部
- 不要在业务函数内部重复创建 `MicosAuthUtil`

## 14. 生产建议

- Redis 建议使用独立库或独立前缀
- `client_name` 建议按服务名区分
- 高并发场景应压测 `max_connections`、`socket_timeout`、`login_lock_wait_timeout_seconds`
- 高并发鉴权场景建议保留短 TTL ACL 缓存，不要把每次权限判断都直打数据库
- `temp_token_one_time=True` 适合一次性下载、确认、回调
- 当前包默认保留 token 明文，便于人工管理、客服排查、运营处理

已完成的稳定性增强：

- FastAPI 生命周期自动组合，不覆盖用户已有 lifespan
- session 查询和统计已去掉明显的 N+1
- ACL/provider 调用支持并行拉取、统一超时和进程内短 TTL 缓存
- 登录过程使用 Redis 锁控制并发配额
- 登录主链路中的新 token 写入、旧 token 撤销、session 更新已收敛到 Lua 脚本提交
- `revoke_token` 与 `touch_token` 也已收敛到 Lua 脚本路径
- temp token 单次消费使用 Lua 原子删除
- 登录锁释放使用 Lua compare-and-delete
- Redis 故障达到阈值后直接 fail-fast，不做静默降级

当前边界：

- 登录关键写路径已是 Lua 提交，但查询、统计、部分聚合刷新路径仍是普通 Redis 调用
- ACL 缓存是单进程内存缓存，多副本部署下属于最终一致
- 当前更适合作为“服务内部认证/会话组件”，不是完整 IAM 平台

## 15. 升级说明

当前版本相较旧接入方式，推荐统一为：

1. 使用 `build_runtime(service)` 作为唯一运行时入口
2. FastAPI 使用 `install_fastapi_auth(app, runtime)`
3. 包源码统一位于 `micosauth/`
4. 如需一次性临时 token，再显式开启 `MicosSecuritySetting(temp_token_one_time=True)`

## 16. OR / AND 鉴权模式

角色和权限校验都支持：

- `AND`
- `OR`
- 权限支持通配符匹配

```python
await micos_auth.require_roles(token, ["ADMIN", "AUDITOR"], "admin", mode="AND")
await micos_auth.require_permissions(token, ["sys:user:*", "sys:log:view"], "admin", mode="OR")
```

## 17. 当前状态

当前版本已经完成：

- Redis 真测
- FastAPI 集成真测
- 多 realm 接入真测
- 多设备、多 token 策略真测
- ACL provider 超时/缓存真测
- 设备摘要查询真测
- 角色、权限 `AND/OR` 模式测试
- 权限通配符测试
- 包构建真测

测试约定：

- 仓库只保留根 `tests/` 作为唯一测试目录
- 发布包不携带测试代码
