Metadata-Version: 2.4
Name: micosauth
Version: 0.1.2
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` 是一个面向 Python 服务端场景的 Redis 认证鉴权组件。

当前版本聚焦于：

- 多 `realm`
- 基于 `login_id` 的登录模型
- 普通 token 与临时 token
- 多设备、多 token 策略控制
- Web 场景与非 Web 场景共用同一套核心能力

它不是完整 IAM 平台，也不负责用户、角色、权限的数据源管理。  
这些数据应由业务系统通过 `MicosAccessProvider` 提供。

## 特性

- 核心认证、鉴权、会话查询能力解耦
- 支持装饰器、FastAPI `Depends`、编程式调用三种入口
- 支持 URL 和 `host/port/db/password` 两种 Redis 配置方式
- 登录并发控制、ACL 短 TTL 缓存、Redis 故障快速失败
- FastAPI 生命周期自动组合，不覆盖已有 lifespan

## 安装

```bash
pip install micosauth
```

如果需要 FastAPI 适配层：

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

## 公开 API

以下对象视为当前版本的公开接入面：

- 配置与装配
  - `MicosSetting`
  - `MicosRedisSetting`
  - `MicosRealmSetting`
  - `MicosAccessSetting`
  - `MicosDefaultsSetting`
  - `MicosSecuritySetting`
  - `MicosResilienceSetting`
  - `MicosService`
  - `MicosRuntime`
  - `build_runtime()`
  - `build_guard()`
- 核心能力
  - `MicosAuthUtil`
  - `MicosGuard`
  - `MicosSessionUtil`
  - `MicosPermissionUtil`
  - `MicosTokenUtil`
- Provider 协议
  - `MicosAccessProvider`
  - `EmptyMicosAccessProvider`
- 常量
  - `MODE_AND`
  - `MODE_OR`
  - `AUTH_MODES`
- 装饰器
  - `micosauth.guards.require_login`
  - `micosauth.guards.require_roles`
  - `micosauth.guards.require_permissions`
- FastAPI 适配层
  - `install_fastapi_auth()`
  - `set_default_realm()`
  - `get_request_guard()`
  - `require_login_dep()`
  - `require_roles_dep()`
  - `require_permissions_dep()`

不建议直接依赖以下内部实现：

- `micosauth.storage._*`
- `micosauth.decorators._support`
- `micosauth.adapters.fastapi.context` 中未导出的内部细节
- `app.state.micos_*` 之外的临时内部字段

## 设计边界

`micosauth` 负责：

- token 签发、校验、撤销
- 登录态校验
- 角色/权限判断
- 会话查询与统计
- Web 框架适配

业务系统负责：

- 用户是否存在、是否允许登录
- 角色列表
- 权限列表
- 数据范围
- 额外 claims / extra 信息

也就是说，`micosauth` 定义“需要什么 ACL 数据”，但不定义“这些数据从哪里来”。

## 快速开始

### 1. 定义 Provider

```python
from micosauth import MicosAccessProvider


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

    async def get_permissions(self, realm_id: str, login_id: str) -> list[str]:
        if login_id == "admin":
            return ["sys:user:view", "sys:user:edit", "sys:audit:view"]
        return ["sys:user:view"]

    async def get_data_scopes(self, realm_id: str, login_id: str) -> list[str]:
        return ["ALL"] if login_id == "admin" else ["SELF"]

    async def get_extra(self, realm_id: str, login_id: str) -> dict:
        return {"display_name": login_id.upper()}
```

### 2. 创建配置与服务

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

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

service = MicosService(setting)
service.register_realm(
    MicosRealmSetting(
        realm_id="admin",
        token_name="Authorization",
        token_ttl_seconds=3600,
        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=2,
        max_tokens_per_device=1,
        access_provider=AdminAccessProvider(),
    )
)

runtime = build_runtime(service)
```

推荐把 `runtime` 视为统一运行时入口。  
不要在业务函数里重复创建 `MicosAuthUtil` / `MicosSessionUtil`。

### 3. 启动与关闭

`micosauth` 依赖 Redis 连接，因此需要在进程生命周期内做初始化和关闭。

非 Web 场景：

```python
await runtime.startup()
try:
    # 在这里使用 runtime.auth / runtime.guard / runtime.session
    ...
finally:
    await runtime.shutdown()
```

FastAPI 场景：

```python
app = FastAPI()
install_fastapi_auth(app, runtime)
```

安装后会自动把 `runtime.startup()` / `runtime.shutdown()` 接入应用生命周期，不需要你再手动调用。

## 推荐项目结构

推荐把接入代码拆成 3 层：

1. 配置层
   负责读环境变量、构造 `MicosSetting`
