张胜宇
e239eb778b
1) 客服 Agent 四份交付文档 + 构建脚手架：品牌由包装占位 XX科技 / 旧名 南方财富 统一为南方基金（热线 400-889-8899 / 官网 nffund.com），系统名改为「智能服务系统」； 同步追加 §0.4 修订记录行，工程记录行保留原占位字面以支撑硬编码扫描验收。 2) 开发文档：清理 28 份已作废/残留文档（14 份移出归档 + 14 份仓库副本）， 新增《文档规整方案与开发前待决事项-2026-09-17》。 3) 客服agent 四份交付文档首次纳入本分支。
97 lines
3.0 KiB
Python
97 lines
3.0 KiB
Python
"""只读探查：`sys_user` 的认证字段现状 —— 判断"账号密码换令牌"这条路通不通。
|
||
|
||
只做 SELECT；`password_hash` **只统计前缀特征与长度，不输出完整值**（金融项目的基本习惯，
|
||
即使是测试库）。
|
||
|
||
python tools/probe_auth_state.py
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
from collections import Counter
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from sqlalchemy import text
|
||
|
||
from app.infrastructure.db import SessionFactory
|
||
|
||
OUTPUT = Path("docs/evidence/auth-state.json")
|
||
|
||
#: 常见密码哈希方案的特征前缀。任何一项命中，说明库里存在**真实可校验**的密码。
|
||
KNOWN_HASH_PREFIXES = (
|
||
"$2a$", "$2b$", "$2y$", # bcrypt
|
||
"$argon2", # argon2
|
||
"$pbkdf2", "$pbkdf2-sha256", # passlib pbkdf2
|
||
"$5$", "$6$", "$1$", # crypt sha256/sha512/md5
|
||
"$scrypt",
|
||
)
|
||
|
||
|
||
def _classify(values: list[str]) -> dict[str, Any]:
|
||
prefixes: Counter[str] = Counter()
|
||
lengths: Counter[int] = Counter()
|
||
for value in values:
|
||
lengths[len(value)] += 1
|
||
matched = next(
|
||
(prefix for prefix in KNOWN_HASH_PREFIXES if value.startswith(prefix)), None
|
||
)
|
||
prefixes[matched or f"<未知格式，首字符 {value[:1]!r}>"] += 1
|
||
return {
|
||
"length_distribution": dict(sorted(lengths.items())),
|
||
"prefix_distribution": dict(prefixes.most_common()),
|
||
}
|
||
|
||
|
||
async def collect() -> dict[str, Any]:
|
||
report: dict[str, Any] = {}
|
||
async with SessionFactory() as session:
|
||
rows = (
|
||
await session.execute(
|
||
text(
|
||
"SELECT id, username, user_type, status, password_hash "
|
||
"FROM sys_user ORDER BY id"
|
||
)
|
||
)
|
||
).mappings().all()
|
||
|
||
report["user_count"] = len(rows)
|
||
report["by_status"] = dict(
|
||
Counter(str(row["status"]) for row in rows).most_common()
|
||
)
|
||
report["by_user_type"] = dict(
|
||
Counter(str(row["user_type"]) for row in rows).most_common()
|
||
)
|
||
report["password_hash_shape"] = _classify(
|
||
[str(row["password_hash"] or "") for row in rows]
|
||
)
|
||
# 判断"能不能校验密码"的关键结论，直接给出来。
|
||
hashes = [str(row["password_hash"] or "") for row in rows]
|
||
report["has_real_password_hash"] = any(
|
||
value.startswith(KNOWN_HASH_PREFIXES) for value in hashes
|
||
)
|
||
report["placeholder_examples"] = sorted(
|
||
{
|
||
value
|
||
for value in hashes
|
||
if not value.startswith(KNOWN_HASH_PREFIXES)
|
||
}
|
||
)[:5]
|
||
return report
|
||
|
||
|
||
async def main() -> None:
|
||
report = await collect()
|
||
OUTPUT.parent.mkdir(parents=True, exist_ok=True)
|
||
OUTPUT.write_text(
|
||
json.dumps(report, ensure_ascii=False, indent=2, default=str),
|
||
encoding="utf-8",
|
||
)
|
||
print(f"wrote {OUTPUT}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|