tmp: kb_diag2诊断Bearer认证链

wwwroot/api/kb_diag2.dspy

@@ -0,0 +1,48 @@
+# kb_diag2.dspy - 深挖 Bearer key 401：dapi 查询匹配 + 注册表 + rbac 认证链
+import json as _j
+
+uid = await get_user()
+if not uid:
+    return _j.dumps({"ok": False, "error": "未登录"}, ensure_ascii=False)
+
+out = {}
+try:
+    from pipeline_service.rag_client import resolve_project_owner, get_owner_apikey
+    pid = "d9jE6l8vx-yHO6-_pIWnY"
+    owner, _e = await resolve_project_owner(pid)
+    key, err = await get_owner_apikey(owner)
+    out["key_len"] = len(key or "")
+    out["key_prefix"] = (key or "")[:10]
+
+    dbname = get_module_dbname('pipeline-sdlc')
+    async with DBPools().sqlorContext(dbname) as sor:
+        # 1. 库里 downapikey 存储值
+        recs = await sor.sqlExe(
+            "SELECT id, dappid, apikey FROM downapikey WHERE userid=${u}$", {"u": owner})
+        await sor.sqlExe("COMMIT", {})
+        out["stored"] = [{"id": r.id, "dappid": r.dappid, "enc_len": len(r.apikey or ""),
+                          "enc_prefix": (r.apikey or "")[:10]} for r in (recs or [])]
+
+        # 2. password_encode 对称性：编码明文 key 是否 == 存储值
+        from ahserver.serverenv import get_serverenv
+        enc_fn = get_serverenv('password_encode')
+        encoded = enc_fn(key) if key else ""
+        stored_enc = (recs[0].apikey if recs else "") or ""
+        out["encode_symmetric"] = (encoded == stored_enc)
+        out["encoded_len"] = len(encoded)
+
+        # 3. dapi 的 get_apikey_user 能否查到（传入明文，内部会 encode 再查）
+        from dapi.dapi import get_apikey_user
+        u = await get_apikey_user(sor, key, "127.0.0.1")
+        out["get_apikey_user"] = str(getattr(u, 'id', None)) if u else "NONE"
+
+        # 4. register_auth_method 注册表
+        from rbac import check_perm
+        for an in ('AUTH_METHODS', 'auth_methods', '_auth_methods', 'AUTHS', '_AUTH_METHODS'):
+            if hasattr(check_perm, an):
+                v = getattr(check_perm, an)
+                out["auth_methods_" + an] = list(v.keys()) if isinstance(v, dict) else str(type(v))
+except Exception as e:
+    import traceback
+    out["traceback"] = traceback.format_exc()[-700:]
+return _j.dumps(out, ensure_ascii=False, default=str)