2. 装配层
   负责创建 `MicosService`、注册 `realm`、构建 `runtime`
3. 业务层
   只消费 `runtime`、`guard`、FastAPI adapter

一个典型示例：

```python
# auth/bootstrap.py
from micosauth import (
    MicosAccessSetting,
    MicosRealmSetting,
    MicosRedisSetting,
    MicosService,
    MicosSetting,
    build_runtime,
)

from .provider import AdminAccessProvider


def build_auth_runtime():
    setting = MicosSetting(
        redis=MicosRedisSetting(url="redis://:123456@127.0.0.1:6379/15"),
        access=MicosAccessSetting(provider_timeout_seconds=3.0),
    )
    service = MicosService(setting)
    service.register_realm(
        MicosRealmSetting(
            realm_id="admin",
            token_name="Authorization",
            access_provider=AdminAccessProvider(),
        )
    )
    return build_runtime(service)
```

```python
# main.py
from fastapi import FastAPI
from micosauth.adapters.fastapi import install_fastapi_auth
from auth.bootstrap import build_auth_runtime

runtime = build_auth_runtime()

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

这样做的好处是：

- Web、脚本、Worker 可以共用同一套认证装配
- 不会把 provider、Redis 配置散落到各个模块
- 后续扩容多框架或多进程时不需要重写接入逻辑

## Redis 配置

### URL 方式

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

### 字段方式

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

当前版本已覆盖两种连接方式的真实测试。

## 三种使用方式

`micosauth` 当前提供三种一等入口：

1. 编程式调用
2. 装饰器
3. FastAPI `Depends`

三者底层都复用同一套 guard 语义。

### 1. 编程式调用

适合脚本、任务、消息消费者、普通业务服务代码。

```python
from micosauth import MODE_AND, build_guard

guard = build_guard(runtime)

result = await guard.require_login(
    token=token,
    realm="admin",
)

await guard.require_roles(
    ["ADMIN", "AUDITOR"],
    token=token,
    realm="admin",
    mode=MODE_AND,
)

await guard.require_permissions(
    ["sys:user:view", "sys:user:edit"],
    token=token,
    realm="admin",
    mode=MODE_AND,
)
```

最常见的编程式用法有三类：

1. 已经拿到 token，只想校验登录态

```python
result = await guard.require_login(token=token, realm="admin")
print(result.login_id)
```

2. 已经拿到 token，想在业务代码里判断角色/权限

```python
await guard.require_roles("ADMIN", token=token, realm="admin")
await guard.require_permissions("sys:user:edit", token=token, realm="admin")
```

3. 需要先登录，再把 token 返回给调用方

```python
login = await runtime.auth.login("admin", "1001", device_id="pc")
token = login["token"]
```

建议：

- `runtime.auth` 负责登录、登出、反查、下线
- `runtime.guard` 或 `build_guard(runtime)` 负责“要求必须满足某个鉴权条件”
- `runtime.session` 负责会话查询、统计、分析

### 2. 装饰器

适合少量声明式鉴权场景。

```python
from micosauth import MODE_AND
from micosauth.guards import require_login, require_permissions, require_roles


@require_login(realm="admin")
async def login_required(request):
    return {"login_id": request.state.micos_login_id}


@require_roles(["ADMIN", "AUDITOR"], realm="admin", mode=MODE_AND)
async def role_required(request):
    return {"ok": True}


@require_permissions(["sys:user:edit", "sys:audit:view"], realm="admin", mode=MODE_AND)
async def permission_required(request):
    return {"ok": True}
```

装饰器更适合：

- 少量路由函数
- 内部管理接口
- 你希望直接在函数定义上看到鉴权声明

不建议用装饰器解决所有接入问题，复杂业务代码更适合直接调用 `guard`。

### 3. FastAPI `Depends`

适合路由声明式接入。

```python
from fastapi import Depends, FastAPI, Request
from micosauth.adapters.fastapi import install_fastapi_auth, require_login_dep

app = FastAPI()
install_fastapi_auth(app, runtime)


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

依赖执行成功后，认证结果会自动写入 `request.state`，常用字段有：

- `request.state.micos_token`
- `request.state.micos_realm_id`
- `request.state.micos_login_id`
- `request.state.micos_claims`
- `request.state.micos_session`
- `request.state.micos_acl`

如果依赖执行失败：

- 未登录或 token 无效时返回 `401`
- 角色/权限不足时返回 `403`

## FastAPI 接入

下面给一个更完整的 FastAPI 示例。

