05 · 账号与权限：pbkdf2、会话与"判定收敛"的越权教训

PBKDF2_ITERATIONS = 120_000

def hash_password(password: str, salt: str | None = None) -> tuple[str, str]:
    """返回 (salt, hex_hash)。"""
    salt = salt or secrets.token_hex(16)
    dk = hashlib.pbkdf2_hmac(
        "sha256", password.encode("utf-8"), bytes.fromhex(salt), PBKDF2_ITERATIONS
    )
    return salt, dk.hex()

def verify_password(password: str, salt: str, expected_hash: str) -> bool:
    _, h = hash_password(password, salt)
    return secrets.compare_digest(h, expected_hash)

def create_session(conn, user_id: int) -> str:
    token = secrets.token_hex(32)               # 64 个十六进制字符的高熵随机串
    conn.execute(
        "INSERT INTO sessions(token,user_id,created_at,expires_at) VALUES(?,?,?,?)",
        (token, user_id, now(), session_expiry()),
    )
    conn.commit()
    return token

def make_user_dependency(get_db_path):
    """构造绑定到具体 app 的当前用户依赖。get_db_path(request)->str"""
    from fastapi import Request

    def dep(request: Request, authorization: str = Header(default="")) -> dict:
        if not authorization.lower().startswith("bearer "):
            raise HTTPException(status_code=401, detail="未登录或缺少 Token")
        token = authorization.split(" ", 1)[1].strip()
        conn = get_conn(get_db_path(request))
        try:
            row = conn.execute(
                "SELECT u.*, s.expires_at AS session_expires_at "
                "FROM sessions s JOIN users u ON u.id = s.user_id WHERE s.token=?",
                (token,),
            ).fetchone()
            if row is not None:
                expires_at = row["session_expires_at"] or ""
                if not expires_at or expires_at <= now():
                    conn.execute("DELETE FROM sessions WHERE token=?", (token,))
                    conn.commit()
                    row = None
        finally:
            conn.close()
        if row is None:
            raise HTTPException(status_code=401, detail="登录已过期，请重新登录")
        user = _row_to_user(row)
        user["db_path"] = get_db_path(request)
        user["token"] = token   # 内部字段：对外返回用户信息前必须 pop
        return user

    return dep


def make_admin_dependency(user_dep):
    """在 user_dep 基础上要求 admin 角色。"""
    def dep(user: dict = Depends(user_dep)) -> dict:
        if user["role"] != "admin":
            raise HTTPException(status_code=403, detail="需要管理员权限")
        return user
    return dep

@router.get("/users")
def list_users(admin: dict = Depends(admin_dep)):   # 后台接口统一挂 admin 依赖
    ...

@router.get("/conversations")
def list_conversations(user: dict = Depends(user_dep)):  # 登录即可
    ...

# access.py —— 文件开头的自白
"""可见性与访问控制规则：助手与知识库统一按「部门匹配 + 显式授权」判定。
...
聊天发消息、RAG 检索、工作流 knowledge_search 节点都必须走这里的同一套判定，
避免"列表处校验、使用处漏校验"的横向越权。
"""

def assistant_accessible(conn, assistant_row, user: dict) -> bool:
    """岗位助手可见性。assistant_row 需含 id/departments 字段。"""
    if user.get("role") == "admin":
        return True
    allowed = split_depts(assistant_row["departments"])
    if not allowed or _overlaps(user.get("department", ""), allowed):
        return True                        # departments 为空 = 全部门可用
    granted = conn.execute(
        "SELECT 1 FROM user_assistants WHERE user_id=? AND assistant_id=?",
        (user["id"], assistant_row["id"]),
    ).fetchone()
    return granted is not None