```python
from fastapi import Depends, FastAPI, Request

from micosauth import build_runtime
from micosauth.adapters.fastapi import (
    get_request_guard,
    install_fastapi_auth,
    require_login_dep,
    require_permissions_dep,
    require_roles_dep,
)

runtime = build_runtime(service)

app = FastAPI()
install_fastapi_auth(app, runtime)


@app.post("/auth/login")
async def login(payload: dict, request: Request):
    return await request.app.state.micos_auth.login(
        "admin",
        str(payload.get("login_id") or ""),
        device_id=str(payload.get("device_id") or "default"),
    )


@app.get("/me", dependencies=[Depends(require_login_dep(realm="admin"))])
async def me(request: Request):
    return {
        "login_id": request.state.micos_login_id,
        "roles": list(request.state.micos_acl.roles),
        "permissions": list(request.state.micos_acl.permissions),
    }


@app.get("/admin-only", dependencies=[Depends(require_roles_dep("ADMIN", realm="admin"))])
async def admin_only():
    return {"ok": True}


@app.get(
    "/user-edit",
    dependencies=[Depends(require_permissions_dep("sys:user:edit", realm="admin"))],
)
async def user_edit():
    return {"ok": True}


@app.get("/manual")
async def manual(request: Request):
    result = await get_request_guard(request, realm="admin").require_login()
    return {"login_id": result.login_id}
```

### FastAPI 中如何组织路由

推荐按 realm 或业务域组织：

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

admin_router = APIRouter(prefix="/admin", tags=["admin"])
set_default_realm(admin_router, "admin")


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


@admin_router.get("/users", dependencies=[Depends(require_roles_dep("ADMIN"))])
async def admin_users():
    return {"ok": True}
```

这样能减少每个路由都重复写 `realm="admin"`。

### 安装到应用

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

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

默认行为：

- 自动把 runtime 挂到 `app.state`
- 自动组合 FastAPI lifespan
- 不覆盖已有 lifespan，而是在外层组合执行
- 不做隐式自动登录

当前会挂载的状态字段：

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

### 模块级默认 realm

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

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


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

### 请求级编程式调用

适合你不想只依赖 `Depends` 的场景。

```python
from fastapi import Request
from micosauth.adapters.fastapi import get_request_guard


async def handler(request: Request):
    result = await get_request_guard(request, realm="admin").require_login()
    return {"login_id": result.login_id}
```

## 非 Web 场景

任务、脚本、消息消费者不需要框架 adapter。

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

    guard = build_guard(runtime)
    await guard.require_login(token=login["token"], realm="admin")
finally:
    await runtime.shutdown()
```

推荐把 `startup()` / `shutdown()` 放在进程入口，不要放在业务函数内部。

### 脚本场景

```python
async def main():
    runtime = build_auth_runtime()
    await runtime.startup()
    try:
        token = "..."
        result = await runtime.guard.require_login(token=token, realm="admin")
        print(result.login_id)
    finally:
        await runtime.shutdown()
```

### Worker / 消息消费场景

```python
async def handle_message(message: dict):
    token = str(message["token"])
    realm_id = str(message["realm_id"])
    await runtime.guard.require_permissions(
        "sys:job:execute",
        token=token,
        realm=realm_id,
    )
```

### 定时任务或管理任务

如果是系统任务，不一定需要 token，可以直接使用 `runtime.session` 和 `runtime.auth` 做后台操作：

```python
sessions = await runtime.session.list_sessions("admin")
await runtime.auth.kickout_login_id("admin", "1001")
```

## 登录、校验、下线

### 登录

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

token = result["token"]
```

登录参数说明：

- 第一个参数：`realm_id`
- 第二个参数：`login_id`
- `device_id`
  建议始终传，便于多端策略控制
- `extra`
  业务扩展字段，会写入 token/session 相关记录
- `audit`
  审计字段，适合记录 IP、来源、渠道等信息

返回值示例：

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

### 校验与反查

```python
valid = await runtime.auth.is_token_valid(token, "admin")
inspect_result = await runtime.auth.inspect_token(token, "admin")
login_id = await runtime.auth.get_login_id_by_token(token, "admin")
claims = await runtime.auth.get_claims_by_token(token, "admin")
```

这些接口适用场景：

- `is_token_valid`
  只想知道 token 是否有效
- `inspect_token`
  需要拿到完整登录态、ACL、claims、session
- `get_login_id_by_token`
  只想做 token 到账号的反查
- `get_claims_by_token`
  想拿 token 元数据而不是完整 ACL

### 下线与撤销

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

兼容别名：

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

## 会话查询与统计

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

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

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

常见用途：

- 管理后台查看当前在线账号
- 查询单账号当前有多少设备、多少 token
- 分析一段时间内 token 创建趋势
- 统计各 realm 的在线分布

## 临时 token

### 创建

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

### 校验

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

如果启用：

```python
MicosSecuritySetting(temp_token_one_time=True)
```

则临时 token 成功校验后会被立即消费。

适用场景：

- 一次性下载链接
- 回调确认链接
- 短期共享访问凭证
- 非登录态下的短时业务授权

## Realm 策略

```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`
  单账号最大设备数，`0` 表示不限制
- `max_tokens_per_login_id`
  单账号最大 token 数，`0` 表示不限制
- `max_tokens_per_device`
  单设备最大 token 数，`0` 表示不限制

## `AND` / `OR`

角色与权限判断都支持：

- `MODE_AND`
- `MODE_OR`

```python
from micosauth import MODE_AND, MODE_OR

await runtime.guard.require_roles(
    ["ADMIN", "AUDITOR"],
    token=token,
    realm="admin",
    mode=MODE_AND,
)

await runtime.guard.require_permissions(
    ["sys:user:*", "sys:log:view"],
    token=token,
    realm="admin",
    mode=MODE_OR,
)
```

权限判断支持通配符匹配。

## 生产建议

### Redis

- 建议使用独立库或独立 `redis_prefix`
- 建议设置 `client_name`
- 高并发场景压测 `max_connections`、`socket_timeout`
- 跨机房或云 Redis 保持 `socket_keepalive=True`

### Provider

- 避免每次鉴权都直连慢 SQL
- `get_roles/get_permissions/get_data_scopes/get_extra` 应尽量走缓存或聚合服务
- 生产建议保留短 TTL ACL 缓存，不要把每次权限判断都打到数据库

### 运行方式

- 整个进程只创建一次 `runtime`
- Web、任务、脚本共用同一套装配代码
- 不要在业务函数里重复 `startup()` / `shutdown()`

### 如何选择入口

- 业务代码里主动做鉴权：
  选 `guard`
- 少量函数声明式保护：
  选装饰器
- FastAPI 路由统一声明式接入：
  选 `Depends`
- FastAPI 中偶尔需要手动取当前登录态：
  选 `get_request_guard()`

### 安全与边界

- 临时 token 下载、确认、回调建议开启 `temp_token_one_time=True`
- 当前 token 默认按明文存储，便于人工排查和会话管理
- 如果你需要更强的密文策略，应在接入前自行评估与扩展

## 常见接入问题

### 1. 为什么还要自己实现 Provider

因为 `micosauth` 是组件，不负责存储用户、角色、权限主数据。  
它只负责认证流程和鉴权判断，业务数据来源必须由你自己提供。

### 2. 应该直接用 `MicosAuthUtil` 还是 `MicosGuard`

- 做登录、登出、撤销、会话控制、token 反查：
  用 `MicosAuthUtil`
- 做“必须登录”“必须有某角色”“必须有某权限”：
  用 `MicosGuard`

### 3. FastAPI 里为什么推荐 `request.state`

因为依赖或 request guard 成功后，会把当前请求的认证结果写入 `request.state`，这样后续 handler 不需要重复解析 token。

### 4. 一个项目多个 realm 怎么处理

- 不同业务域使用不同 `realm_id`
- 路由层尽量显式传 `realm`
- 或者在 router 级使用 `set_default_realm()`

### 5. token 应该放 header 还是 cookie

都支持。实际读取使用的是当前 realm 的 `token_name`：

- 先读 header
- 再读 cookie

例如：

```python
MicosRealmSetting(
    realm_id="admin",
    token_name="Authorization",
)
```

那就会优先读取 `Authorization` 请求头，否则再读取同名 cookie。

## 当前状态

当前版本已经覆盖的测试范围包括：

- Redis URL 连接
- Redis 非 URL 连接
- FastAPI 集成
- request guard
- 编程式 guard
- 多 realm
- 多设备、多 token 策略
- ACL provider 超时、缓存
- 角色/权限 `AND` / `OR`
- 权限通配符
- 临时 token

当前仓库测试结果：

- `52 passed`

## 生产可用性判断

当前版本适合：

- 内部系统接入
- 自己可控的服务
- 受控灰度上线

如果你的要求是“稳定组件”，建议把下面几条视为接入前检查项：

1. 明确公开 API，只依赖 README 中声明的接入面。
2. 按真实流量压测 Redis 连接池、provider 超时和登录并发。
3. 对自身业务 provider 做缓存与降载设计。
4. 明确 token 生命周期、踢下线策略和临时 token 使用边界。

## 测试

运行全量测试：

```bash
pytest -q tests
```

当前仅剩 1 个测试告警，来自 `fastapi/starlette/httpx` 上游依赖栈，不属于项目逻辑错误。
