Merge origin/ai-11-2 into order-9-22

Integrate group password sign-in and live authorization synchronization while
retaining user editing and the restored knowledge review workspace.

Keep local database snapshots and dependency cache trees from the target branch.
Complete English translations and resolve test dependencies through their
owning workspace packages so the new tests work with pnpm's strict layout.
This commit is contained in:
gr
2026-11-04 10:27:26 +08:00
27 changed files with 2588 additions and 63 deletions
+19 -1
View File
@@ -4,6 +4,20 @@
已有有效 zyt 后台账号可从集团工作台一键进入客服。首次创建固定“集团基础用户”角色,没有业务权限;客服管理员使用原本地账号在“系统管理 → 用户管理”分配角色,在“角色权限”勾选具体能力。再次集团登录保留管理员分配的角色和本地停用状态,不按同名账号合并,不继承医院 root。密码仍由 zyt 验证,绑定账号不能使用或设置客服本地密码。
网页登录使用同一个账号密码表单:输入zyt账号密码或客服本地账号密码,点击“登录”直接进入客服,不跳转集团后台。本地凭据验证成功时沿用本地账号和角色;否则通过服务器配置的ZYT API实时验密。无医院菜单权限的有效zyt账号也可使用。
直接登录仅需客服API配置`ZYT_GROUP_API_URL`。中央端点`POST /adminapi/group/kefu/login`调用原项目无会话验密`POST /adminapi/identity.session/verify`,不创建或注销zyt网页会话,不依赖集团前端运行。密码仅用于本次验证,不在客服保存为本地密码、放入URL或返回给浏览器;密码仍由zyt管理。
同名账号不会合并:有效本地凭据优先;只有zyt凭据匹配时,按提供方和稳定GID创建/查找独立绑定账号,首次为无业务权限的“集团基础用户”。原客服管理员在用户管理分配本地业务角色,再次直接登录保持该角色与本地停用状态。不能在客服重置绑定账号的zyt密码。
管理端标准API前缀为`/api/v2`,构建产物应设置`VITE_GLOB_API_URL=/api/v2`,可以由8766服务直接托管;开发代理及本机5666预览同时兼容旧`/api`与标准前缀。如果通过其他地址访问,集团登录会先进入已配置的客服回调地址,再发起登录,避免不同域或端口丢失登录事务。回调仍严格固定,不接受任意跳转。
本地管理员在用户列表“来源/原账号”找到集团用户,选择业务角色即可;“集团基础用户”固定无权限,只读不可扩权。后端每请求使用新权限,已登录前端在页面可见时每3秒同步菜单和按钮,切回窗口立即同步。仅有角色读取权限的角色管理员也能进入角色页;分配用户角色仍需用户管理权限,不因登录成功自动授予管理能力。
集团管理员可查看这里分配的角色和有效权限、同步动态新增的权限目录,并直接编辑已绑定用户的客服授权。客服API配置`ZYT_GROUP_METADATA_KEY_FILE`,集团配置`IAM_KEFU_METADATA_API_URL`及`IAM_KEFU_METADATA_KEY_FILE`,两个文件内容为同一份32字节随机密钥的64位hex表示。文件须放在非公开目录、限制服务用户读取,勿写入浏览器、URL或仓库。Docker需要额外挂载该文件,配置值是容器内只读路径;只有api服务需要它。该密钥同时用于专用授权管理协议,必须仅提供给受信任的集团服务。
元数据接口`POST /api/v2/group/access-snapshot`只接受专用HMAC请求并签名响应,不接受普通客服用户token代替机器认证,也不能登录或改授权。按稳定GID关联,查询不会创建账号或认领同名本地账号。集团用户页“客服授权”展示当前角色、启停和有效权限,打开期间每30秒刷新;集团菜单管理的客服公司目录自动拉取,也可点击“同步客服权限”。自定义权限增改删会反映到集团管理目录,删除采用软停用并保留既有引用,恢复沿用原节点。不会把客服权限并入集团/医院授权,也不会自动给集团角色加权限。
客服 API 配置以下公开地址,缺少任一项时关闭集团登录:
```dotenv
@@ -12,8 +26,12 @@ ZYT_GROUP_API_URL=https://<集团API域名>/adminapi
KEFU_PUBLIC_BASE_URL=https://<客服前端域名>
```
同时在集团配置 `IAM_KEFU_BASE_URL=<客服前端地址>`、`IAM_KEFU_COMPANY_CODE=<实际客服公司编码>`。回调严格为客服前端根路径 `/auth/group-login`,使用 history 路由。生产 HTTPS;只有回环本地开发可用 HTTP。Docker compose 已透传三项配置,不能把密钥或密码放进这些地址。
同时在集团配置 `IAM_KEFU_BASE_URL=<已登记客服地址>`、`IAM_KEFU_COMPANY_CODE=<实际客服公司编码>`以启用该客户端。上面的前端/回调配置仅为保留集团工作台旧单点入口所需;客服主登录框直接验密只依赖API地址。旧回调仍严格为`/auth/group-login`。生产HTTPS;只有回环本地开发可用HTTP。Docker compose已透传配置,不能把密钥或密码放进这些地址。
集团菜单目录仅控制集团入口显示,客服业务菜单和按钮由客服本地角色决定。集团基础角色固定为空,需给用户分配其他业务角色,而不是给基础角色扩权。客服 SSO 会话最长 8 小时;本地停用和权限调整每请求生效。当前无跨系统全局注销,集团改密/退出会阻止新的 SSO,但不会声称立即撤销所有已签发客服会话。
集团管理员也可在集团“用户管理 → 客服授权 → 编辑客服授权”中直接分配客服角色,或勾选仅该用户的权限。独立权限通过专属客服角色保存,不修改共享角色或其他用户;实际授权仍存于客服。保存提交后,既有客服会话下一次请求即按新权限执行。2026-11-04起前台页面每3秒同步菜单、按钮和路由,切回窗口或恢复可见立即同步;当前页面失权后返回账号页,不注销登录,仍有权页面及其草稿保留。前端升级首次加载新版本需要刷新一次,后续授权变化无需整页刷新。未绑定用户须先用ZYT账号登录客服,不能按同名本地账号认领。
此管理使用独立的`/api/v2/group/access-options`、`/api/v2/group/access-update`端点,沿用服务密钥但绑定实际端点签名,持久化防重放并校验旧版本。普通客服token不能调用。集团仅允许显式集团管理员,并在每次操作核对来源用户/GID/应用归属;两端审计。原`access-snapshot`仍只读。联调协议见集团`integration/kefu-group-authorization-contract.md`。
新增回归:`wechat_rpa/test_group_sso.py` 与 `admin-web/apps/web-antd/tests/group-sso.test.mjs`。后端测试需要在合成配置及隔离临时目录运行,勿初始化或重置真实账号库。集团侧跨项目联合测试见 `C:/zyt/identity-admin/tests/kefu-joint-flow.test.mjs`。
@@ -36,6 +36,7 @@ export namespace AuthApi {
}
}
/** ZYT 与客服本地账号共用密码登录;凭证只放 POST body,由服务端选择验证来源。 */
export async function loginApi(data: AuthApi.LoginParams) {
const resp = await requestClient.post<AuthApi.BackendLoginResult>(
'/auth/login',
+18 -9
View File
@@ -17,6 +17,7 @@ import { message } from 'ant-design-vue';
import { t } from '#/locales/t';
import { useAuthStore } from '#/store';
import { requestBelongsToSession } from '#/utils/authorization-sync';
import { refreshTokenApi } from './core';
import { responseErrorMessage } from './error-message';
@@ -109,15 +110,23 @@ function createRequestClient(baseURL: string, options?: RequestClientOptions) {
);
// token过期的处理
client.addResponseInterceptor(
authenticateResponseInterceptor({
client,
doReAuthenticate,
doRefreshToken,
enableRefreshToken: preferences.app.enableRefreshToken,
formatToken,
}),
);
const authentication = authenticateResponseInterceptor({
client,
doReAuthenticate,
doRefreshToken,
enableRefreshToken: preferences.app.enableRefreshToken,
formatToken,
});
client.addResponseInterceptor({
...authentication,
rejected: (error) => {
// A delayed 401 from a previous account must not log out the new account.
if (!requestBelongsToSession(error?.config?.headers?.Authorization, useAccessStore().accessToken)) {
return Promise.reject(error);
}
return authentication.rejected?.(error);
},
});
// 通用的错误处理,如果没有进入上面的错误处理逻辑,就会进入这里
client.addResponseInterceptor(
+2
View File
@@ -17,6 +17,7 @@ import { initComponentAdapter } from './adapter/component';
import { initSetupVbenForm } from './adapter/form';
import App from './app.vue';
import { router } from './router';
import { installAuthorizationSync } from './router/authorization-sync';
async function bootstrap(namespace: string) {
// 初始化组件适配器
@@ -57,6 +58,7 @@ async function bootstrap(namespace: string) {
// 配置路由及路由守卫
app.use(router);
app.onUnmount(installAuthorizationSync(router));
// 配置Motion插件
const { MotionPlugin } = await import('@vben/plugins/motion');
@@ -6,6 +6,9 @@
* 管理员自己加的条目(custom)没有译文,原样显示。
*/
export default {
'已保存,后端权限立即生效;在线用户的菜单和按钮将在数秒内自动更新。': 'Saved. Server permissions take effect immediately; menus and buttons update automatically within a few seconds for online users.',
'后端权限修改后立即生效,无需重启或重新登录;在线用户的菜单和按钮将在数秒内自动更新,切回窗口时立即同步。管理员和集团基础用户角色只读,内置角色和仍有人在用的角色不可删除。要新增权限点或左侧菜单链接,去「系统管理 → 权限菜单」。': 'Server permission changes take effect immediately without restarting or signing in again. Menus and buttons update within a few seconds for online users and sync immediately when the window regains focus. Administrator and group basic roles are read-only. Built-in roles and roles assigned to users cannot be deleted. Add permissions or navigation links under System → Permission menus.',
'集团基础用户固定无业务权限,不能修改。请到「系统管理 → 用户管理」为账号分配其他业务角色。': 'The group basic role has no business permissions and cannot be changed. Assign another business role to the account under System → Users.',
// ── 权限菜单页 permissions.vue ──────────────────────────────────────────────
'已新增 {code}。管理员自动拥有它;其他角色去「角色权限」里勾选':
'Added {code}. Administrators get it automatically; grant it to other roles under "Roles"',
@@ -1,5 +1,9 @@
/** views/console/users.vue */
export default {
'ZYT 账号或客服本地账号': 'ZYT account or local service account',
'请输入账号': 'Enter your account',
'ZYT 密码或客服本地密码': 'ZYT password or local service password',
'ZYT 首次登录为基础用户,业务权限由客服管理员分配。': 'First-time ZYT users start with basic access. Service administrators assign business permissions.',
'登录响应无效,请重新登录': 'Invalid login response. Please sign in again',
'集团登录地址配置无效': 'Invalid group sign-in URL configuration',
'集团登录尚未启用': 'Group sign-in is not enabled',
@@ -0,0 +1,133 @@
import type { UserInfo } from '@vben/types';
import type { RouteRecordRaw, Router } from 'vue-router';
import { watch } from 'vue';
import { createMemoryHistory, createRouter } from 'vue-router';
import { getTabKey, useAccessStore, useTabbarStore, useUserStore } from '@vben/stores';
import { getUserInfoApi } from '#/api/core/user';
import { createAuthorizationSync } from '#/utils/authorization-sync';
import { generateAccess } from './access';
import { accessRoutes, routes } from './routes';
function copyRoutes(input: RouteRecordRaw[]): RouteRecordRaw[] {
return input.map((route) => ({ ...route, meta: { ...route.meta },
children: route.children ? copyRoutes(route.children) : undefined } as RouteRecordRaw));
}
// Vben adds dynamic children into the normalized Root. Keep a clean template,
// including component functions, rather than cloning that mutated tree later.
const staticRoutes = copyRoutes(routes).map((route) => route.path === '/' ? { ...route, children: [] } : route);
const installations = new WeakMap<Router, () => void>();
function routeNames(input: RouteRecordRaw[]): Set<string | symbol> {
return new Set(input.flatMap((route) => [
...(route.name ? [route.name] : []), ...routeNames(route.children ?? []),
]));
}
export function isAuthorizedPath(router: Router, path: string, codes: string[]) {
const target = router.resolve(path);
return target.matched.length > 0 && target.name !== 'FallbackNotFound' && target.matched.every((record) => {
const authority = record.meta.authority;
return !Array.isArray(authority) || authority.length === 0 || authority.some((code) => codes.includes(code));
});
}
/** Reconcile by name/path, retaining authorized matcher/component instances. */
export function reconcileAccessRoutes(router: Router, nextRoutes: RouteRecordRaw[]) {
const staticNames = routeNames(staticRoutes);
const desired = routeNames(nextRoutes);
const root = router.getRoutes().find((route) => route.path === '/');
if (!root?.name) throw new Error('Missing application root route');
for (const record of router.getRoutes()) {
if (record.name && !staticNames.has(record.name) && !desired.has(record.name)) router.removeRoute(record.name);
}
function install(input: RouteRecordRaw[], parent: string | symbol) {
for (const route of input) {
const parentPath = router.getRoutes().find((record) => record.name === parent)?.path ?? '/';
const expectedPath = route.path.startsWith('/') ? route.path : `${parentPath.replace(/\/$/, '')}/${route.path}`;
let existing = route.name ? router.getRoutes().find((record) => record.name === route.name) : undefined;
if (existing && existing.path !== expectedPath) {
router.removeRoute(existing.name!);
existing = undefined;
}
if (existing) {
// Existing authorized pages keep their resolved components and drafts.
// Recurse so a grant adds missing siblings without replacing their parent.
Object.assign(existing.meta, route.meta);
install(route.children ?? [], existing.name!);
} else if (route.meta?.noBasicLayout) {
router.addRoute(route);
} else {
router.addRoute(parent, route);
}
}
}
install(nextRoutes, root.name);
// Keep Vben's source tree exact as well as Vue Router's matcher table.
root.children.splice(0, root.children.length, ...nextRoutes.filter((route) => !route.meta?.noBasicLayout));
}
export function pruneAuthorizationTabs(router: Router, codes: string[], tabbar: ReturnType<typeof useTabbarStore>) {
tabbar.tabs = tabbar.tabs.filter((tab) => isAuthorizedPath(router, tab.fullPath || tab.path, codes));
tabbar.visitHistory.retain(tabbar.tabs.map((tab) => tab.key || getTabKey(tab)));
for (const [key, cached] of tabbar.cachedRoutes) {
if (!isAuthorizedPath(router, cached.route.fullPath || cached.route.path, codes)) tabbar.removeCachedRoute(key);
}
for (const name of tabbar.excludeCachedTabs) {
if (!router.hasRoute(name)) tabbar.excludeCachedTabs.delete(name);
}
void tabbar.updateCacheTabs();
}
/** Install once for this app, after Pinia and Router are available. */
export function installAuthorizationSync(router: Router) {
installations.get(router)?.();
const access = useAccessStore();
const user = useUserStore();
const tabbar = useTabbarStore();
const sync = createAuthorizationSync({
window, document,
fetch: getUserInfoApi,
getCodes: () => access.accessCodes,
getSession: () => access.accessToken && access.isAccessChecked && user.userInfo?.userId
? { token: access.accessToken, userId: String(user.userInfo.userId) } : undefined,
async prepare(info: UserInfo) {
const preparedRouter = createRouter({ history: createMemoryHistory(), routes: copyRoutes(staticRoutes) });
return generateAccess({ roles: info.roles ?? [], router: preparedRouter, routes: accessRoutes });
},
apply(info, snapshot) {
const codes = info.roles ?? [];
user.setUserInfo(info);
access.setAccessCodes(codes);
if (!snapshot) return;
reconcileAccessRoutes(router, snapshot.accessibleRoutes);
access.setAccessMenus(snapshot.accessibleMenus);
access.setAccessRoutes(snapshot.accessibleRoutes);
pruneAuthorizationTabs(router, codes, tabbar);
// Keep an authorized current component mounted; replacing its route would
// discard unsaved forms merely because a different permission changed.
if (!isAuthorizedPath(router, router.currentRoute.value.fullPath, codes)) {
return router.replace('/group/welcome').then(() => {});
}
},
});
const unwatch = watch(() => JSON.stringify([
access.accessToken, user.userInfo?.userId, access.isAccessChecked,
]), () => {
sync.stop();
if (access.accessToken && user.userInfo?.userId && access.isAccessChecked) sync.start();
}, { immediate: true, flush: 'sync' });
const dispose = () => {
unwatch();
sync.stop();
if (installations.get(router) === dispose) installations.delete(router);
};
installations.set(router, dispose);
return dispose;
}
@@ -40,7 +40,6 @@ const routes: RouteRecordRaw[] = [
icon: 'lucide:bot',
order: 0,
title: 'AI 模型',
authority: ['model:read', 'agent:read', 'stats:read'],
},
name: 'ModelCenter',
path: '/model',
@@ -94,7 +93,6 @@ const routes: RouteRecordRaw[] = [
icon: 'lucide:monitor-smartphone',
order: 1,
title: '桌面端',
authority: ['desktop-account:read', 'config:read'],
},
name: 'DesktopCenter',
path: '/desktop',
@@ -138,7 +136,6 @@ const routes: RouteRecordRaw[] = [
icon: 'lucide:messages-square',
order: 2,
title: '聊天归档',
authority: ['im:read', 'im:content:read', 'im:export', 'im:identity:write', 'im:storage:write'],
},
name: 'ArchiveCenter',
path: '/archive',
@@ -200,7 +197,6 @@ const routes: RouteRecordRaw[] = [
icon: 'lucide:shield',
order: 3,
title: '系统管理',
authority: ['user:read', 'audit:read'],
},
name: 'SystemCenter',
path: '/system',
@@ -212,7 +208,7 @@ const routes: RouteRecordRaw[] = [
meta: {
icon: 'lucide:key-round',
title: '角色权限',
authority: ['role:read', 'role:write'],
authority: ['role:read', 'user:read'],
},
},
{
@@ -262,4 +258,12 @@ const routes: RouteRecordRaw[] = [
},
];
// A business directory is visible when at least one of its pages is granted.
// Derive its codes so newly split page permissions cannot be hidden by stale parents.
for (const route of routes) {
if (route.meta && route.children?.length && route.children.every(child => child.meta?.authority?.length)) {
route.meta.authority = [...new Set(route.children.flatMap(child => child.meta?.authority || []))];
}
}
export default routes;
@@ -0,0 +1,91 @@
interface AuthorizationInfo {
roles?: string[];
userId: string;
}
interface Session {
token: string;
userId: string;
}
interface SyncOptions<Info extends AuthorizationInfo, Snapshot> {
apply: (info: Info, snapshot?: Snapshot) => Promise<void> | void;
document: Pick<Document, 'addEventListener' | 'removeEventListener' | 'visibilityState'>;
fetch: () => Promise<Info>;
getCodes: () => string[];
getSession: () => Session | undefined;
interval?: number;
prepare: (info: Info) => Promise<Snapshot>;
window: Pick<Window, 'addEventListener' | 'clearInterval' | 'removeEventListener' | 'setInterval'>;
}
function permissionKey(codes: string[]) {
return JSON.stringify([...new Set(codes)].sort());
}
/** Ignore failures from a request made by a different, already-ended session. */
export function requestBelongsToSession(authorization: unknown, token: null | string) {
return (authorization || null) === (token ? `Bearer ${token}` : null);
}
/** One foreground poller. Preparation never mutates the displayed router/stores. */
export function createAuthorizationSync<Info extends AuthorizationInfo, Snapshot>(options: SyncOptions<Info, Snapshot>) {
let active = false;
let epoch = 0;
let timer: number | undefined;
let pending: Promise<void> | undefined;
let appliedPermissions = permissionKey(options.getCodes());
function refresh(): Promise<void> {
if (!active || options.document.visibilityState !== 'visible') return Promise.resolve();
if (pending) return pending;
const session = options.getSession();
if (!session) return Promise.resolve();
const generation = epoch;
const current = () => {
const now = options.getSession();
return active && epoch === generation && now?.token === session.token && now.userId === session.userId;
};
const task = (async () => {
try {
const info = await options.fetch();
if (!current() || String(info.userId) !== session.userId) return;
const changed = permissionKey(info.roles ?? []) !== appliedPermissions;
const snapshot = changed ? await options.prepare(info) : undefined;
if (!current()) return;
await options.apply(info, snapshot);
if (current()) appliedPermissions = permissionKey(info.roles ?? []);
} catch {
// requestClient owns HTTP/401 handling. A later foreground poll can recover;
// failed background reads must never clear grants or replay a write.
}
})();
pending = task;
void task.finally(() => { if (pending === task) pending = undefined; });
return task;
}
const trigger = () => { void refresh(); };
function stop() {
active = false;
epoch += 1;
pending = undefined;
if (timer !== undefined) options.window.clearInterval(timer);
timer = undefined;
options.window.removeEventListener('focus', trigger);
options.document.removeEventListener('visibilitychange', trigger);
}
function start() {
if (active) return;
active = true;
epoch += 1;
appliedPermissions = permissionKey(options.getCodes());
options.window.addEventListener('focus', trigger);
options.document.addEventListener('visibilitychange', trigger);
timer = options.window.setInterval(trigger, options.interval ?? 3000);
trigger();
}
return { refresh, start, stop };
}
@@ -42,6 +42,15 @@ export function validateGroupConfig(config: GroupConfig, origin: string): GroupC
return config as GroupConfig & Required<Omit<GroupConfig, 'enabled'>>;
}
/** Reach the configured callback origin before storing its tab-bound PKCE proof. */
export function groupLoginEntry(config: GroupConfig, origin: string): string {
const callback = safeUrl(config?.redirect_uri);
validateGroupConfig(config, callback.origin);
const current = safeUrl(origin);
if (current.href !== `${current.origin}/`) throw new Error(t('集团登录地址配置无效'));
return callback.origin === current.origin ? '/auth/group-login' : callback.href;
}
function base64url(value: Uint8Array): string {
return btoa(String.fromCharCode(...value)).replaceAll('+', '-').replaceAll('/', '_').replaceAll('=', '');
}
@@ -6,7 +6,7 @@ import { Alert, Button, Spin } from 'ant-design-vue';
import { getGroupConfigApi } from '#/api';
import { useAuthStore } from '#/store';
import { beginGroupLogin, consumeGroupCallback, GROUP_TRANSACTION_KEY } from '#/utils/group-sso';
import { beginGroupLogin, consumeGroupCallback, groupLoginEntry, GROUP_TRANSACTION_KEY } from '#/utils/group-sso';
const error = ref('');
const router = useRouter();
@@ -24,7 +24,13 @@ onMounted(async () => {
await auth.authGroupLogin(transaction);
} else {
sessionStorage.removeItem(GROUP_TRANSACTION_KEY);
const target = await beginGroupLogin(await getGroupConfigApi());
const config = await getGroupConfigApi();
const entry = groupLoginEntry(config, window.location.origin);
if (entry !== '/auth/group-login') {
window.location.replace(entry);
return;
}
const target = await beginGroupLogin(config);
window.location.assign(target);
}
} catch {
@@ -1,14 +1,9 @@
<script lang="ts" setup>
import type { VbenFormSchema } from '@vben/common-ui';
import { computed, onMounted, ref } from 'vue';
import { useRouter } from 'vue-router';
import { computed } from 'vue';
import { AuthenticationLogin, z } from '@vben/common-ui';
import { Button } from 'ant-design-vue';
import { getGroupConfigApi } from '#/api';
import { validateGroupConfig } from '#/utils/group-sso';
import { t } from '#/locales/t';
import { useAuthStore } from '#/store';
@@ -16,14 +11,6 @@ import { useAuthStore } from '#/store';
defineOptions({ name: 'Login' });
const authStore = useAuthStore();
const router = useRouter();
const groupEnabled = ref(false);
onMounted(async () => {
try {
validateGroupConfig(await getGroupConfigApi(), window.location.origin);
groupEnabled.value = true;
} catch { groupEnabled.value = false; }
});
// 模板自带的演示账号下拉、滑块验证码、手机号/扫码/第三方登录、注册和找回密码
// 都删掉了——后端没有对应的接口,留着只会让人点进死路。滑块验证码尤其要注意:
@@ -31,14 +18,14 @@ onMounted(async () => {
const formSchema = computed((): VbenFormSchema[] => [
{
component: 'VbenInput',
componentProps: { placeholder: t('用户名') },
componentProps: { placeholder: t('ZYT 账号或客服本地账号') },
fieldName: 'username',
label: t('用户名'),
rules: z.string().min(1, { message: t('请输入用户名') }),
label: t('账号'),
rules: z.string().min(1, { message: t('请输入账号') }),
},
{
component: 'VbenInputPassword',
componentProps: { placeholder: t('密码') },
componentProps: { placeholder: t('ZYT 密码或客服本地密码') },
fieldName: 'password',
label: t('密码'),
rules: z.string().min(1, { message: t('请输入密码') }),
@@ -54,15 +41,10 @@ const formSchema = computed((): VbenFormSchema[] => [
:show-forget-password="false"
:show-qrcode-login="false"
:show-register="false"
:show-remember-me="false"
:show-third-party-login="false"
:sub-title="t('请使用管理员分配的账号登录')"
:sub-title="t('ZYT 首次登录为基础用户,业务权限由客服管理员分配。')"
:title="t('真羊 AI 客服 · 管理后台')"
@submit="authStore.authLogin"
>
<template #third-party-login>
<Button v-if="groupEnabled" block class="mt-4" @click="router.push('/auth/group-login')">
{{ t('使用集团账号登录') }}
</Button>
</template>
</AuthenticationLogin>
/>
</template>
@@ -27,8 +27,9 @@ const open = ref(false);
const isNew = ref(false);
const form = reactive({ code: '', name: '', permissions: [] as string[] });
const protectedRole = (code: string) => code === 'admin' || code === 'zyt_basic';
// 新增和修改是两个按钮:只能新增的人打开已有角色只能看
const readonly = computed(() => !(isNew.value ? canCreate.value : canWrite.value) || form.code === 'admin');
const readonly = computed(() => !(isNew.value ? canCreate.value : canWrite.value) || protectedRole(form.code));
/** 一级菜单 → 页面 → 权限码,和左侧菜单一一对应。 */
const grouped = computed(() => groupPermissions(permissions.value));
@@ -49,6 +50,7 @@ function expandAll() {
}
function onCheck(keys: unknown) {
if (readonly.value) return;
// 不是严格模式时 antd 给的是数组;兜底处理 { checked } 形状
const list = Array.isArray(keys) ? keys : ((keys as { checked?: (number | string)[] }).checked ?? []);
form.permissions = applyTreeCheck(form.permissions, list, visibleCodes.value, permissions.value);
@@ -98,10 +100,11 @@ async function submit() {
const known = new Set(permissions.value.map((item) => item.code));
roles.value = (await saveRole({ ...form, permissions: form.permissions.filter((code) => known.has(code)) })).roles;
open.value = false;
message.success(t('已保存,权限立即生效(无需重启,也无需重新登录)'));
message.success(t('已保存,后端权限立即生效;在线用户的菜单和按钮将在数秒内自动更新。'));
}
async function remove(row: RoleItem) {
if (!canDelete.value || protectedRole(row.code) || row.builtin || row.user_count > 0) return;
roles.value = (await deleteRole(row.code)).roles;
message.success(t('已删除'));
}
@@ -125,7 +128,7 @@ onMounted(load);
type="info"
show-icon
:message="t('权限按左侧菜单组织:菜单 → 页面 → 页面里的每个按钮')"
:description="t('改完立刻生效——不用重启服务,用户也不用重新登录。管理员角色不可削权(削了就没人能加回来),内置角色和仍有人在用的角色不可删除。要新增权限点或左侧菜单链接,去「系统管理 → 权限菜单」。')"
:description="t('后端权限修改后立即生效,无需重启或重新登录;在线用户的菜单和按钮将在数秒内自动更新,切回窗口时立即同步。管理员和集团基础用户角色只读,内置角色和仍有人在用的角色不可删除。要新增权限点或左侧菜单链接,去「系统管理 → 权限菜单」。')"
/>
<Card :title="t('角色与权限')">
<template #extra>
@@ -158,12 +161,12 @@ onMounted(load);
</template>
<template v-else-if="column.key === 'action'">
<Button type="link" size="small" @click="openEdit(record as RoleItem)">
{{ canWrite && record.code !== 'admin' ? t('编辑') : t('查看') }}
{{ canWrite && !protectedRole(record.code) ? t('编辑') : t('查看') }}
</Button>
<Popconfirm :title="t('确定删除这个角色?')" @confirm="remove(record as RoleItem)">
<Button
type="link" size="small" danger
:disabled="!canDelete || record.builtin || record.user_count > 0"
:disabled="!canDelete || protectedRole(record.code) || record.builtin || record.user_count > 0"
>
{{ t('删除') }}
</Button>
@@ -190,6 +193,9 @@ onMounted(load);
<p v-if="form.code === 'admin'" class="mb-2 text-xs text-gray-500">
{{ t('管理员始终拥有全部权限(包括以后新增的),不能修改。') }}
</p>
<p v-else-if="form.code === 'zyt_basic'" class="mb-2 text-xs text-gray-500">
{{ t('集团基础用户固定无业务权限,不能修改。请到「系统管理 → 用户管理」为账号分配其他业务角色。') }}
</p>
<div class="mb-2 flex flex-wrap items-center gap-2">
<Input v-model:value="keyword" allow-clear size="small" class="w-60" :placeholder="t('搜索按钮名称或权限码')"
@change="expandAll" />
@@ -0,0 +1,21 @@
import assert from 'node:assert/strict';
import {readFileSync} from 'node:fs';
import {createRequire} from 'node:module';
import vm from 'node:vm';
import test from 'node:test';
const require=createRequire(import.meta.url);
const ts=require(process.env.KEFU_SSO_TYPESCRIPT||'typescript');
test('实际Vite代理同时接受旧前缀与标准API前缀,保留独立后端配置',async()=>{
const source=readFileSync(new URL('../vite.config.ts',import.meta.url),'utf8');
const code=ts.transpileModule(source,{compilerOptions:{module:ts.ModuleKind.CommonJS}}).outputText;
for(const target of [undefined,'http://127.0.0.1:18766']) {
const module={exports:{}};
vm.runInNewContext(code,{module,exports:module.exports,process:{env:{VITE_API_TARGET:target}},
require(name){assert.equal(name,'@vben/vite-config');return {defineConfig:factory=>factory};}});
const proxy=(await module.exports.default()).vite.server.proxy['/api'];
assert.equal(proxy.rewrite('/api/auth/login'),'/api/v2/auth/login');
assert.equal(proxy.rewrite('/api/v2/auth/group/config'),'/api/v2/auth/group/config');
assert.equal(proxy.rewrite('/api/v2/users?active=true'),'/api/v2/users?active=true');
assert.equal(proxy.target,target||'http://127.0.0.1:8766');
}
});
@@ -0,0 +1,328 @@
import assert from 'node:assert/strict';
import { readFile } from 'node:fs/promises';
import { createRequire } from 'node:module';
import { resolve } from 'node:path';
import test from 'node:test';
import { fileURLToPath, pathToFileURL } from 'node:url';
import vm from 'node:vm';
const require = createRequire(process.env.KEFU_SSO_TYPESCRIPT
? pathToFileURL(resolve(process.env.KEFU_SSO_TYPESCRIPT, 'package.json'))
: new URL('../package.json', import.meta.url));
const ts = require('typescript');
const vue = require('vue');
const pinia = require('pinia');
const routing = require('vue-router');
const workspace = process.env.KEFU_SSO_WORKSPACE || fileURLToPath(new URL('../../../', import.meta.url));
const source = relative => readFile(resolve(workspace, relative), 'utf8');
const staged = relative => readFile(new URL(`../src/${relative}`, import.meta.url), 'utf8');
const normalized = value => JSON.parse(JSON.stringify(value));
const clone = value => Array.isArray(value) ? value.map(clone)
: value && typeof value === 'object' ? Object.fromEntries(Object.entries(value).map(([key, item]) => [key, clone(item)])) : value;
function load(code, dependencies = {}, globals = {}) {
const compiled = ts.transpileModule(code.replaceAll('import.meta.hot', 'undefined'), { compilerOptions: {
module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2022,
}, reportDiagnostics: true });
assert.deepEqual(compiled.diagnostics, []);
const module = { exports: {} };
vm.runInNewContext(compiled.outputText, { module, exports: module.exports,
require(name) { assert.ok(name in dependencies, `Unexpected import ${name}`); return dependencies[name]; },
console, ...globals });
return module.exports;
}
const preferences = { app: { accessMode: 'frontend', loginExpiredMode: 'modal', enableRefreshToken: false },
tabbar: { visitHistory: true, maxCount: 0 } };
const tree = load(await source('packages/@core/base/shared/src/utils/tree.ts'));
const stack = load(await source('packages/@core/base/shared/src/utils/stack.ts'));
const { useAccessStore } = load(await source('packages/stores/src/modules/access.ts'), { pinia });
const { useUserStore } = load(await source('packages/stores/src/modules/user.ts'), { pinia });
const { useTabbarStore, getTabKey } = load(await source('packages/stores/src/modules/tabbar.ts'), {
vue, pinia, '@vben-core/preferences': { preferences },
'@vben-core/shared/utils': { ...stack, startProgress() {}, stopProgress() {}, openWindow() {} },
}, { sessionStorage: {} });
const stores = { useAccessStore, useUserStore, useTabbarStore, getTabKey };
const { useAccess } = load(await source('packages/effects/access/src/use-access.ts'), {
vue, '@vben/preferences': { preferences }, '@vben/stores': stores,
});
const { generateRoutesByFrontend } = load(await source('packages/utils/src/helpers/generate-routes-frontend.ts'), {
'@vben-core/shared/utils': tree,
});
const { generateMenus } = load(await source('packages/utils/src/helpers/generate-menus.ts'), {
'@vben-core/shared/utils': tree,
});
const { generateAccessible } = load(await source('packages/effects/access/src/accessible.ts'), {
vue, '@vben/utils': { ...tree, cloneDeep: clone, generateMenus, generateRoutesByFrontend,
generateRoutesByBackend() { throw new Error('Unexpected backend access mode'); },
isFunction: value => typeof value === 'function', isString: value => typeof value === 'string' },
});
const syncModule = load(await staged('utils/authorization-sync.ts'));
const menuHelpers = load(await source('apps/web-antd/src/router/menu-access.ts'));
const originalRoutes = load(await staged('router/routes/modules/console.ts')).default;
function pageRoutes(input) {
return input.map(route => ({ ...route, meta: { ...route.meta },
...(route.component ? { component: { name: `Test${route.name}`, render: () => null } } : {}),
children: route.children ? pageRoutes(route.children) : undefined }));
}
const accessRoutes = pageRoutes(originalRoutes);
const routes = [
{ name: 'Root', path: '/', component: { render: () => vue.h(routing.RouterView) }, children: [] },
{ name: 'Login', path: '/auth/login', component: { render: () => null } },
{ name: 'FallbackNotFound', path: '/:path(.*)*', component: { render: () => null } },
];
function environment() {
const window = new EventTarget();
const document = new EventTarget();
document.visibilityState = 'visible';
const timers = new Map();
let sequence = 0;
window.setInterval = (callback, delay) => { const id = ++sequence; timers.set(id, { callback, delay }); return id; };
window.clearInterval = id => timers.delete(id);
return { window, document, timers,
tick() { for (const timer of [...timers.values()]) timer.callback(); },
focus() { window.dispatchEvent(new Event('focus')); },
visible(value) { document.visibilityState = value ? 'visible' : 'hidden'; document.dispatchEvent(new Event('visibilitychange')); } };
}
const settle = () => new Promise(resolve => setImmediate(resolve));
const info = (roles = [], userId = '7') => ({ userId, username: `user${userId}`, realName: 'Test', roles });
const deferred = () => { let resolve, reject; const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); return { promise, resolve, reject }; };
async function harness(initial = []) {
pinia.setActivePinia(pinia.createPinia());
const access = useAccessStore();
const user = useUserStore();
const tabbar = useTabbarStore();
const env = environment();
const router = routing.createRouter({ history: routing.createMemoryHistory(), routes: clone(routes) });
const requests = [];
let answer = info(initial);
let preparations = 0;
const generateAccess = async options => {
preparations += 1;
const generated = await generateAccessible('frontend', options);
return { ...generated, accessibleMenus: menuHelpers.pruneEmptyMenus(generated.accessibleMenus, menuHelpers.parentPaths(options.routes)) };
};
const adapter = load(await staged('router/authorization-sync.ts'), {
vue, 'vue-router': routing, '@vben/stores': stores,
'#/api/core/user': { getUserInfoApi() { requests.push('GET /me'); return Promise.resolve(typeof answer === 'function' ? answer() : clone(answer)); } },
'#/utils/authorization-sync': syncModule, './access': { generateAccess },
'./routes': { accessRoutes, routes: clone(routes) },
}, env);
access.setAccessToken('same-session');
access.setAccessCodes(initial);
user.setUserInfo(info(initial));
const first = await generateAccess({ roles: initial, routes: accessRoutes, router });
access.setAccessMenus(first.accessibleMenus);
access.setAccessRoutes(first.accessibleRoutes);
access.setIsAccessChecked(true);
await router.push('/group/welcome');
const dispose = adapter.installAuthorizationSync(router);
await settle();
return { access, user, tabbar, env, router, requests, adapter, dispose,
preparations: () => preparations,
answer(value) { answer = value; },
async poll(value) { answer = value; env.tick(); await settle(); },
};
}
test('same logged-in token gains/revokes real menus, matcher routes and reactive action rights', async () => {
const h = await harness();
try {
const { hasAccessByCodes } = useAccess();
const createButton = vue.computed(() => hasAccessByCodes(['role:create']));
assert.equal(h.router.hasRoute('RoleManage'), false);
assert.equal(createButton.value, false);
await h.poll(info(['role:read']));
assert.equal(h.router.hasRoute('RoleManage'), true);
assert.ok(h.access.accessMenus.some(menu => menu.children?.some(item => item.path === '/system/roles')));
await h.router.push('/system/roles');
const current = h.router.currentRoute.value;
const record = h.router.getRoutes().find(route => route.name === 'RoleManage');
const component = record.components.default;
const cachedDraft = { component: vue.markRaw({ draft: 'unsaved role form' }), key: '/system/roles', route: vue.markRaw(current) };
h.tabbar.cachedRoutes.set('/system/roles', cachedDraft);
await h.poll(info(['role:read', 'role:create']));
assert.equal(createButton.value, true);
assert.equal(h.router.currentRoute.value, current, 'authorized current route was navigated/replaced');
assert.equal(h.router.getRoutes().find(route => route.name === 'RoleManage').components.default, component);
assert.equal(h.tabbar.cachedRoutes.get('/system/roles').component.draft, 'unsaved role form');
await h.poll(info(['role:read']));
assert.equal(createButton.value, false);
assert.equal(h.router.currentRoute.value, current);
assert.equal(h.access.accessToken, 'same-session');
await h.poll(info([]));
assert.equal(h.router.hasRoute('RoleManage'), false);
assert.equal(h.router.hasRoute('SystemCenter'), false, 'fully revoked branch survived');
assert.equal(h.router.resolve('/system/roles').name, 'FallbackNotFound');
assert.ok(!h.access.accessMenus.some(menu => menu.path === '/system'));
assert.equal(h.router.currentRoute.value.path, '/group/welcome');
assert.equal(h.access.accessToken, 'same-session');
assert.equal(h.tabbar.cachedRoutes.has('/system/roles'), false);
} finally { h.dispose(); }
});
test('grant/revoke siblings preserves an authorized current matcher, component and pinned tab', async () => {
const h = await harness(['user:read']);
try {
await h.router.push('/system/users');
const current = h.router.currentRoute.value;
const components = h.router.getRoutes().find(route => route.name === 'UserManage').components;
h.tabbar.addTab({ ...current, key: '/system/users', meta: { ...current.meta, affixTab: true, keepAlive: true } });
await h.poll(info(['user:read', 'audit:read']));
assert.equal(h.router.hasRoute('AuditLog'), true);
assert.equal(h.router.currentRoute.value, current);
assert.equal(h.router.getRoutes().find(route => route.name === 'UserManage').components, components);
const audit = h.router.resolve('/system/audit');
h.tabbar.addTab({ ...audit, key: '/system/audit', meta: { ...audit.meta, affixTab: true, keepAlive: true } });
h.tabbar.cachedRoutes.set('/system/audit', { key: '/system/audit', route: vue.markRaw(audit), component: vue.markRaw({}) });
await h.poll(info(['user:read']));
assert.equal(h.router.hasRoute('AuditLog'), false);
assert.equal(h.router.currentRoute.value, current);
assert.deepEqual(normalized(h.tabbar.tabs.map(tab => tab.path)), ['/system/users']);
assert.equal(h.tabbar.cachedTabs.has('AuditLog'), false);
assert.equal(h.tabbar.cachedRoutes.has('/system/audit'), false);
assert.equal(h.tabbar.visitHistory.size, 1);
} finally { h.dispose(); }
});
test('equal/reordered permissions refresh profile without rebuilding menus; hidden polls pause and focus resumes', async () => {
const h = await harness(['role:read', 'role:create']);
try {
const before = h.preparations();
await h.poll({ ...info(['role:create', 'role:read', 'role:read']), realName: 'Changed display name' });
assert.equal(h.preparations(), before);
assert.equal(h.user.userInfo.realName, 'Changed display name');
assert.deepEqual([...h.env.timers.values()].map(timer => timer.delay), [3000]);
const requests = h.requests.length;
h.env.visible(false);
h.env.tick();
h.env.focus();
await settle();
assert.equal(h.requests.length, requests);
h.answer(info([]));
h.env.visible(true);
await settle();
assert.equal(h.router.hasRoute('RoleManage'), false);
assert.equal(h.requests.length, requests + 1);
h.answer(info(['role:read']));
h.env.focus();
await settle();
assert.equal(h.router.hasRoute('RoleManage'), true);
h.dispose();
const disposed = h.requests.length;
h.env.tick(); h.env.focus(); h.env.visible(true);
await settle();
assert.equal(h.env.timers.size, 0);
assert.equal(h.requests.length, disposed);
} finally { h.dispose(); }
});
test('concurrent events coalesce; logout disposes; old account response cannot overwrite a new session', async () => {
const h = await harness();
try {
const old = deferred();
h.answer(() => old.promise);
const before = h.requests.length;
h.env.tick(); h.env.focus(); h.env.visible(true); h.env.tick();
assert.equal(h.requests.length, before + 1);
h.access.setAccessToken(null);
assert.equal(h.env.timers.size, 0);
h.user.setUserInfo(info([], '8'));
h.access.setAccessCodes([]);
h.answer(info(['role:read'], '8'));
h.access.setAccessToken('new-session');
await settle();
assert.equal(h.router.hasRoute('RoleManage'), true);
old.resolve(info(['user:read', 'role:create'], '7'));
await settle();
assert.equal(h.access.accessToken, 'new-session');
assert.equal(h.user.userInfo.userId, '8');
assert.deepEqual(normalized(h.access.accessCodes), ['role:read']);
assert.equal(h.router.hasRoute('UserManage'), false);
assert.equal(h.env.timers.size, 1);
} finally { h.dispose(); }
});
test('wrong identity and late preparation are ignored; failed preparation can recover', async () => {
const env = environment();
let session = { token: 'A', userId: '7' };
let answer = info(['role:read'], '8');
const stagedSnapshot = deferred();
let prepares = 0, applied = 0;
let prepare = async () => { prepares += 1; return stagedSnapshot.promise; };
const sync = syncModule.createAuthorizationSync({ ...env, getSession: () => session, getCodes: () => [],
fetch: async () => answer, prepare: value => prepare(value), apply() { applied += 1; } });
sync.start();
await sync.refresh();
assert.equal(prepares, 0);
answer = info(['role:read']);
const pending = sync.refresh();
await Promise.resolve();
session = { token: 'B', userId: '9' };
stagedSnapshot.resolve({});
await pending;
assert.equal(applied, 0);
sync.stop();
session = { token: 'A', userId: '7' };
prepare = async () => { throw new Error('Temporary menu read failure'); };
sync.start();
await sync.refresh();
assert.equal(applied, 0);
prepare = async () => ({});
await sync.refresh();
assert.equal(applied, 1);
sync.stop();
});
test('reinstall is singleton and an old disposer cannot stop its replacement', async () => {
const h = await harness();
const replacement = h.adapter.installAuthorizationSync(h.router);
try {
h.dispose();
await settle();
assert.equal(h.env.timers.size, 1);
h.answer(info(['role:read']));
h.env.focus();
await settle();
assert.equal(h.router.hasRoute('RoleManage'), true);
} finally { replacement(); }
assert.equal(h.env.timers.size, 0);
});
test('actual request auth interceptor ignores an old 401; current 401 retains expiry handling; 403 never retries', async () => {
pinia.setActivePinia(pinia.createPinia());
const access = useAccessStore();
access.setAccessToken('new-session'); access.setIsAccessChecked(true);
const clients = [];
class RequestClient {
constructor() { this.responses = []; clients.push(this); }
addRequestInterceptor() {}
addResponseInterceptor(value) { this.responses.push(value); }
}
const preset = load(await source('packages/effects/request/src/request-client/preset-interceptors.ts'), {
'@vben/locales': { $t: key => key }, '@vben/utils': { isFunction: value => typeof value === 'function' },
axios: createRequire(require.resolve('@vben/request'))('axios'),
});
let logouts = 0;
load((await staged('api/request.ts')).replaceAll('import.meta.env.PROD', 'false').replaceAll('import.meta.env', '{}'), {
'@vben/hooks': { useAppConfig: () => ({ apiURL: '/api/v2' }) },
'@vben/preferences': { preferences }, '@vben/request': { ...preset, RequestClient }, '@vben/stores': stores,
'ant-design-vue': { message: { error() {} } }, '#/locales/t': { t: key => key },
'#/store': { useAuthStore: () => ({ logout: async () => { logouts += 1; } }) },
'#/utils/authorization-sync': syncModule, './core': { refreshTokenApi: async () => { throw new Error('No refresh token'); } },
'./error-message': { responseErrorMessage: () => 'Error' },
});
const authentication = clients[0].responses[1];
const error = (status, token) => ({ config: { headers: { Authorization: `Bearer ${token}` } }, response: { status } });
await assert.rejects(authentication.rejected(error(401, 'old-session')));
assert.equal(access.accessToken, 'new-session');
assert.equal(access.loginExpired, false);
await assert.rejects(authentication.rejected(error(403, 'new-session')));
assert.equal(access.accessToken, 'new-session');
assert.equal(logouts, 0);
await assert.rejects(authentication.rejected(error(401, 'new-session')));
assert.equal(access.accessToken, null);
assert.equal(access.loginExpired, true);
});
@@ -12,6 +12,9 @@ const engine=await readFile(enginePath,'utf8');
const tree=ts.createSourceFile('engine.ts',engine,ts.ScriptTarget.Latest,true);
const functions=tree.statements.filter(node=>ts.isFunctionDeclaration(node)&&['hasAuthority','menuHasVisibleWithForbidden'].includes(node.name?.text)).map(node=>node.getText(tree)).join('\n');
const {hasAuthority}=await import(moduleUrl(compile(functions+'\nexport {hasAuthority};')));
const sharedPath=process.env.KEFU_SSO_WORKSPACE?pathToFileURL(resolve(process.env.KEFU_SSO_WORKSPACE,'packages/@core/base/shared/src/utils/tree.ts')):new URL('../../../packages/@core/base/shared/src/utils/tree.ts',import.meta.url);
const shared=moduleUrl(compile(await readFile(sharedPath,'utf8')));
const {generateRoutesByFrontend}=await import(moduleUrl(compile(engine).replace("from '@vben-core/shared/utils'",`from '${shared}'`)));
test('基础角色只显示我的账号,业务目录使用实际Vben权限算法按子页面读权限开放',()=>{
const visible=permissions=>routes.filter(route=>!route.meta?.hideInMenu&&hasAuthority(route,permissions)).map(route=>route.name);
assert.deepEqual(visible([]),['GroupAccount']);
@@ -22,3 +25,31 @@ test('基础角色只显示我的账号,业务目录使用实际Vben权限算
const all=[...new Set(routes.flatMap(route=>(route.children||[]).flatMap(child=>child.meta?.authority||[])))];
assert.equal(visible(all).length,6);
});
test('角色读取权限经过实际Vben父子路由过滤后可进入角色页,兼容用户读取权限',async()=>{
for(const permissions of [['role:read'],['role:read','role:write'],['user:read']]){
const filtered=await generateRoutesByFrontend(structuredClone(routes.map(({component,...route})=>({...route,children:route.children?.map(({component,...child})=>child)}))),permissions);
const system=filtered.find(route=>route.name==='SystemCenter');
assert.ok(system,permissions.join(','));
assert.ok(system.children.some(route=>route.name==='RoleManage'),permissions.join(','));
if(!permissions.includes('user:read')) assert.ok(!system.children.some(route=>route.name==='UserManage'));
}
for(const permissions of [[],['role:write'],['role:create'],['role:delete']]){
const filtered=await generateRoutesByFrontend(structuredClone(routes.map(({component,...route})=>({...route,children:route.children?.map(({component,...child})=>child)}))),permissions);
assert.ok(!filtered.some(route=>route.name==='SystemCenter'),permissions.join(','));
}
});
test('每个已授予的页面都能通过实际父目录过滤,包括新增的细分读取权限',async()=>{
for(const parent of routes.filter(route=>!route.meta?.hideInMenu)){
for(const child of parent.children||[]){
for(const code of child.meta?.authority||[]){
const input=routes.map(({component,...route})=>({...route,children:route.children?.map(({component,...page})=>page)}));
const filtered=await generateRoutesByFrontend(structuredClone(input),[code]);
const directory=filtered.find(route=>route.name===parent.name);
assert.ok(directory,`${code}: ${parent.name}`);
assert.ok(directory.children.some(page=>page.name===child.name),`${code}: ${child.name}`);
}
}
}
});
@@ -29,6 +29,19 @@ const origin = 'http://127.0.0.1:5666';
const now = 1_000_000;
const code = 'C'.repeat(43);
test('不同访问地址先进入固定客服回调域,再在该域创建PKCE事务', async () => {
assert.equal(sso.groupLoginEntry(config, origin), '/auth/group-login');
assert.equal(sso.groupLoginEntry(config, 'http://127.0.0.1:8766'), config.redirect_uri);
assert.equal(sso.groupLoginEntry(config, 'http://localhost:5666'), config.redirect_uri);
const pending = storage();
await assert.rejects(sso.beginGroupLogin(config, 'http://127.0.0.1:8766', pending, webcrypto, now));
assert.equal(pending.values.size, 0);
assert.throws(() => sso.groupLoginEntry({...config, enabled: false}, origin));
assert.throws(() => sso.groupLoginEntry({...config, redirect_uri: 'https://evil.example/other'}, origin));
assert.throws(() => sso.groupLoginEntry({...config, authorize_url: 'https://evil.example/sso/authorize'}, origin));
assert.throws(() => sso.groupLoginEntry(config, `${origin}/auth/login`));
});
function storage() {
const values = new Map();
return { getItem: key => values.get(key) ?? null, setItem: (key, value) => values.set(key, value),
@@ -152,6 +165,90 @@ test('password login shares authoritative initialization and preserves forced pa
assert.equal(h.events.at(-1), 'route:/profile/password');
});
test('local and ZYT credentials share password API and use only current /me permissions', async () => {
for (const account of [
{ username: 'local_operator', permissions: ['stats:read', 'conversation:read'], home: '/model/stats' },
{ username: 'zyt_new', permissions: [], home: '/group/welcome' },
{ username: 'zyt_authorized', permissions: ['stats:read', 'user:read'], home: '/model/stats' },
]) {
const h = authHarness();
const requests = [];
const requestClient = {
async post(path, body) {
requests.push({ method: 'POST', path, body });
// Even stale/overbroad permissions in the login response are ignored.
return { access_token: `token-${account.username}`, user: { permissions: ['user:write', 'role:write'] } };
},
async get(path) {
requests.push({ method: 'GET', path });
return { id: 12, username: account.username, permissions: account.permissions,
must_change_password: false, role: account.permissions.length ? 'operator' : 'zyt_basic' };
},
};
h.api.loginApi = loadTs('../src/api/core/auth.ts', {
'#/api/request': { requestClient, baseRequestClient: { get() { assert.fail('Password login requested group config'); } } },
}).loginApi;
h.api.getUserInfoApi = loadTs('../src/api/core/user.ts', { '#/api/request': { requestClient } }).getUserInfoApi;
h.api.groupExchangeApi = async () => { assert.fail('Password login used group exchange'); };
await h.store.authLogin({ username: account.username, password: 'Fixture@2026' });
assert.equal(h.access.accessToken, `token-${account.username}`);
assert.deepEqual(h.access.accessCodes, account.permissions);
assert.deepEqual(h.user.userInfo.roles, account.permissions);
assert.equal(h.user.userInfo.username, account.username);
assert.equal(h.access.accessMenus.length, 0);
assert.equal(h.access.accessRoutes.length, 0);
assert.equal(h.access.isAccessChecked, false);
assert.equal(h.persisted.tabs.length, 0);
assert.equal(h.persisted.cachedUser, null);
assert.equal(h.persisted.lock, false);
assert.equal(h.store.loginLoading.value, false);
assert.deepEqual(h.events, ['reset-routes', 'reset-stores', `token:token-${account.username}`, 'route:' + account.home]);
assert.deepEqual(JSON.parse(JSON.stringify(requests)), [
{ method: 'POST', path: '/auth/login', body: { username: account.username, password: 'Fixture@2026', device_name: 'web' } },
{ method: 'GET', path: '/me' },
]);
}
});
test('ordinary password errors release loading and permit retry through the same authLogin', async () => {
const h = authHarness();
let attempts = 0;
h.api.loginApi = async () => {
if (++attempts === 1) throw new Error('账号或密码错误');
return { accessToken: 'retry-basic' };
};
const credentials = { username: 'zyt_new', password: 'Fixture@2026' };
await assert.rejects(h.store.authLogin(credentials), /账号或密码错误/);
assert.equal(h.store.loginLoading.value, false);
assert.deepEqual(h.events, []);
h.setUser({ userId: '8', username: 'zyt_new', roles: [], homePath: '/group/welcome' });
await h.store.authLogin(credentials);
assert.equal(attempts, 2);
assert.equal(h.store.loginLoading.value, false);
assert.equal(h.access.accessToken, 'retry-basic');
assert.deepEqual(h.access.accessCodes, []);
assert.equal(h.events.at(-1), 'route:/group/welcome');
});
test('ordinary password /me failure clears new identity and allows a clean retry', async () => {
const h = authHarness();
h.api.getUserInfoApi = async () => { throw new Error('revoked'); };
const credentials = { username: 'zyt_new', password: 'Fixture@2026' };
await assert.rejects(h.store.authLogin(credentials), /revoked/);
assert.equal(h.store.loginLoading.value, false);
assert.equal(h.access.accessToken, null);
assert.equal(h.access.accessCodes.length, 0);
assert.equal(h.access.accessMenus.length, 0);
assert.equal(h.access.accessRoutes.length, 0);
assert.equal(h.user.userInfo, null);
assert.equal(h.persisted.cachedUser, null);
h.api.getUserInfoApi = async () => ({ userId: '8', username: 'zyt_new', roles: [], homePath: '/group/welcome' });
await h.store.authLogin(credentials);
assert.equal(h.access.accessToken, 'local-user');
assert.equal(h.access.accessCodes.length, 0);
assert.equal(h.events.at(-1), 'route:/group/welcome');
});
test('failure to load current local identity clears the new token and authorization', async () => {
const h = authHarness();
h.api.getUserInfoApi = async () => { throw new Error('revoked'); };
@@ -0,0 +1,141 @@
import assert from 'node:assert/strict';
import { readFile } from 'node:fs/promises';
import { createRequire } from 'node:module';
import { resolve } from 'node:path';
import test from 'node:test';
import { fileURLToPath, pathToFileURL } from 'node:url';
import vm from 'node:vm';
const require = createRequire(process.env.KEFU_SSO_TYPESCRIPT
? pathToFileURL(resolve(process.env.KEFU_SSO_TYPESCRIPT, 'package.json'))
: new URL('../package.json', import.meta.url));
const ts = require('typescript');
const vue = require('vue');
const vueRequire = createRequire(require.resolve('vue/package.json'));
const { parse, compileScript, compileTemplate } = vueRequire('@vue/compiler-sfc');
const { renderToString } = vueRequire('@vue/server-renderer');
const workspace = process.env.KEFU_SSO_WORKSPACE || fileURLToPath(new URL('../../../', import.meta.url));
const translate = { t: value => value };
function load(source, dependencies) {
const compiled = ts.transpileModule(source, { compilerOptions: {
module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2022,
}, reportDiagnostics: true });
assert.deepEqual(compiled.diagnostics, []);
const module = { exports: {} };
vm.runInNewContext(compiled.outputText, { module, exports: module.exports,
require(name) { assert.ok(name in dependencies, `Unexpected import ${name}`); return dependencies[name]; },
console });
return module.exports;
}
// Compile the actual SFC and its actual permission-tree helpers. Ant components
// only render their props/slots; Vue setup, computed state and template run as-is.
const filename = fileURLToPath(new URL('../src/views/console/roles.vue', import.meta.url));
const { descriptor } = parse(await readFile(filename, 'utf8'), { filename });
const script = compileScript(descriptor, { id: 'local-role-page-test' });
const template = compileTemplate({ source: descriptor.template.content, filename,
id: 'local-role-page-test', compilerOptions: { bindingMetadata: script.bindings } });
assert.deepEqual(template.errors, []);
const helpers = load(await readFile(resolve(workspace, 'apps/web-antd/src/views/console/permission-menu.ts'), 'utf8'), {
'#/locales/t': translate,
});
const plain = tag => vue.defineComponent({ inheritAttrs: false,
setup(_props, { attrs, slots }) { return () => vue.h(tag, attrs, slots.default?.()); } });
const ant = Object.fromEntries(['Alert', 'Card', 'Form', 'FormItem', 'Input', 'Popconfirm', 'Tag']
.map(name => [name, plain('div')]));
ant.Button = plain('button');
ant.Tree = vue.defineComponent({ props: ['disabled'], setup(props) {
return () => vue.h('div', { 'data-permission-tree': String(Boolean(props.disabled)) });
} });
ant.Modal = vue.defineComponent({ props: ['open', 'title', 'okText'], setup(props, { slots }) {
return () => props.open ? vue.h('section', { 'data-modal-title': props.title }, [
vue.h('button', { 'data-modal-action': '' }, props.okText), slots.default?.(),
]) : null;
} });
ant.Table = vue.defineComponent({ props: ['dataSource', 'columns'], setup(props, { slots }) {
return () => vue.h('div', {}, props.dataSource.map(record => vue.h('div', { 'data-role': record.code },
props.columns.map(column => slots.bodyCell?.({ column, record })))));
} });
const catalog = [
{ code: 'user:read', name: '查看用户', group_name: '系统管理', page_name: '用户管理', kind: 'page' },
{ code: 'custom:report', name: '自定义报表', group_name: '报表', page_name: '报表', kind: 'link' },
];
const rows = [
{ code: 'admin', name: '管理员', permissions: ['user:read'], builtin: false, user_count: 0 },
{ code: 'zyt_basic', name: '集团基础用户', permissions: [], builtin: false, user_count: 0 },
{ code: 'reviewer', name: '审核员', permissions: ['user:read', 'custom:report', 'removed:code'], builtin: false, user_count: 0 },
];
async function page(codes, configure = () => {}) {
const calls = { save: [], remove: [], success: [] };
let state;
const component = load(script.content, {
vue, 'vue-router': { useRouter: () => ({ push() {} }) },
'@vben/access': { useAccess: () => ({ hasAccessByCodes: values => values.some(value => codes.includes(value)) }) },
'ant-design-vue': { ...ant, message: { warning() {}, success(value) { calls.success.push(value); } } },
'#/locales/t': translate, './permission-menu': helpers,
'#/api/console': {
async fetchRoles() { return { roles: rows }; }, async fetchPermissions() { return { permissions: catalog }; },
async saveRole(data) { calls.save.push(data); return { roles: rows }; },
async deleteRole(code) { calls.remove.push(code); return { roles: rows }; },
},
}).default;
component.render = load(template.code, { vue }).render;
const original = component.setup;
component.setup = (props, context) => {
state = original(props, context);
state.roles.value = rows;
state.permissions.value = catalog;
configure(state);
return state;
};
const html = await renderToString(vue.createSSRApp(component));
return { state, calls, html };
}
test('管理员和集团基础角色实际组件均为只读:名称/权限禁用、查看与关闭、禁止写删', async () => {
for (const row of rows.slice(0, 2)) {
const { state, calls, html } = await page(['role:write', 'role:delete'], state => state.openEdit(row));
assert.equal(state.readonly.value, true);
assert.match(html, /data-modal-title="查看角色"/);
assert.match(html, /data-modal-action(?:="")?>关闭/);
assert.match(html, new RegExp(`value="${row.name}" disabled`));
assert.match(html, /data-permission-tree="true"/);
assert.match(html, new RegExp(`data-role="${row.code}"[\\s\\S]*?<button[^>]*>查看<\\/button>[\\s\\S]*?<button[^>]*disabled`));
if (row.code === 'zyt_basic') assert.match(html, /请到「系统管理 → 用户管理」为账号分配其他业务角色/);
const before = [...state.form.permissions];
state.onCheck(['user:read']);
assert.deepEqual([...state.form.permissions], before);
await state.submit();
await state.remove(row);
assert.deepEqual(calls.save, []);
assert.deepEqual(calls.remove, []);
}
});
test('普通业务角色仍可编辑并保留自定义权限,保存只过滤失效码并说明在线用户自动同步', async () => {
const { state, calls, html } = await page(['role:read', 'role:write', 'role:delete'], state => state.openEdit(rows[2]));
assert.equal(state.readonly.value, false);
assert.match(html, /data-modal-title="编辑角色"/);
assert.match(html, /data-permission-tree="false"/);
await state.submit();
assert.equal(calls.save.length, 1);
assert.deepEqual([...calls.save[0].permissions], ['user:read', 'custom:report']);
assert.equal(calls.save[0].code, 'reviewer');
assert.match(calls.success[0], /后端权限立即生效/);
assert.match(calls.success[0], /菜单和按钮将在数秒内自动更新/);
await state.remove(rows[2]);
assert.deepEqual(calls.remove, ['reviewer']);
});
test('角色只读用户不能通过组件方法保存或删除业务角色', async () => {
const { state, calls } = await page(['role:read'], state => state.openEdit(rows[2]));
assert.equal(state.readonly.value, true);
await state.submit();
await state.remove(rows[2]);
assert.deepEqual(calls.save, []);
assert.deepEqual(calls.remove, []);
});
@@ -0,0 +1,127 @@
import assert from 'node:assert/strict';
import { readFile } from 'node:fs/promises';
import { createRequire } from 'node:module';
import { resolve } from 'node:path';
import test from 'node:test';
import { fileURLToPath, pathToFileURL } from 'node:url';
import vm from 'node:vm';
const require = createRequire(process.env.KEFU_SSO_TYPESCRIPT
? pathToFileURL(resolve(process.env.KEFU_SSO_TYPESCRIPT, 'package.json'))
: new URL('../package.json', import.meta.url));
const ts = require('typescript');
const vue = require('vue');
const vueRequire = createRequire(require.resolve('vue/package.json'));
const { parse, compileScript, compileTemplate } = vueRequire('@vue/compiler-sfc');
const { renderToString } = vueRequire('@vue/server-renderer');
function load(source, dependencies) {
const compiled = ts.transpileModule(source, { compilerOptions: {
module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2022,
}, reportDiagnostics: true });
assert.deepEqual(compiled.diagnostics, []);
const module = { exports: {} };
vm.runInNewContext(compiled.outputText, { module, exports: module.exports,
require(name) { assert.ok(name in dependencies, `Unexpected import ${name}`); return dependencies[name]; },
// Normal password login must not read browser redirect/storage APIs.
get window() { assert.fail('Password login accessed window'); },
get localStorage() { assert.fail('Password login accessed localStorage'); },
console });
return module.exports;
}
const filename = fileURLToPath(new URL('../src/views/_core/authentication/login.vue', import.meta.url));
const { descriptor } = parse(await readFile(filename, 'utf8'), { filename });
const script = compileScript(descriptor, { id: 'password-login-page-test' });
const template = compileTemplate({ source: descriptor.template.content, filename,
id: 'password-login-page-test', compilerOptions: { bindingMetadata: script.bindings } });
assert.deepEqual(template.errors, []);
async function page(authLogin) {
let submit;
let formProps;
const store = vue.reactive({ loginLoading: false, authLogin });
// Render the real page SFC. The shared form adapter displays its schema/slots
// and exposes the submit listener assigned by the compiled page template.
const AuthenticationLogin = vue.defineComponent({
props: ['formSchema', 'loading', 'title', 'subTitle', 'showCodeLogin', 'showForgetPassword',
'showQrcodeLogin', 'showRegister', 'showRememberMe', 'showThirdPartyLogin'],
emits: ['submit'],
setup(props, { slots }) {
formProps = props;
const instance = vue.getCurrentInstance();
submit = values => instance.vnode.props.onSubmit(values);
return () => vue.h('section', {}, [
vue.h('h1', {}, props.title), vue.h('p', {}, props.subTitle),
vue.h('form', {}, [
...props.formSchema.map(field => vue.h('input', { name: field.fieldName,
type: field.component === 'VbenInputPassword' ? 'password' : 'text',
placeholder: field.componentProps.placeholder })),
vue.h('button', { type: 'submit', disabled: props.loading }, '登录'),
]),
slots['third-party-login']?.(),
]);
},
});
// No config API, group URL helper or router is provided: an accidental import
// or use fails this actual SFC test instead of silently becoming a mock call.
const component = load(script.content, { vue,
'@vben/common-ui': { AuthenticationLogin, z: { string: () => ({ min: () => ({}) }) } },
'#/locales/t': { t: value => value }, '#/store': { useAuthStore: () => store },
}).default;
component.render = load(template.code, { vue }).render;
const html = await renderToString(vue.createSSRApp(component));
return { html, submit, store, formProps };
}
test('实际登录页只有一份账号密码表单和登录按钮,ZYT/本地均走普通提交', async () => {
const calls = [];
const view = await page(async values => { calls.push(values); });
assert.equal((view.html.match(/<form(?:\s|>)/g) || []).length, 1);
assert.equal((view.html.match(/<input\b/g) || []).length, 2);
assert.equal((view.html.match(/<button\b/g) || []).length, 1);
assert.match(view.html, /name="username"[^>]*placeholder="ZYT 账号或客服本地账号"/);
assert.match(view.html, /name="password" type="password"[^>]*placeholder="ZYT 密码或客服本地密码"/);
assert.match(view.html, /ZYT 首次登录为基础用户,业务权限由客服管理员分配/);
assert.doesNotMatch(view.html, /集团账号登录|重试获取集团入口|href=|group-login/);
for (const name of ['showCodeLogin', 'showForgetPassword', 'showQrcodeLogin', 'showRegister',
'showRememberMe', 'showThirdPartyLogin']) assert.equal(view.formProps[name], false);
for (const username of ['local_operator', 'zyt_operator']) {
await view.submit({ username, password: 'Fixture@2026' });
}
assert.deepEqual(calls, [
{ username: 'local_operator', password: 'Fixture@2026' },
{ username: 'zyt_operator', password: 'Fixture@2026' },
]);
});
test('实际登录页普通提交失败后仍可用同一份表单重试', async () => {
let attempts = 0;
const view = await page(async () => {
if (++attempts === 1) throw new Error('账号或密码错误');
return { userInfo: { username: 'zyt_operator' } };
});
const credentials = { username: 'zyt_operator', password: 'Fixture@2026' };
await assert.rejects(view.submit(credentials), /账号或密码错误/);
assert.equal((await view.submit(credentials)).userInfo.username, 'zyt_operator');
assert.equal(attempts, 2);
});
test('ZYT/本地密码只随统一POST body发送,不进入URL或localStorage', async () => {
const requests = [];
const api = load(await readFile(new URL('../src/api/core/auth.ts', import.meta.url), 'utf8'), {
'#/locales/t': { t: value => value },
'#/api/request': {
requestClient: { async post(...args) { requests.push(args); return { access_token: 'current-token' }; } },
baseRequestClient: { get() { assert.fail('Password login fetched group config'); } },
},
});
for (const username of ['local_operator', 'zyt_operator']) {
const result = await api.loginApi({ username, password: 'Fixture@2026', redirect: 'https://unused.example/' });
assert.equal(result.accessToken, 'current-token');
}
assert.deepEqual(JSON.parse(JSON.stringify(requests)), [
['/auth/login', { username: 'local_operator', password: 'Fixture@2026', device_name: 'web' }],
['/auth/login', { username: 'zyt_operator', password: 'Fixture@2026', device_name: 'web' }],
]);
});
+2 -2
View File
@@ -6,12 +6,12 @@ export default defineConfig(async () => {
vite: {
server: {
proxy: {
// 前端统一请求 /api/*,代理到 admin_api.py 的 /api/v2/*。
// 兼容 /api/* 和标准 /api/v2/*,避免重复添加版本前缀。
// 走代理而不是直连,是为了开发态和生产态同源、没有 CORS 差异——
// 生产环境前端是静态文件,由同一个 Nginx 反代到后端。
'/api': {
changeOrigin: true,
rewrite: (path) => path.replace(/^\/api/, '/api/v2'),
rewrite: (path) => path.startsWith('/api/v2/') ? path : path.replace(/^\/api/, '/api/v2'),
// 端口可用环境变量覆盖:本机同时跑多套后端时(比如拿一次性的库
// 验证前端,不碰生产库)不用改代码。
target: process.env.VITE_API_TARGET || 'http://127.0.0.1:8766',
+5 -1
View File
@@ -6,7 +6,7 @@ x-common: &common
context: ../..
dockerfile: deploy/im-admin/Dockerfile
restart: unless-stopped
environment:
environment: &common-environment
PYTHONDONTWRITEBYTECODE: "1"
PYTHONUNBUFFERED: "1"
KNOWLEDGE_QDRANT_URL: "${KNOWLEDGE_QDRANT_URL:-}"
@@ -32,6 +32,10 @@ x-common: &common
services:
api:
<<: *common
environment:
<<: *common-environment
# File must be mounted read-only outside public assets; never put the key in URLs.
ZYT_GROUP_METADATA_KEY_FILE: "${ZYT_GROUP_METADATA_KEY_FILE:-}"
command:
- python
- admin_api.py
+42 -6
View File
@@ -49,6 +49,8 @@ import knowledge_api
import remote_api
import zyt_auth
import group_sso
import group_access
import group_authorization
# 账号下拉框里「全部」的取值。用 -1 而不是 0:0 早就有含义了(旧版未归属数据),
# 而"不传"在 FastAPI 里就是 None,同样已经被占用。
@@ -64,7 +66,7 @@ DEV_ORIGINS = [
# ── 请求体 ───────────────────────────────────────────────────────────────────
class LoginBody(BaseModel):
username: str
password: str
password: str = Field(repr=False)
device_name: str = "web"
@@ -370,10 +372,12 @@ def create_app(
release_directory = database.path.parent / "releases"
release_directory.mkdir(parents=True, exist_ok=True)
app = FastAPI(title="企微客服助手管理 API", version="2.0")
group_access_snapshot = group_access.AccessSnapshot(database)
group_authorization_service = group_authorization.GroupAuthorization(database)
@app.exception_handler(RequestValidationError)
async def safe_group_validation(request: Request, exc: RequestValidationError):
if request.url.path == "/api/v2/auth/group/exchange":
if request.url.path in {"/api/v2/auth/group/exchange", "/api/v2/auth/login"}:
# Pydantic's default error projection includes the submitted input.
return JSONResponse(status_code=422, content={"detail": "集团登录请求无效,请重新登录"})
return await request_validation_exception_handler(request, exc)
@@ -386,7 +390,7 @@ def create_app(
# callback codes in the ASGI scope later used by access logging.
request.scope["query_string"] = b""
response = await call_next(request)
if path == "/auth/group-login" or path.startswith("/api/v2/auth/group/"):
if path in {"/auth/group-login", "/api/v2/auth/login"} or path.startswith("/api/v2/auth/group/"):
response.headers["Cache-Control"] = "no-store"
response.headers["Referrer-Policy"] = "no-referrer"
return response
@@ -698,6 +702,22 @@ def create_app(
)
return {"ok": True}
@app.post("/api/v2/group/access-snapshot")
async def group_access_metadata(request: Request):
# Machine credentials only: no human login/enrollment or authorization
# mutation is reachable from this dedicated read-only projection.
return await group_access.handle_request(request, group_access_snapshot)
@app.post("/api/v2/group/access-options")
async def group_authorization_options(request: Request):
return await group_authorization.handle_request(
request, group_authorization_service, group_authorization.OPTIONS_PATH)
@app.post("/api/v2/group/access-update")
async def group_authorization_update(request: Request):
return await group_authorization.handle_request(
request, group_authorization_service, group_authorization.UPDATE_PATH)
# ── 管理端认证 ───────────────────────────────────────────────────────
@app.get("/api/v2/auth/group/config")
async def group_config() -> dict:
@@ -733,22 +753,38 @@ def create_app(
if not backend.LOGIN_LIMITER.allowed(key):
raise HTTPException(status_code=429, detail="登录失败次数过多,请稍后再试")
user = database.authenticate(body.username, body.password)
group_login = False
if user is None:
config = group_sso.GroupPasswordConfig.from_env()
if config is not None:
try:
identity = await asyncio.to_thread(
group_sso.password_login, config, body.username, body.password,
)
user = await asyncio.to_thread(database.group_sso_user, identity)
group_login = True
except ValueError:
# Includes remote failures and locally disabled bindings.
# Do not reflect upstream responses, URLs, or credentials.
user = None
if not user:
backend.LOGIN_LIMITER.failure(key)
database.audit(None, "login.failed", f"username={body.username}", ip)
raise HTTPException(status_code=401, detail="用户名或密码不正确")
backend.LOGIN_LIMITER.success(key)
lifetime = 8 * 3600 if group_login else 30 * 86400
token, _ = database.create_token(
int(user["id"]), "api", body.device_name, 30 * 86400
int(user["id"]), "api", body.device_name, lifetime
)
database.audit(int(user["id"]), "login.api", body.device_name, ip)
database.audit(int(user["id"]), "login.group" if group_login else "login.api",
"provider=zyt_group" if group_login else body.device_name, ip)
principal = Principal(user, database.permissions_for_user(int(user["id"])))
# 首次登录不拦 API:改密页面本身也要调接口。前端凭
# must_change_password 把用户锁在改密页上,比后端一刀切 403 好用得多
# ——老实现那样连改密接口都调不通,只能去网页后台改。
return {
"access_token": token,
"expires_in": 30 * 86400,
"expires_in": lifetime,
"user": principal.payload(),
}
+255
View File
@@ -0,0 +1,255 @@
"""Authenticated, read-only group metadata snapshots; never enroll or grant."""
from __future__ import annotations
import hashlib
import hmac
import json
import os
import re
import threading
import time
from pathlib import Path
from typing import Any, Callable
SIGNED_PATH = "/group/access-snapshot"
MAX_REQUEST_BYTES = 16 * 1024
MAX_RESPONSE_BYTES = 4 * 1024 * 1024
MAX_GIDS = 100
MAX_PERMISSIONS = 2000
WINDOW_SECONDS = 90
MAX_NONCES = 8192
_HEX_32 = re.compile(r"[0-9a-f]{32}\Z")
_HEX_64 = re.compile(r"[0-9a-f]{64}\Z")
_GID = re.compile(r"gid_[0-9a-f]{32}\Z")
_TIMESTAMP = re.compile(r"(?:0|[1-9][0-9]{0,11})\Z")
class AccessError(ValueError):
def __init__(self, status: int, message: str):
super().__init__(message)
self.status = status
def _unavailable() -> AccessError:
return AccessError(503, "集团授权元数据暂不可用")
def load_key() -> bytes:
"""Load only the explicitly configured, bounded, non-public key file."""
configured = os.environ.get("ZYT_GROUP_METADATA_KEY_FILE", "")
try:
path = Path(configured)
if not configured or not path.is_absolute() or path.is_symlink():
raise ValueError("Invalid key location")
resolved = path.resolve(strict=True)
project = Path(__file__).resolve().parent.parent
public_roots = (
project / "admin-web/apps/web-antd/dist",
project / "admin-web/apps/mobile-h5/dist",
)
if any(resolved.is_relative_to(root.resolve()) for root in public_roots):
raise ValueError("Public key location")
with resolved.open("rb") as source:
raw = source.read(65)
if not re.fullmatch(rb"[0-9a-fA-F]{64}", raw):
raise ValueError("Invalid key contents")
return bytes.fromhex(raw.decode("ascii"))
except (OSError, ValueError, RuntimeError):
raise _unavailable() from None
def request_signature(key: bytes, body: bytes, timestamp: str, nonce: str) -> str:
signed = f"POST\n{SIGNED_PATH}\n{timestamp}\n{nonce}\n{hashlib.sha256(body).hexdigest()}"
return hmac.new(key, signed.encode("ascii"), hashlib.sha256).hexdigest()
def response_signature(key: bytes, body: bytes, nonce: str) -> str:
signed = f"response\n{nonce}\n{hashlib.sha256(body).hexdigest()}"
return hmac.new(key, signed.encode("ascii"), hashlib.sha256).hexdigest()
def _object(pairs):
result = {}
for key, value in pairs:
if key in result:
raise ValueError("Duplicate JSON key")
result[key] = value
return result
def parse_request(body: bytes) -> dict[str, Any]:
try:
def invalid_constant(_value):
raise ValueError("Invalid JSON constant")
value = json.loads(body.decode("utf-8"), object_pairs_hook=_object,
parse_constant=invalid_constant)
if not isinstance(value, dict) or set(value) != {"client_id", "gids", "include_catalog"}:
raise ValueError("Invalid fields")
gids = value["gids"]
if value["client_id"] != "kefu" or type(value["include_catalog"]) is not bool:
raise ValueError("Invalid client")
if not isinstance(gids, list) or len(gids) > MAX_GIDS:
raise ValueError("Invalid subjects")
if any(not isinstance(gid, str) or not _GID.fullmatch(gid) for gid in gids):
raise ValueError("Invalid identity")
if len(set(gids)) != len(gids):
raise ValueError("Duplicate identities")
return value
except (ValueError, UnicodeError, RecursionError):
raise AccessError(422, "集团授权元数据请求无效") from None
def _positive(value: Any) -> int:
if type(value) is not int or value < 1:
raise _unavailable()
return value
def _text(value: Any) -> str:
if not isinstance(value, str):
raise _unavailable()
return value
def snapshot(database, request: dict[str, Any], generated_at: int) -> dict[str, Any]:
"""Project existing bindings only. Authorization stays in its existing reader."""
gids = request["gids"]
rows = {}
if gids:
with database.connect() as db:
placeholders = ",".join("?" for _ in gids)
found = db.execute(f"""SELECT b.gid,b.source_admin_id,b.company_id,
b.source_account,b.source_name,u.id,u.username,u.active,
u.role AS role_code,COALESCE(r.name,u.role) AS role_name
FROM group_sso_bindings b JOIN users u ON u.id=b.user_id
LEFT JOIN roles r ON r.code=u.role
WHERE b.provider=? AND b.gid IN ({placeholders})""",
["zyt_group", *gids]).fetchall()
rows = {row["gid"]: dict(row) for row in found}
subjects = []
for gid in gids:
row = rows.get(gid)
if row is None:
subjects.append({"gid": gid, "found": False})
continue
user_id = _positive(row["id"])
# Deliberately use the same effective-permission reader as ordinary API
# authentication. Inactive users are empty there; do not reinterpret
# catalog enabled flags as API authorization rules.
permissions = sorted(database.permissions_for_user(user_id))
if len(permissions) > MAX_PERMISSIONS or any(not isinstance(code, str) for code in permissions):
raise _unavailable()
# A concurrent deactivation must not pair an inactive metadata row with
# nonempty permissions. This does not claim a cross-request transaction.
if not row["active"]:
permissions = []
subjects.append({
"gid": gid, "found": True,
"source_admin_id": _positive(row["source_admin_id"]),
"company_id": _positive(row["company_id"]),
"user": {
"id": user_id, "username": _text(row["username"]),
"source_account": _text(row["source_account"]),
"source_name": _text(row["source_name"]), "active": bool(row["active"]),
"role": {"code": _text(row["role_code"]), "name": _text(row["role_name"])},
"permissions": permissions,
},
})
catalog = None
if request["include_catalog"]:
definitions = database.permission_catalog()
if len(definitions) > MAX_PERMISSIONS:
raise _unavailable()
permissions = []
seen = set()
for item in definitions:
code = _text(item["code"])
if not code or code in seen or item["kind"] not in {"page", "action", "link"}:
raise _unavailable()
seen.add(code)
# A custom link is a page in this protocol; never expose its URL.
permissions.append({
"code": code, "name": _text(item["name"]),
"group": _text(item["group_name"]), "page": _text(item["page_name"]),
"kind": "page" if item["kind"] == "link" else item["kind"],
"source": _text(item["source"]), "enabled": bool(item["enabled"]),
"sort": int(item["sort_order"]),
})
catalog = {"permissions": permissions}
return {"version": 1, "client_id": "kefu", "generated_at": generated_at,
"subjects": subjects, "catalog": catalog}
class AccessSnapshot:
"""One application instance's bounded, fail-closed read-protocol replay cache.
Memory replay protection is permitted by metadata v1. Multi-worker installs
should share a nonce store when replay rejection across workers is required.
"""
def __init__(self, database, clock: Callable[[], float] | None = None,
max_nonces: int = MAX_NONCES):
self.database = database
self.clock = clock or time.time
self.max_nonces = max_nonces
self._nonces: dict[str, int] = {}
self._lock = threading.Lock()
def respond(self, body: bytes, headers) -> tuple[bytes, dict[str, str]]:
key = load_key()
if len(body) > MAX_REQUEST_BYTES:
raise AccessError(413, "集团授权元数据请求过大")
timestamp = headers.get("x-group-timestamp", "")
nonce = headers.get("x-group-nonce", "")
signature = headers.get("x-group-signature", "")
now = int(self.clock())
if (not _TIMESTAMP.fullmatch(timestamp) or not _HEX_32.fullmatch(nonce)
or not _HEX_64.fullmatch(signature) or abs(int(timestamp) - now) > WINDOW_SECONDS):
raise AccessError(401, "集团服务认证无效")
if not hmac.compare_digest(request_signature(key, body, timestamp, nonce), signature):
raise AccessError(401, "集团服务认证无效")
with self._lock:
self._nonces = {value: expires for value, expires in self._nonces.items() if expires >= now}
if nonce in self._nonces:
raise AccessError(401, "集团服务认证无效")
if len(self._nonces) >= self.max_nonces:
raise _unavailable()
self._nonces[nonce] = int(timestamp) + WINDOW_SECONDS
request = parse_request(body)
try:
result = snapshot(self.database, request, now)
raw = json.dumps(result, ensure_ascii=False, separators=(",", ":"), allow_nan=False).encode("utf-8")
if len(raw) > MAX_RESPONSE_BYTES:
raise _unavailable()
except AccessError:
raise
except Exception:
# No upstream/database values or deployment paths may reach callers.
raise _unavailable() from None
return raw, {"Cache-Control": "no-store",
"X-Group-Response-Signature": response_signature(key, raw, nonce)}
async def handle_request(request, service: AccessSnapshot):
"""Read bounded raw bytes; authenticate before any JSON interpretation."""
from fastapi.responses import JSONResponse, Response
try:
if request.url.query:
raise AccessError(422, "集团授权元数据请求无效")
for name in ("x-group-timestamp", "x-group-nonce", "x-group-signature"):
if len(request.headers.getlist(name)) != 1:
raise AccessError(401, "集团服务认证无效")
raw = bytearray()
async for chunk in request.stream():
if len(raw) + len(chunk) > MAX_REQUEST_BYTES:
raise AccessError(413, "集团授权元数据请求过大")
raw.extend(chunk)
# The DB work and key-file access run outside the application's loop.
import asyncio
body, headers = await asyncio.to_thread(service.respond, bytes(raw), request.headers)
return Response(content=body, media_type="application/json", headers=headers)
except AccessError as exc:
return JSONResponse(status_code=exc.status, content={"detail": str(exc)},
headers={"Cache-Control": "no-store"})
+360
View File
@@ -0,0 +1,360 @@
"""Machine-only edits of existing Group bindings in KF's native role model.
The caller authenticates the Group owner; the signed actor ID is its assertion,
not a KF user ID. SQLite serializes revisions, private roles, assignment and audit.
Neither this module nor its HTTP routes enroll users or accept human sessions.
"""
from __future__ import annotations
import hashlib
import hmac
import json
import secrets
import time
from typing import Any, Callable
import group_access
import rbac
OPTIONS_PATH = "/group/access-options"
UPDATE_PATH = "/group/access-update"
MAX_REQUEST_BYTES = 256 * 1024
MAX_RESPONSE_BYTES = group_access.MAX_RESPONSE_BYTES
MAX_PERMISSIONS = group_access.MAX_PERMISSIONS
MAX_NONCES = group_access.MAX_NONCES
WINDOW_SECONDS = group_access.WINDOW_SECONDS
_POSITIVE_MAX = 9_223_372_036_854_775_807
_BASE_FIELDS = {"client_id", "gid", "source_admin_id", "company_id"}
_WRITE_FIELDS = {"actor_source_admin_id", "request_id", "expected_revision", "role_code", "permissions"}
_SCHEMA = (
"""CREATE TABLE IF NOT EXISTS group_authorization_nonces (
nonce TEXT PRIMARY KEY, expires_at INTEGER NOT NULL)""",
"""CREATE INDEX IF NOT EXISTS idx_group_authorization_nonce_expiry
ON group_authorization_nonces(expires_at)""",
"""CREATE TABLE IF NOT EXISTS group_authorization_requests (
request_id TEXT PRIMARY KEY, body_digest TEXT NOT NULL,
result_json BLOB NOT NULL, created_at INTEGER NOT NULL)""",
"""CREATE TABLE IF NOT EXISTS group_authorization_private_roles (
role_code TEXT PRIMARY KEY REFERENCES roles(code) ON DELETE CASCADE,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
permissions_digest TEXT NOT NULL, created_at INTEGER NOT NULL)""",
"""CREATE INDEX IF NOT EXISTS idx_group_authorization_private_owner
ON group_authorization_private_roles(user_id,permissions_digest)""",
"""CREATE TABLE IF NOT EXISTS group_authorization_audit (
request_id TEXT PRIMARY KEY, actor_source_admin_id INTEGER NOT NULL,
target_user_id INTEGER NOT NULL, detail_json TEXT NOT NULL,
created_at INTEGER NOT NULL)""",
)
class AuthorizationError(group_access.AccessError):
pass
def _unavailable() -> AuthorizationError:
return AuthorizationError(503, "客服授权服务暂不可用")
def request_signature(key: bytes, body: bytes, timestamp: str, nonce: str, path: str) -> str:
if path not in {OPTIONS_PATH, UPDATE_PATH}:
raise ValueError("Unsupported authorization path")
canonical = f"POST\n{path}\n{timestamp}\n{nonce}\n{hashlib.sha256(body).hexdigest()}"
return hmac.new(key, canonical.encode("ascii"), hashlib.sha256).hexdigest()
def _identifier(value: Any) -> bool:
return type(value) is int and 0 < value <= _POSITIVE_MAX
def _code(value: Any) -> bool:
return (isinstance(value, str) and 0 < len(value) <= 256
and all(ord(char) >= 32 and ord(char) != 127
and not 0xD800 <= ord(char) <= 0xDFFF for char in value))
def parse_request(body: bytes, path: str) -> dict[str, Any]:
"""Validate exact types/fields without normalizing identity or permission codes."""
try:
def invalid_constant(_value):
raise ValueError("Invalid JSON constant")
value = json.loads(body.decode("utf-8"), object_pairs_hook=group_access._object,
parse_constant=invalid_constant)
fields = _BASE_FIELDS | (_WRITE_FIELDS if path == UPDATE_PATH else set())
if not isinstance(value, dict) or set(value) != fields:
raise ValueError("Invalid fields")
if (value["client_id"] != "kefu" or not isinstance(value["gid"], str)
or not group_access._GID.fullmatch(value["gid"])
or not _identifier(value["source_admin_id"]) or not _identifier(value["company_id"])):
raise ValueError("Invalid subject")
if path == UPDATE_PATH:
if (not _identifier(value["actor_source_admin_id"])
or not isinstance(value["request_id"], str)
or not group_access._HEX_32.fullmatch(value["request_id"])
or not isinstance(value["expected_revision"], str)
or not group_access._HEX_64.fullmatch(value["expected_revision"])
or not _code(value["role_code"])):
raise ValueError("Invalid mutation")
permissions = value["permissions"]
if permissions is not None and (not isinstance(permissions, list)
or len(permissions) > MAX_PERMISSIONS
or any(not _code(code) for code in permissions)
or len(set(permissions)) != len(permissions)):
raise ValueError("Invalid permissions")
return value
except (ValueError, UnicodeError, RecursionError):
raise AuthorizationError(422, "客服授权请求无效") from None
def _encode(value: Any) -> bytes:
raw = json.dumps(value, ensure_ascii=False, separators=(",", ":"), allow_nan=False).encode("utf-8")
if len(raw) > MAX_RESPONSE_BYTES:
raise _unavailable()
return raw
def _binding(db, request: dict[str, Any]):
row = db.execute("""SELECT b.provider,b.gid,b.source_admin_id,b.company_id,
b.source_account,b.source_name,u.id,u.username,u.active,u.role AS role_code,
r.name AS role_name FROM group_sso_bindings b JOIN users u ON u.id=b.user_id
JOIN roles r ON r.code=u.role WHERE b.provider='zyt_group' AND b.gid=?""",
(request["gid"],)).fetchone()
if row is None:
raise AuthorizationError(404, "客服绑定用户不存在")
if (row["source_admin_id"] != request["source_admin_id"]
or row["company_id"] != request["company_id"]):
raise AuthorizationError(409, "客服绑定身份已变化,请重新加载")
return row
def _assigned(db, role: str) -> list[str]:
# Exact native authorization semantics: enabled is navigation metadata only.
result = sorted(rbac.permissions_of(db, role))
if len(result) > MAX_PERMISSIONS or any(not _code(code) for code in result):
raise _unavailable()
return result
def _revision(db, row) -> str:
identity = {name: row[name] for name in ("provider", "gid", "source_admin_id", "company_id",
"id", "username", "source_account", "source_name", "role_code")}
identity["active"] = bool(row["active"])
identity["permissions"] = _assigned(db, row["role_code"])
return hashlib.sha256(_encode(identity)).hexdigest()
def _result(db, row, now: int) -> dict[str, Any]:
definitions = rbac.catalog(db)
if len(definitions) > MAX_PERMISSIONS:
raise _unavailable()
catalog = []
known = set()
for item in definitions:
if (not _code(item["code"]) or item["code"] in known
or item["kind"] not in {"page", "action", "link"}
or any(not isinstance(item[name], str) for name in ("name", "group_name", "page_name", "source"))):
raise _unavailable()
known.add(item["code"])
catalog.append({"code": item["code"], "name": item["name"], "group": item["group_name"],
"page": item["page_name"], "kind": "page" if item["kind"] == "link" else item["kind"],
"source": item["source"], "enabled": bool(item["enabled"]), "sort": int(item["sort_order"])})
roles = []
for role in db.execute("""SELECT r.code,r.name FROM roles r
LEFT JOIN group_authorization_private_roles p ON p.role_code=r.code
WHERE p.role_code IS NULL OR p.user_id=? OR r.code=?
ORDER BY r.builtin DESC,r.code""", (row["id"], row["role_code"])):
assigned = _assigned(db, role["code"])
if not set(assigned) <= known or not _code(role["code"]) or not isinstance(role["name"], str):
raise _unavailable()
roles.append({"code": role["code"], "name": role["name"], "permissions": assigned})
assigned = _assigned(db, row["role_code"])
if not any(role["code"] == row["role_code"] for role in roles):
raise _unavailable()
subject = {"gid": row["gid"], "found": True, "source_admin_id": row["source_admin_id"],
"company_id": row["company_id"], "user": {
"id": row["id"], "username": row["username"], "source_account": row["source_account"],
"source_name": row["source_name"], "active": bool(row["active"]),
"role": {"code": row["role_code"], "name": row["role_name"]},
"permissions": assigned if row["active"] else []}}
return {"version": 1, "client_id": "kefu", "generated_at": now, "subject": subject,
"revision": _revision(db, row), "roles": roles, "catalog": {"permissions": catalog}}
class GroupAuthorization:
def __init__(self, database, clock: Callable[[], float] | None = None,
max_nonces: int = MAX_NONCES):
self.database = database
self.clock = clock or time.time
self.max_nonces = max_nonces
with database.connect() as db:
db.execute("BEGIN IMMEDIATE")
for statement in _SCHEMA:
db.execute(statement)
db.commit()
def _authenticate(self, body: bytes, headers, path: str, now: int) -> tuple[bytes, str]:
key = group_access.load_key()
timestamp = headers.get("x-group-timestamp", "")
nonce = headers.get("x-group-nonce", "")
signature = headers.get("x-group-signature", "")
if (not group_access._TIMESTAMP.fullmatch(timestamp)
or not group_access._HEX_32.fullmatch(nonce)
or not group_access._HEX_64.fullmatch(signature)
or abs(int(timestamp) - now) > WINDOW_SECONDS
or not hmac.compare_digest(request_signature(key, body, timestamp, nonce, path), signature)):
raise AuthorizationError(401, "集团服务认证无效")
return key, nonce
def _reserve_nonce(self, nonce: str, expires: int, now: int) -> None:
# A separate durable transaction consumes even requests whose domain
# transaction rolls back. Multiple instances/workers share this store.
with self.database.connect() as db:
db.execute("BEGIN IMMEDIATE")
db.execute("DELETE FROM group_authorization_nonces WHERE expires_at < ?", (now,))
if db.execute("SELECT 1 FROM group_authorization_nonces WHERE nonce=?", (nonce,)).fetchone():
raise AuthorizationError(401, "集团服务认证无效")
if db.execute("SELECT COUNT(*) FROM group_authorization_nonces").fetchone()[0] >= self.max_nonces:
raise _unavailable()
db.execute("INSERT INTO group_authorization_nonces(nonce,expires_at) VALUES (?,?)", (nonce, expires))
db.commit()
def _private_role(self, db, user_id: int, permissions: list[str], now: int) -> str:
granted = sorted(permissions)
digest = hashlib.sha256(_encode(granted)).hexdigest()
for row in db.execute("""SELECT p.role_code FROM group_authorization_private_roles p
JOIN roles r ON r.code=p.role_code WHERE p.user_id=? AND p.permissions_digest=?
AND NOT EXISTS(SELECT 1 FROM users u WHERE u.role=p.role_code AND u.id!=?)
ORDER BY p.role_code""", (user_id, digest, user_id)):
# Native KF admins can edit roles: never trust only the stored hash.
if _assigned(db, row["role_code"]) == granted:
return row["role_code"]
code = f"zyt_group_u{user_id}_{digest[:32]}"
while db.execute("SELECT 1 FROM roles WHERE code=?", (code,)).fetchone():
code = f"zyt_group_u{user_id}_{digest[:32]}_{secrets.token_hex(4)}"
import admin_backend
stamp = admin_backend.now_text()
db.execute("""INSERT INTO roles(code,name,builtin,created_at,updated_at)
VALUES (?,?,0,?,?)""", (code, f"集团个人授权(用户 {user_id})", stamp, stamp))
# Use the native grant validator but only ever on a newly-created role.
rbac.grant(db, code, granted, stamp)
db.execute("""INSERT INTO group_authorization_private_roles
(role_code,user_id,permissions_digest,created_at) VALUES (?,?,?,?)""", (code, user_id, digest, now))
return code
def _process(self, body: bytes, request: dict[str, Any], path: str, now: int) -> bytes:
with self.database.connect() as db:
db.execute("BEGIN IMMEDIATE" if path == UPDATE_PATH else "BEGIN")
row = _binding(db, request)
if path == OPTIONS_PATH:
return _encode(_result(db, row, now))
body_digest = hashlib.sha256(body).hexdigest()
previous = db.execute("SELECT body_digest,result_json FROM group_authorization_requests WHERE request_id=?",
(request["request_id"],)).fetchone()
if previous is not None:
if not hmac.compare_digest(previous["body_digest"], body_digest):
raise AuthorizationError(409, "客服授权请求编号已使用,请重新提交")
cached = bytes(previous["result_json"])
if len(cached) > MAX_RESPONSE_BYTES:
raise _unavailable()
if json.loads(cached)["subject"]["user"]["id"] != row["id"]:
raise AuthorizationError(409, "客服绑定身份已变化,请重新加载")
return cached
before_revision = _revision(db, row)
if not hmac.compare_digest(before_revision, request["expected_revision"]):
raise AuthorizationError(409, "客服授权已变化,请重新加载后保存")
selected = db.execute("SELECT code FROM roles WHERE code=?", (request["role_code"],)).fetchone()
private = db.execute("SELECT user_id FROM group_authorization_private_roles WHERE role_code=?",
(request["role_code"],)).fetchone()
if selected is None or (private is not None and private["user_id"] != row["id"]):
raise AuthorizationError(422, "客服角色无效,请重新加载")
permissions = request["permissions"]
if permissions is not None:
known = {item[0] for item in db.execute("SELECT code FROM permissions")}
if len(known) > MAX_PERMISSIONS:
raise _unavailable()
if not set(permissions) <= known:
raise AuthorizationError(422, "客服权限无效,请重新加载")
# Preserve the existing KF invariant: one active native admin role.
# Customizing admin creates an independent role and is a demotion.
demotes_admin = permissions is not None or request["role_code"] != "admin"
if row["active"] and row["role_code"] == "admin" and demotes_admin:
count = db.execute("SELECT COUNT(*) FROM users WHERE role='admin' AND active=1").fetchone()[0]
if count <= 1:
raise AuthorizationError(409, "系统至少需要保留一个启用的管理员")
role_code = (request["role_code"] if permissions is None else
self._private_role(db, row["id"], permissions, now))
import admin_backend
db.execute("UPDATE users SET role=?,updated_at=? WHERE id=?",
(role_code, admin_backend.now_text(), row["id"]))
updated = _binding(db, request)
result = _result(db, updated, now)
raw = _encode(result) # Bound the full DTO before anything commits.
detail = {"provider": "zyt_group", "gid": request["gid"],
"source_admin_id": request["source_admin_id"], "company_id": request["company_id"],
"actor_source_admin_id": request["actor_source_admin_id"], "target_user_id": row["id"],
"request_id": request["request_id"], "before_role": row["role_code"], "after_role": role_code,
"before_permissions": _assigned(db, row["role_code"]),
"after_permissions": _assigned(db, role_code),
"before_revision": before_revision, "after_revision": result["revision"]}
db.execute("""INSERT INTO group_authorization_audit
(request_id,actor_source_admin_id,target_user_id,detail_json,created_at) VALUES (?,?,?,?,?)""",
(request["request_id"], request["actor_source_admin_id"], row["id"],
_encode(detail).decode("utf-8"), now))
# Group actor IDs belong to another identity domain; do not forge a
# local actor FK. The full transactional audit above is authoritative.
self.database._audit(db, None, "group.authorization.update", json.dumps({
"request_id": request["request_id"], "actor_source_admin_id": request["actor_source_admin_id"],
"gid": request["gid"], "source_admin_id": request["source_admin_id"],
"company_id": request["company_id"], "target_user_id": row["id"]}, separators=(",", ":")), "")
db.execute("""INSERT INTO group_authorization_requests
(request_id,body_digest,result_json,created_at) VALUES (?,?,?,?)""",
(request["request_id"], body_digest, raw, now))
db.commit()
return raw
def respond(self, body: bytes, headers, path: str, *, query: bool = False) -> tuple[int, bytes, dict[str, str]]:
if path not in {OPTIONS_PATH, UPDATE_PATH}:
raise AuthorizationError(401, "集团服务认证无效")
if len(body) > MAX_REQUEST_BYTES:
raise AuthorizationError(413, "客服授权请求过大")
now = int(self.clock())
key, nonce = self._authenticate(body, headers, path, now)
try:
self._reserve_nonce(nonce, int(headers["x-group-timestamp"]) + WINDOW_SECONDS, now)
if query:
raise AuthorizationError(422, "客服授权请求无效")
request = parse_request(body, path)
raw = self._process(body, request, path, now)
status = 200
except group_access.AccessError as exc:
if exc.status == 401:
raise
status = exc.status
raw = _encode({"version": 1, "client_id": "kefu", "generated_at": now,
"error": {"code": status, "message": str(exc)}})
except Exception:
status = 503
raw = _encode({"version": 1, "client_id": "kefu", "generated_at": now,
"error": {"code": status, "message": "客服授权服务暂不可用"}})
return status, raw, {"Cache-Control": "no-store",
"X-Group-Response-Signature": group_access.response_signature(key, raw, nonce)}
async def handle_request(request, service: GroupAuthorization, path: str):
from fastapi.responses import JSONResponse, Response
try:
for name in ("x-group-timestamp", "x-group-nonce", "x-group-signature"):
if len(request.headers.getlist(name)) != 1:
raise AuthorizationError(401, "集团服务认证无效")
raw = bytearray()
async for chunk in request.stream():
if len(raw) + len(chunk) > MAX_REQUEST_BYTES:
raise AuthorizationError(413, "客服授权请求过大")
raw.extend(chunk)
import asyncio
status, body, headers = await asyncio.to_thread(service.respond, bytes(raw), request.headers,
path, query=bool(request.url.query))
return Response(content=body, status_code=status, media_type="application/json", headers=headers)
except group_access.AccessError as exc:
return JSONResponse(status_code=exc.status, content={"detail": str(exc)}, headers={"Cache-Control": "no-store"})
+38 -6
View File
@@ -77,6 +77,19 @@ class GroupSSOConfig:
"redirect_uri": self.redirect_uri}
@dataclass(frozen=True)
class GroupPasswordConfig:
"""Server-only direct login; no browser entry or callback is required."""
group_api_url: str
@classmethod
def from_env(cls) -> GroupPasswordConfig | None:
try:
return cls(_base_url(os.environ.get("ZYT_GROUP_API_URL", "")))
except ValueError:
return None
class _NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
@@ -103,12 +116,31 @@ def validate_identity(value: Any) -> dict[str, Any]:
def exchange(config: GroupSSOConfig, code: str, code_verifier: str) -> dict[str, Any]:
if not CODE_PATTERN.fullmatch(code) or not VERIFIER_PATTERN.fullmatch(code_verifier):
raise GroupSSOError("集团登录凭据无效,请重新登录")
payload = json.dumps({"client_id": CLIENT_ID, "redirect_uri": config.redirect_uri,
"code": code, "code_verifier": code_verifier}).encode("utf-8")
request = urllib.request.Request(config.group_api_url + "/group/sso/exchange",
data=payload, method="POST",
headers={"Content-Type": "application/json", "Accept": "application/json"})
return _post_identity(config.group_api_url, "/group/sso/exchange", {
"client_id": CLIENT_ID, "redirect_uri": config.redirect_uri,
"code": code, "code_verifier": code_verifier,
})
def password_login(config: GroupPasswordConfig, account: str, password: str) -> dict[str, Any]:
if (not isinstance(account, str) or not account.strip() or len(account) > 200
or any(ord(c) < 32 or ord(c) == 127 for c in account)
or not isinstance(password, str) or not password or len(password) > 4096):
raise GroupSSOError("用户名或密码不正确,或集团登录暂不可用")
# Credentials exist only in this request. Neither source tokens nor roles are
# accepted, persisted, or returned by the strict identity projection.
return _post_identity(config.group_api_url, "/group/kefu/login", {
"client_id": CLIENT_ID, "account": account, "password": password,
})
def _post_identity(api_url: str, path: str, payload: dict[str, str]) -> dict[str, Any]:
try:
# Revalidate even an explicitly constructed config. The path is supplied
# only by the two fixed call sites, never from the login request.
request = urllib.request.Request(_base_url(api_url) + path,
data=json.dumps(payload).encode("utf-8"), method="POST",
headers={"Content-Type": "application/json", "Accept": "application/json"})
# Neither proxy environment nor a remote redirect may change the recipient.
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), _NoRedirect())
with opener.open(request, timeout=10) as response:
@@ -122,4 +154,4 @@ def exchange(config: GroupSSOConfig, code: str, code_verifier: str) -> dict[str,
raise ValueError("Invalid envelope")
return validate_identity(result.get("data"))
except Exception:
raise GroupSSOError("集团登录失败或已过期,请重新登录") from None
raise GroupSSOError("用户名或密码不正确,或集团登录暂不可用") from None
+341
View File
@@ -0,0 +1,341 @@
"""Real FastAPI metadata checks using copied public modules and temporary SQLite.
Run this file with C:/kefu/wechat_rpa/.venv/Scripts/python.exe. No deployed
configuration, database, existing key or server is read or started.
"""
from __future__ import annotations
import atexit
import faulthandler
import hashlib
import hmac
import json
import os
import secrets
import shutil
import sys
import tempfile
import unittest
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from unittest import mock
MIRROR = Path(__file__).resolve().parent
PROJECT = MIRROR.parents[2]
SOURCE = Path(os.environ.get("IAM_KEFU_SOURCE_ROOT", "C:/kefu")) / "wechat_rpa"
TEST_RUNTIME = PROJECT / ".test-runtime"
TEST_RUNTIME.mkdir(parents=True, exist_ok=True)
RUNTIME = tempfile.TemporaryDirectory(prefix="iam-kefu-metadata-", dir=TEST_RUNTIME)
ISOLATED = Path(RUNTIME.name)
def cleanup_runtime():
os.chdir(TEST_RUNTIME)
RUNTIME.cleanup()
atexit.register(cleanup_runtime)
# Only .py public sources: never .env, JSON settings, DB or an existing key.
for source in SOURCE.glob("*.py"):
shutil.copyfile(source, ISOLATED / source.name)
for name in ("admin_api.py", "admin_backend.py", "rbac.py", "group_sso.py", "group_access.py", "group_authorization.py"):
shutil.copyfile(MIRROR / name, ISOLATED / name)
os.chdir(ISOLATED)
RUNTIME_ENV = {key: value for key, value in os.environ.items()
if key.upper() in {"SYSTEMROOT", "WINDIR", "TEMP", "TMP"}}
os.environ.clear()
os.environ.update(RUNTIME_ENV)
sys.path.insert(0, str(ISOLATED))
from fastapi.testclient import TestClient
import admin_api
import admin_backend as backend
import group_access
import rbac
KEY = bytes.fromhex("12" * 32)
NOW = 1_800_000_000
URL = "/api/v2/group/access-snapshot"
GID = "gid_" + "a" * 32
OTHER_GID = "gid_" + "b" * 32
PASSWORD = "Fixture-Metadata-Password-2026"
class GroupAccessTests(unittest.TestCase):
def setUp(self):
self.directory = tempfile.TemporaryDirectory(dir=ISOLATED)
self.addCleanup(self.directory.cleanup)
self.root = Path(self.directory.name)
self.key_file = self.root / "metadata.key"
self.key_file.write_text(KEY.hex(), encoding="ascii")
self.env = mock.patch.dict(os.environ, {"ZYT_GROUP_METADATA_KEY_FILE": str(self.key_file)}, clear=True)
self.env.start()
self.addCleanup(self.env.stop)
self.db = backend.Database(self.root / "fixture.sqlite")
self.clock = mock.Mock(return_value=NOW)
actual_service = group_access.AccessSnapshot
self.service = actual_service(self.db, clock=self.clock)
with mock.patch.object(backend, "load_initial_config", return_value=backend.CONFIG_DEFAULTS.copy()):
self.db.initialize(PASSWORD)
with mock.patch.object(group_access, "AccessSnapshot", return_value=self.service):
self.client = TestClient(admin_api.create_app(self.db.path))
self.addCleanup(self.client.close)
def body(self, gids=None, catalog=False):
return {"client_id": "kefu", "gids": [GID] if gids is None else gids,
"include_catalog": catalog}
def raw(self, value):
return json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
def headers(self, raw, *, nonce=None, timestamp=None, key=KEY, path="/group/access-snapshot"):
nonce = nonce or secrets.token_hex(16)
timestamp = str(NOW if timestamp is None else timestamp)
message = f"POST\n{path}\n{timestamp}\n{nonce}\n{hashlib.sha256(raw).hexdigest()}"
return {"Content-Type": "application/json", "X-Group-Timestamp": timestamp,
"X-Group-Nonce": nonce,
"X-Group-Signature": hmac.new(key, message.encode("ascii"), hashlib.sha256).hexdigest()}
def fetch(self, body=None, *, raw=None, **signing):
raw = self.raw(self.body() if body is None else body) if raw is None else raw
headers = self.headers(raw, **signing)
response = self.client.post(URL, content=raw, headers=headers)
if response.status_code == 200:
signed = f"response\n{headers['X-Group-Nonce']}\n{hashlib.sha256(response.content).hexdigest()}"
expected = hmac.new(KEY, signed.encode("ascii"), hashlib.sha256).hexdigest()
self.assertEqual(response.headers["X-Group-Response-Signature"], expected)
self.assertEqual(response.headers.get("Cache-Control"), "no-store")
return response
def enroll(self, gid=GID, source_id=7, account="same-account", name="合成集团用户"):
return self.db.group_sso_user({"client_id": "kefu", "gid": gid,
"source_admin_id": source_id, "account": account, "name": name, "company_id": 2})
def fingerprint(self):
tables = ("users", "roles", "role_permissions", "auth_tokens", "permissions", "group_sso_bindings")
with self.db.connect() as db:
return {table: [tuple(row) for row in db.execute(f"SELECT * FROM {table} ORDER BY rowid")]
for table in tables}
def test_missing_subject_is_not_created_and_names_never_match(self):
self.db.create_user("same-account", PASSWORD, "admin", 1, "fixture")
before = self.fingerprint()
with mock.patch.object(backend.Database, "group_sso_user", side_effect=AssertionError("No enrollment")), \
mock.patch.object(backend.Database, "sync_code_permissions", side_effect=AssertionError("No sync")), \
mock.patch.object(rbac, "seed", side_effect=AssertionError("No seed")), \
mock.patch.object(rbac, "grant", side_effect=AssertionError("No grant")):
response = self.fetch(self.body(catalog=True))
self.assertEqual(response.status_code, 200, response.text)
self.assertEqual(response.json()["subjects"], [{"gid": GID, "found": False}])
self.assertEqual(before, self.fingerprint())
def test_exact_projection_and_signed_raw_utf8_response(self):
self.enroll()
result = self.fetch(self.body(catalog=True))
self.assertEqual(result.status_code, 200, result.text)
data = result.json()
self.assertEqual(set(data), {"version", "client_id", "generated_at", "subjects", "catalog"})
self.assertEqual((data["version"], data["client_id"], data["generated_at"]), (1, "kefu", NOW))
subject = data["subjects"][0]
self.assertEqual(set(subject), {"gid", "found", "source_admin_id", "company_id", "user"})
self.assertEqual(subject["source_admin_id"], 7)
user = subject["user"]
self.assertEqual(set(user), {"id", "username", "source_account", "source_name", "active", "role", "permissions"})
self.assertEqual(set(user["role"]), {"code", "name"})
self.assertEqual(user["role"]["code"], "zyt_basic")
self.assertEqual(user["permissions"], [])
self.assertEqual(user["source_name"], "合成集团用户")
for item in data["catalog"]["permissions"]:
self.assertEqual(set(item), {"code", "name", "group", "page", "kind", "source", "enabled", "sort"})
self.assertIn(item["kind"], {"page", "action"})
self.assertIs(type(item["enabled"]), bool)
for secret in ("password_salt", "password_digest", "auth_tokens", "link_url", "user_count", str(self.key_file), KEY.hex()):
self.assertNotIn(secret, result.text)
def test_binding_identity_and_requested_order_are_exact(self):
first = self.enroll()
second = self.enroll(OTHER_GID, 8, name="同名")
self.db.update_user(int(first["id"]), "operator", True, "", 1, "fixture")
self.db.update_user(int(second["id"]), "viewer", True, "", 1, "fixture")
third = "gid_" + "c" * 32
response = self.fetch(self.body([OTHER_GID, third, GID]))
self.assertEqual(response.status_code, 200)
subjects = response.json()["subjects"]
self.assertEqual([row["gid"] for row in subjects], [OTHER_GID, third, GID])
self.assertEqual(subjects[1], {"gid": third, "found": False})
self.assertEqual(subjects[0]["user"]["role"]["code"], "viewer")
self.assertEqual(subjects[2]["user"]["role"]["code"], "operator")
with self.db.connect() as db:
db.execute("UPDATE group_sso_bindings SET provider='other_provider' WHERE gid=?", (GID,))
db.commit()
self.assertEqual(self.fetch().json()["subjects"], [{"gid": GID, "found": False}])
def test_role_and_active_changes_are_live_and_effective_reader_is_used(self):
user = self.enroll()
user_id = int(user["id"])
self.assertEqual(self.fetch().json()["subjects"][0]["user"]["permissions"], [])
self.db.update_user(user_id, "operator", True, "", 1, "fixture")
expected = sorted(self.db.permissions_for_user(user_id))
self.assertTrue(expected)
with mock.patch.object(backend.Database, "permissions_for_user", wraps=self.db.permissions_for_user) as effective:
actual = self.fetch().json()["subjects"][0]["user"]
effective.assert_called_once_with(user_id)
self.assertEqual(actual["permissions"], expected)
self.assertEqual(actual["role"]["code"], "operator")
self.db.update_user(user_id, "operator", False, "", 1, "fixture")
actual = self.fetch().json()["subjects"][0]["user"]
self.assertFalse(actual["active"])
self.assertEqual(actual["permissions"], [])
self.assertEqual(actual["role"]["code"], "operator")
def test_dynamic_catalog_add_update_delete_and_link_projection(self):
payload = {"code": "fixture:dynamic", "name": "动态测试", "group_name": "合成菜单",
"page_name": "合成页面", "kind": "action", "enabled": True, "sort_order": 14}
self.db.create_permission(payload, 1, "fixture")
catalog = lambda: {row["code"]: row for row in self.fetch(self.body([], True)).json()["catalog"]["permissions"]}
self.assertEqual(catalog()[payload["code"]]["name"], "动态测试")
self.db.update_permission(payload["code"], {**payload, "name": "已更新", "enabled": False}, 1, "fixture")
changed = catalog()[payload["code"]]
self.assertEqual(changed["name"], "已更新")
self.assertFalse(changed["enabled"])
self.db.create_permission({**payload, "code": "link:fixture", "kind": "link",
"link_url": "https://example.invalid/private?secret=fixture", "link_mode": "tab"}, 1, "fixture")
link = catalog()["link:fixture"]
self.assertEqual(link["kind"], "page")
self.assertNotIn("link_url", link)
self.db.delete_permission(payload["code"], 1, "fixture")
self.assertNotIn(payload["code"], catalog())
def test_reading_does_not_change_users_roles_grants_tokens_or_catalog(self):
self.enroll()
self.db.create_token(1, "api", "fixture", 3600)
before = self.fingerprint()
self.assertEqual(self.fetch(self.body(catalog=True)).status_code, 200)
self.assertEqual(self.fetch(self.body([], True)).status_code, 200)
self.assertEqual(self.fetch(self.body([OTHER_GID], False)).status_code, 200)
self.assertEqual(before, self.fingerprint())
def test_no_catalog_requested_does_not_read_it(self):
with mock.patch.object(backend.Database, "permission_catalog", side_effect=AssertionError("No catalog")):
response = self.fetch(self.body([], False))
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["subjects"], [])
self.assertIsNone(response.json()["catalog"])
def test_user_bearer_never_substitutes_for_machine_authentication(self):
token, _ = self.db.create_token(1, "api", "fixture", 3600)
response = self.client.post(URL, json=self.body(), headers={"Authorization": "Bearer " + token})
self.assertEqual(response.status_code, 401)
self.assertEqual(response.headers["Cache-Control"], "no-store")
def test_bad_signature_and_body_or_path_tampering_are_rejected_before_parsing(self):
raw = self.raw(self.body())
for key, path, changed in ((b"x" * 32, "/group/access-snapshot", raw),
(KEY, "/api/v2/group/access-snapshot", raw),
(KEY, "/group/access-snapshot", b"not-json")):
with self.subTest(path=path, body=changed):
headers = self.headers(raw, key=key, path=path)
with mock.patch.object(group_access, "parse_request", side_effect=AssertionError("Not authenticated")):
response = self.client.post(URL, content=changed, headers=headers)
self.assertEqual(response.status_code, 401)
def test_clock_window_and_replay(self):
for offset in (-91, 91):
self.assertEqual(self.fetch(timestamp=NOW + offset).status_code, 401)
for offset in (-90, 90):
self.assertEqual(self.fetch(timestamp=NOW + offset).status_code, 200)
nonce = "d" * 32
self.assertEqual(self.fetch(nonce=nonce).status_code, 200)
self.assertEqual(self.fetch(nonce=nonce).status_code, 401)
self.clock.return_value = NOW + 91
self.assertEqual(self.fetch(nonce=nonce, timestamp=NOW + 91).status_code, 200)
def test_replay_cache_full_fails_closed_and_does_not_evict_live_nonces(self):
self.service.max_nonces = 1
self.assertEqual(self.fetch(nonce="1" * 32).status_code, 200)
self.assertEqual(self.fetch(nonce="2" * 32).status_code, 503)
self.assertEqual(self.fetch(nonce="1" * 32).status_code, 401)
def test_concurrent_replay_consumes_nonce_once(self):
raw = self.raw(self.body([], False))
headers = self.headers(raw, nonce="e" * 32)
with ThreadPoolExecutor(max_workers=4) as pool:
statuses = list(pool.map(lambda _: self.client.post(URL, content=raw, headers=headers).status_code,
range(8)))
self.assertEqual(statuses.count(200), 1)
self.assertEqual(statuses.count(401), 7)
def test_invalid_and_missing_key_configuration_fail_closed(self):
for value in ("", str(self.root / "missing.key"), "relative.key"):
with self.subTest(value=value), mock.patch.dict(os.environ, {"ZYT_GROUP_METADATA_KEY_FILE": value}):
self.assertEqual(self.fetch().status_code, 503)
for raw in (b"x" * 64, b"1" * 63, b"1" * 65, b"1" * 64 + b"\n"):
self.key_file.write_bytes(raw)
response = self.fetch()
self.assertEqual(response.status_code, 503)
self.assertNotIn(str(self.key_file), response.text)
def test_public_static_key_location_is_rejected(self):
directory = self.root / "admin-web/apps/web-antd/dist"
directory.mkdir(parents=True, exist_ok=True)
key = directory / ("fixture-" + secrets.token_hex(8) + ".key")
key.write_text(KEY.hex(), encoding="ascii")
with mock.patch.object(group_access, "__file__", str(self.root / "wechat_rpa/group_access.py")), \
mock.patch.dict(os.environ, {"ZYT_GROUP_METADATA_KEY_FILE": str(key)}):
self.assertEqual(self.fetch().status_code, 503)
def test_strict_body_shapes_and_no_duplicate_json_keys(self):
bad = [[], {}, {**self.body(), "role": "admin"}, {**self.body(), "client_id": "hospital"},
{**self.body(), "include_catalog": 1}, {**self.body(), "gids": "bad"},
self.body([GID, GID]), self.body(["gid_" + "A" * 32]), self.body([True]),
self.body(["gid_" + f"{number:032x}" for number in range(101)])]
for value in bad:
with self.subTest(body=value):
self.assertEqual(self.fetch(value).status_code, 422)
for raw in (b'{"client_id":"kefu","client_id":"kefu","gids":[],"include_catalog":false}',
b'{"client_id":"kefu","gids":[],"include_catalog":NaN}', b"\xff", b"not-json"):
self.assertEqual(self.fetch(raw=raw).status_code, 422)
def test_request_and_response_bounds(self):
self.assertEqual(self.fetch(raw=b" " * (16 * 1024 + 1)).status_code, 413)
gids = ["gid_" + f"{number:032x}" for number in range(100)]
self.assertEqual(len(self.fetch(self.body(gids)).json()["subjects"]), 100)
definition = {"code": "test:one", "name": "one", "group_name": "g", "page_name": "p",
"kind": "action", "source": "custom", "enabled": True, "sort_order": 0}
with mock.patch.object(backend.Database, "permission_catalog", return_value=[definition] * 2001):
self.assertEqual(self.fetch(self.body([], True)).status_code, 503)
self.enroll()
with mock.patch.object(backend.Database, "permissions_for_user", return_value={f"test:{i}" for i in range(2001)}):
self.assertEqual(self.fetch().status_code, 503)
with mock.patch.object(backend.Database, "permission_catalog", return_value=[{**definition, "name": "a" * (4 * 1024 * 1024)}]):
self.assertEqual(self.fetch(self.body([], True)).status_code, 503)
def test_duplicate_security_headers_and_query_are_rejected(self):
raw = self.raw(self.body())
headers = list(self.headers(raw).items())
headers.append(("X-Group-Nonce", "f" * 32))
self.assertEqual(self.client.post(URL, content=raw, headers=headers).status_code, 401)
response = self.client.post(URL + "?gid=extra", content=raw, headers=self.headers(raw))
self.assertEqual(response.status_code, 422)
def test_response_signature_changes_with_payload_and_nonce(self):
self.enroll()
first = self.fetch(nonce="1" * 32)
second = self.fetch(nonce="2" * 32)
self.assertEqual(first.content, second.content)
self.assertNotEqual(first.headers["X-Group-Response-Signature"], second.headers["X-Group-Response-Signature"])
self.db.update_user(int(first.json()["subjects"][0]["user"]["id"]), "operator", True, "", 1, "fixture")
third = self.fetch(nonce="3" * 32)
self.assertNotEqual(first.content, third.content)
if __name__ == "__main__":
# A restricted Windows socketpair can hang Python's test event loop before
# the endpoint runs. Bound the process instead of leaving a test stranded.
faulthandler.dump_traceback_later(60, exit=True)
try:
unittest.main(verbosity=2)
finally:
faulthandler.cancel_dump_traceback_later()
+484
View File
@@ -0,0 +1,484 @@
"""Real FastAPI + isolated public Python copies + synthetic SQLite/key only."""
from __future__ import annotations
import atexit
import faulthandler
import hashlib
import hmac
import json
import os
import secrets
import shutil
import sys
import tempfile
import unittest
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from unittest import mock
MIRROR = Path(__file__).resolve().parent
SOURCE = Path(os.environ.get("IAM_KEFU_SOURCE_ROOT", "C:/kefu")) / "wechat_rpa"
TEST_RUNTIME = MIRROR.parents[2] / ".test-runtime"
TEST_RUNTIME.mkdir(parents=True, exist_ok=True)
RUNTIME = tempfile.TemporaryDirectory(prefix="iam-kefu-authorization-", dir=TEST_RUNTIME)
ISOLATED = Path(RUNTIME.name)
def cleanup_runtime():
os.chdir(TEST_RUNTIME)
RUNTIME.cleanup()
atexit.register(cleanup_runtime)
# Copy no settings, .env, database, production key, or assets from the real app.
for source in SOURCE.glob("*.py"):
shutil.copyfile(source, ISOLATED / source.name)
for name in ("admin_api.py", "admin_backend.py", "rbac.py", "group_sso.py", "group_access.py", "group_authorization.py"):
shutil.copyfile(MIRROR / name, ISOLATED / name)
os.chdir(ISOLATED)
RUNTIME_ENV = {key: value for key, value in os.environ.items()
if key.upper() in {"SYSTEMROOT", "WINDIR", "TEMP", "TMP"}}
os.environ.clear()
os.environ.update(RUNTIME_ENV)
sys.path.insert(0, str(ISOLATED))
from fastapi.testclient import TestClient
import admin_api
import admin_backend as backend
import group_access
import group_authorization as authorization
import rbac
KEY = bytes.fromhex("34" * 32)
NOW = 1_800_000_000
GID = "gid_" + "a" * 32
OTHER_GID = "gid_" + "b" * 32
PASSWORD = "Fixture-Authorization-Password-2026"
class GroupAuthorizationTests(unittest.TestCase):
def setUp(self):
self.directory = tempfile.TemporaryDirectory(dir=ISOLATED)
self.addCleanup(self.directory.cleanup)
self.root = Path(self.directory.name)
self.key_file = self.root / "synthetic.key"
self.key_file.write_text(KEY.hex(), encoding="ascii")
self.env = mock.patch.dict(os.environ, {"ZYT_GROUP_METADATA_KEY_FILE": str(self.key_file)}, clear=True)
self.env.start()
self.addCleanup(self.env.stop)
self.db = backend.Database(self.root / "synthetic.sqlite")
self.clock = mock.Mock(return_value=NOW)
with mock.patch.object(backend, "load_initial_config", return_value=backend.CONFIG_DEFAULTS.copy()):
self.db.initialize(PASSWORD)
self.service = authorization.GroupAuthorization(self.db, clock=self.clock)
with mock.patch.object(authorization, "GroupAuthorization", return_value=self.service):
self.client = TestClient(admin_api.create_app(self.db.path))
self.addCleanup(self.client.close)
def identity(self, gid=GID, source_id=7):
return {"client_id": "kefu", "gid": gid, "source_admin_id": source_id, "company_id": 2}
def enroll(self, gid=GID, source_id=7):
return self.db.group_sso_user({**self.identity(gid, source_id),
"account": "same-account", "name": "合成集团用户"})
def raw(self, value):
return json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
def headers(self, raw, path, *, nonce=None, timestamp=None, key=KEY):
nonce = nonce or secrets.token_hex(16)
timestamp = str(NOW if timestamp is None else timestamp)
message = f"POST\n{path}\n{timestamp}\n{nonce}\n{hashlib.sha256(raw).hexdigest()}"
return {"Content-Type": "application/json", "X-Group-Timestamp": timestamp,
"X-Group-Nonce": nonce,
"X-Group-Signature": hmac.new(key, message.encode("ascii"), hashlib.sha256).hexdigest()}
def fetch(self, body, path=authorization.OPTIONS_PATH, *, raw=None, query="", **signing):
raw = self.raw(body) if raw is None else raw
headers = self.headers(raw, path, **signing)
response = self.client.post("/api/v2" + path + query, content=raw, headers=headers)
self.assertEqual(response.headers.get("Cache-Control"), "no-store")
if response.status_code not in {401, 413} and "X-Group-Response-Signature" in response.headers:
self.assertEqual(response.headers["X-Group-Response-Signature"],
group_access.response_signature(KEY, response.content, headers["X-Group-Nonce"]))
if response.status_code != 200 and response.status_code not in {401, 413}:
self.assertEqual(response.json(), {"version": 1, "client_id": "kefu", "generated_at": int(self.clock()),
"error": {"code": response.status_code, "message": response.json()["error"]["message"]}})
self.assertIn("X-Group-Response-Signature", response.headers)
return response
def options(self, identity=None):
result = self.fetch(identity or self.identity())
self.assertEqual(result.status_code, 200, result.text)
return result.json()
def mutation(self, revision=None, role="operator", permissions=None, identity=None, **changes):
return {**(identity or self.identity()), "actor_source_admin_id": 9,
"request_id": secrets.token_hex(16), "expected_revision": revision or self.options(identity)["revision"],
"role_code": role, "permissions": permissions, **changes}
def update(self, body=None, **changes):
return self.fetch(body or self.mutation(**changes), authorization.UPDATE_PATH)
def fingerprint(self):
tables = ("users", "roles", "role_permissions", "auth_tokens", "permissions", "group_sso_bindings",
"audit_log", "group_authorization_audit", "group_authorization_private_roles", "group_authorization_requests")
with self.db.connect() as db:
return {table: [tuple(row) for row in db.execute(f"SELECT * FROM {table} ORDER BY rowid")]
for table in tables}
def token_headers(self, user_id):
token, _ = self.db.create_token(user_id, "api", "existing-synthetic-session", 3600)
return {"Authorization": "Bearer " + token}
def test_options_exact_native_projection_is_readonly(self):
user = self.enroll()
self.db.update_user(user["id"], "operator", True, "", 1, "fixture")
with self.db.connect() as db:
db.execute("UPDATE permissions SET enabled=0 WHERE code='config:read'")
db.commit()
before = self.fingerprint()
data = self.options()
self.assertEqual(set(data), {"version", "client_id", "generated_at", "subject", "revision", "roles", "catalog"})
self.assertRegex(data["revision"], r"^[0-9a-f]{64}$")
subject = data["subject"]
self.assertEqual(set(subject), {"gid", "found", "source_admin_id", "company_id", "user"})
self.assertEqual(subject["gid"], GID)
self.assertEqual(subject["user"]["permissions"], sorted(self.db.permissions_for_user(user["id"])))
self.assertIn("config:read", subject["user"]["permissions"])
for role in data["roles"]:
self.assertEqual(set(role), {"code", "name", "permissions"})
for item in data["catalog"]["permissions"]:
self.assertEqual(set(item), {"code", "name", "group", "page", "kind", "source", "enabled", "sort"})
text = self.raw(data).decode("utf-8")
for forbidden in ("password_salt", "password_digest", "link_url", "user_count", str(self.key_file), KEY.hex()):
self.assertNotIn(forbidden, text)
self.assertEqual(before, self.fingerprint())
def test_role_assignment_existing_token_gains_and_loses_without_login(self):
user = self.enroll()
other = self.enroll(OTHER_GID, 8)
self.db.update_user(other["id"], "viewer", True, "", 1, "fixture")
headers = self.token_headers(user["id"])
other_headers = self.token_headers(other["id"])
initial = self.fingerprint()
other_before = self.client.get("/api/v2/me", headers=other_headers).json()
self.assertEqual(self.client.get("/api/v2/stats/model-calls", headers=headers).status_code, 403)
self.assertEqual(self.client.get("/api/v2/config", headers=headers).status_code, 403)
first = self.update()
self.assertEqual(first.status_code, 200, first.text)
self.assertEqual(first.json()["subject"]["user"]["role"]["code"], "operator")
self.assertEqual(self.client.get("/api/v2/stats/model-calls", headers=headers).status_code, 200)
self.assertEqual(self.client.get("/api/v2/config", headers=headers).status_code, 200)
self.assertEqual(self.client.get("/api/v2/me", headers=headers).json()["role"], "operator")
second = self.update(role="zyt_basic", revision=first.json()["revision"])
self.assertEqual(second.status_code, 200, second.text)
self.assertEqual(self.client.get("/api/v2/stats/model-calls", headers=headers).status_code, 403)
self.assertEqual(self.client.get("/api/v2/config", headers=headers).status_code, 403)
self.assertEqual(self.client.get("/api/v2/me", headers=headers).json()["permissions"], [])
self.assertEqual(self.client.get("/api/v2/me", headers=other_headers).json(), other_before)
final = self.fingerprint()
for table in ("roles", "role_permissions", "permissions", "group_sso_bindings"):
self.assertEqual(initial[table], final[table])
# session() updates last_used_at; identity, token/CSRF digests, expiry,
# credentials, status, password policy and all session rows are retained.
self.assertEqual([row[:-1] for row in initial["auth_tokens"]], [row[:-1] for row in final["auth_tokens"]])
for before, after in zip(initial["users"], final["users"]):
self.assertEqual(before[:-1], after[:-1]) # updated_at only may move
def test_explicit_permissions_are_private_empty_is_valid_and_other_users_unchanged(self):
user = self.enroll()
other = self.enroll(OTHER_GID, 8)
shared = self.fingerprint()["role_permissions"]
granted = ["config:read", "stats:read"]
result = self.update(permissions=granted)
self.assertEqual(result.status_code, 200, result.text)
first_role = result.json()["subject"]["user"]["role"]["code"]
self.assertNotEqual(first_role, "operator")
self.assertEqual(self.db.permissions_for_user(user["id"]), set(granted))
self.assertEqual(self.db.permissions_for_user(other["id"]), set())
with self.db.connect() as db:
unchanged = [tuple(row) for row in db.execute("SELECT * FROM role_permissions WHERE role_code!=? ORDER BY rowid", (first_role,))]
self.assertEqual(shared, unchanged)
second = self.update(role="operator", permissions=granted, identity=self.identity(OTHER_GID, 8))
self.assertEqual(second.status_code, 200, second.text)
self.assertNotEqual(first_role, second.json()["subject"]["user"]["role"]["code"])
self.assertNotIn(first_role, {role["code"] for role in second.json()["roles"]})
attempt = self.update(role=first_role, identity=self.identity(OTHER_GID, 8))
self.assertEqual(attempt.status_code, 422)
empty = self.update(permissions=[])
self.assertEqual(empty.status_code, 200, empty.text)
self.assertEqual(empty.json()["subject"]["user"]["permissions"], [])
self.assertEqual(self.db.permissions_for_user(other["id"]), set(granted))
def test_reuses_matching_private_role_but_forks_when_natively_shared_or_edited(self):
user = self.enroll()
other = self.enroll(OTHER_GID, 8)
first = self.update(permissions=["config:read"]).json()
role = first["subject"]["user"]["role"]["code"]
again = self.update(permissions=["config:read"]).json()
self.assertEqual(again["subject"]["user"]["role"]["code"], role)
self.db.update_user(other["id"], role, True, "", 1, "fixture")
shared_before = self.db.permissions_for_user(other["id"])
shared_fork = self.update(permissions=["config:read"]).json()["subject"]["user"]["role"]["code"]
self.assertNotEqual(shared_fork, role)
self.assertEqual(self.db.permissions_for_user(other["id"]), shared_before)
self.db.save_role(shared_fork, "KF 原生改名", ["stats:read"], 1, "fixture")
edited_fork = self.update(permissions=["config:read"]).json()["subject"]["user"]["role"]["code"]
self.assertNotIn(edited_fork, {role, shared_fork})
with self.db.connect() as db:
self.assertEqual(rbac.permissions_of(db, shared_fork), {"stats:read"})
def test_inactive_user_stays_inactive_and_revision_binds_assigned_permissions(self):
user = self.enroll()
self.db.update_user(user["id"], "viewer", False, "", 1, "fixture")
before = self.options()
result = self.update(revision=before["revision"], permissions=["config:read"])
self.assertEqual(result.status_code, 200, result.text)
self.assertFalse(result.json()["subject"]["user"]["active"])
self.assertEqual(result.json()["subject"]["user"]["permissions"], [])
role = result.json()["subject"]["user"]["role"]["code"]
self.assertEqual(next(item["permissions"] for item in result.json()["roles"] if item["code"] == role), ["config:read"])
self.assertNotEqual(before["revision"], result.json()["revision"])
def test_last_active_native_admin_is_preserved(self):
user = self.enroll()
self.db.update_user(user["id"], "admin", True, "", 1, "fixture")
with self.db.connect() as db:
db.execute("UPDATE users SET active=0 WHERE id=1")
db.commit()
before = self.fingerprint()
for permissions, role in ((None, "viewer"), ([], "admin"), (list(self.db.permissions_for_user(user["id"])), "admin")):
self.assertEqual(self.update(role=role, permissions=permissions).status_code, 409)
self.assertEqual(before, self.fingerprint())
self.assertEqual(self.update(role="admin").status_code, 200)
def test_missing_native_same_name_or_mismatched_binding_never_enrolls(self):
self.db.create_user("same-account", PASSWORD, "admin", 1, "fixture")
before = self.fingerprint()
with mock.patch.object(backend.Database, "group_sso_user", side_effect=AssertionError("No enrollment")):
self.assertEqual(self.fetch(self.identity()).status_code, 404)
self.assertEqual(self.update(self.mutation(revision="0" * 64)).status_code, 404)
self.assertEqual(before, self.fingerprint())
self.enroll()
before = self.fingerprint()
for changed in ({"company_id": 3}, {"source_admin_id": 8}):
identity = {**self.identity(), **changed}
self.assertEqual(self.fetch(identity).status_code, 409)
self.assertEqual(self.update(self.mutation(revision="0" * 64, identity=identity)).status_code, 409)
self.assertEqual(before, self.fingerprint())
with self.db.connect() as db:
db.execute("UPDATE group_sso_bindings SET provider='unrelated' WHERE gid=?", (GID,))
db.commit()
self.assertEqual(self.fetch(self.identity()).status_code, 404)
def test_unknown_role_permission_and_stale_revision_rollback(self):
self.enroll()
for changes, expected in (({"role": "missing-role"}, 422), ({"permissions": ["missing:permission"]}, 422),
({"revision": "0" * 64}, 409)):
body = self.mutation(**changes)
before = self.fingerprint()
self.assertEqual(self.update(body).status_code, expected)
self.assertEqual(before, self.fingerprint())
def test_revision_detects_native_role_permission_active_and_identity_changes(self):
user = self.enroll()
self.db.update_user(user["id"], "viewer", True, "", 1, "fixture")
sql_changes = [("UPDATE role_permissions SET permission_code='config:write' WHERE role_code='viewer' AND permission_code='config:read'", ()),
("UPDATE users SET active=0 WHERE id=?", (user["id"],)),
("UPDATE users SET role='operator' WHERE id=?", (user["id"],)),
("UPDATE group_sso_bindings SET source_name='更新姓名' WHERE gid=?", (GID,))]
for sql, parameters in sql_changes:
body = self.mutation()
with self.db.connect() as db:
db.execute(sql, parameters)
db.commit()
before = self.fingerprint()
self.assertEqual(self.update(body).status_code, 409)
self.assertEqual(before, self.fingerprint())
def test_idempotency_is_durable_exact_body_and_revalidates_binding(self):
self.enroll()
body = self.mutation(permissions=["stats:read"])
first = self.update(body)
self.assertEqual(first.status_code, 200, first.text)
fingerprint = self.fingerprint()
# Simulate another worker/restart using the same SQLite file.
restarted = authorization.GroupAuthorization(self.db, clock=self.clock)
self.service._reserve_nonce = restarted._reserve_nonce
second = self.update(body)
self.assertEqual(second.status_code, 200, second.text)
self.assertEqual(first.content, second.content)
self.assertNotEqual(first.headers["X-Group-Response-Signature"], second.headers["X-Group-Response-Signature"])
self.assertEqual(fingerprint, self.fingerprint())
for changed in ({"role_code": "viewer"}, {"actor_source_admin_id": 10}, {"permissions": []}):
self.assertEqual(self.update({**body, **changed}).status_code, 409)
self.assertEqual(fingerprint, self.fingerprint())
with self.db.connect() as db:
db.execute("UPDATE group_sso_bindings SET company_id=3 WHERE gid=?", (GID,))
db.commit()
self.assertEqual(self.update(body).status_code, 409)
def test_idempotent_success_never_crosses_a_reassigned_binding(self):
self.enroll()
replacement = self.enroll(OTHER_GID, 8)
body = self.mutation(permissions=["stats:read"])
self.assertEqual(self.update(body).status_code, 200)
with self.db.connect() as db:
db.execute("DELETE FROM group_sso_bindings WHERE gid=?", (GID,))
db.execute("UPDATE group_sso_bindings SET gid=?,source_admin_id=7 WHERE gid=?", (GID, OTHER_GID))
db.commit()
before = self.fingerprint()
self.assertEqual(self.update(body).status_code, 409)
self.assertEqual(self.db.permissions_for_user(replacement["id"]), set())
self.assertEqual(before, self.fingerprint())
def test_atomic_compare_and_swap_allows_only_one_concurrent_writer(self):
self.enroll()
revision = self.options()["revision"]
bodies = [self.mutation(revision=revision, role="operator"), self.mutation(revision=revision, role="viewer")]
with ThreadPoolExecutor(max_workers=2) as pool:
results = list(pool.map(self.update, bodies))
self.assertEqual(sorted(result.status_code for result in results), [200, 409])
with self.db.connect() as db:
self.assertEqual(db.execute("SELECT COUNT(*) FROM group_authorization_requests").fetchone()[0], 1)
self.assertEqual(db.execute("SELECT COUNT(*) FROM group_authorization_audit").fetchone()[0], 1)
def test_both_audits_are_transactional_and_preserve_asserted_actor(self):
user = self.enroll()
body = self.mutation(permissions=["stats:read"])
result = self.update(body)
self.assertEqual(result.status_code, 200, result.text)
with self.db.connect() as db:
audit = db.execute("SELECT * FROM group_authorization_audit").fetchone()
detail = json.loads(audit["detail_json"])
native = db.execute("SELECT user_id,detail FROM audit_log WHERE action='group.authorization.update'").fetchone()
self.assertEqual(audit["actor_source_admin_id"], 9)
self.assertEqual(audit["target_user_id"], user["id"])
self.assertEqual(detail["request_id"], body["request_id"])
self.assertEqual(detail["before_permissions"], [])
self.assertEqual(detail["after_permissions"], ["stats:read"])
self.assertEqual(detail["after_revision"], result.json()["revision"])
self.assertIsNone(native["user_id"])
self.assertEqual(json.loads(native["detail"])["actor_source_admin_id"], 9)
def test_audit_failure_rolls_back_role_grants_assignment_and_idempotency(self):
self.enroll()
for table in ("audit_log", "group_authorization_audit", "group_authorization_requests"):
body = self.mutation(permissions=["stats:read"])
with self.db.connect() as db:
db.execute(f"CREATE TRIGGER synthetic_audit_failure BEFORE INSERT ON {table} BEGIN SELECT RAISE(ABORT,'synthetic-secret-error'); END")
db.commit()
before = self.fingerprint()
response = self.update(body)
self.assertEqual(response.status_code, 503)
self.assertNotIn("synthetic-secret-error", response.text)
self.assertEqual(before, self.fingerprint())
with self.db.connect() as db:
db.execute("DROP TRIGGER synthetic_audit_failure")
db.commit()
def test_response_bound_failure_rolls_back_mutation(self):
self.enroll()
body = self.mutation(permissions=["stats:read"])
before = self.fingerprint()
with mock.patch.object(authorization, "MAX_RESPONSE_BYTES", 1024):
response = self.update(body)
self.assertEqual(response.status_code, 503)
self.assertEqual(before, self.fingerprint())
def test_bearer_is_never_machine_auth_and_wrong_path_or_signature_fail_before_parse(self):
self.enroll()
token, _ = self.db.create_token(1, "api", "fixture", 3600)
for path in (authorization.OPTIONS_PATH, authorization.UPDATE_PATH):
response = self.client.post("/api/v2" + path, json=self.identity(), headers={"Authorization": "Bearer " + token})
self.assertEqual(response.status_code, 401)
raw = self.raw(self.identity())
for key, signed_path, changed in ((b"x" * 32, path, raw), (KEY, "/api/v2" + path, raw),
(KEY, authorization.UPDATE_PATH if path == authorization.OPTIONS_PATH else authorization.OPTIONS_PATH, raw),
(KEY, path, b"invalid-json")):
headers = self.headers(raw, signed_path, key=key)
with mock.patch.object(authorization, "parse_request", side_effect=AssertionError("Before auth")):
response = self.client.post("/api/v2" + path, content=changed, headers=headers)
self.assertEqual(response.status_code, 401)
def test_nonce_replay_persists_across_instances_failures_and_concurrent_workers(self):
self.enroll()
nonce = "d" * 32
self.assertEqual(self.fetch(self.identity(), nonce=nonce).status_code, 200)
restarted = authorization.GroupAuthorization(self.db, clock=self.clock)
self.service._reserve_nonce = restarted._reserve_nonce
self.assertEqual(self.fetch(self.identity(), nonce=nonce).status_code, 401)
invalid = {**self.identity(), "extra": 1}
self.assertEqual(self.fetch(invalid, nonce="e" * 32).status_code, 422)
self.assertEqual(self.fetch(invalid, nonce="e" * 32).status_code, 401)
raw = self.raw(self.identity())
headers = self.headers(raw, authorization.OPTIONS_PATH, nonce="f" * 32)
with ThreadPoolExecutor(max_workers=4) as pool:
statuses = list(pool.map(lambda _: self.client.post("/api/v2" + authorization.OPTIONS_PATH,
content=raw, headers=headers).status_code, range(8)))
self.assertEqual(statuses.count(200), 1)
self.assertEqual(statuses.count(401), 7)
def test_clock_bounds_and_bounded_nonce_store(self):
self.enroll()
for offset in (-91, 91):
self.assertEqual(self.fetch(self.identity(), timestamp=NOW + offset).status_code, 401)
for offset in (-90, 90):
self.assertEqual(self.fetch(self.identity(), timestamp=NOW + offset).status_code, 200)
with self.db.connect() as db:
count = db.execute("SELECT COUNT(*) FROM group_authorization_nonces").fetchone()[0]
self.service.max_nonces = count
self.assertEqual(self.fetch(self.identity()).status_code, 503)
self.clock.return_value = NOW + 181
self.assertEqual(self.fetch(self.identity(), timestamp=NOW + 181).status_code, 200)
def test_strict_json_fields_and_types_signed_safe_errors(self):
self.enroll()
mutation = self.mutation()
invalid_options = [[], {}, {**self.identity(), "role_code": "admin"},
{**self.identity(), "client_id": "hospital"}, {**self.identity(), "source_admin_id": True},
{**self.identity(), "company_id": 0}, {**self.identity(), "gid": "gid_" + "A" * 32}]
for body in invalid_options:
self.assertEqual(self.fetch(body).status_code, 422)
invalid_updates = [{**mutation, "actor_source_admin_id": True}, {**mutation, "request_id": "A" * 32},
{**mutation, "expected_revision": "A" * 64}, {**mutation, "role_code": ""},
{**mutation, "permissions": "stats:read"}, {**mutation, "permissions": [True]},
{**mutation, "permissions": ["stats:read", "stats:read"]},
{**mutation, "permissions": [f"fixture:{i}" for i in range(2001)]},
{**mutation, "actor_id": 1}]
for body in invalid_updates:
self.assertEqual(self.update(body).status_code, 422)
for raw in (b' {"client_id":"kefu","client_id":"kefu","gid":"x","source_admin_id":7,"company_id":2}',
b'{"client_id":NaN}', b"\xff", b"not-json"):
self.assertEqual(self.fetch(None, raw=raw).status_code, 422)
def test_duplicate_security_headers_query_and_oversize_requests(self):
self.enroll()
raw = self.raw(self.identity())
for name in ("X-Group-Nonce", "X-Group-Timestamp", "X-Group-Signature"):
headers = list(self.headers(raw, authorization.OPTIONS_PATH).items())
headers.append((name, "f" * 32))
self.assertEqual(self.client.post("/api/v2" + authorization.OPTIONS_PATH, content=raw, headers=headers).status_code, 401)
self.assertEqual(self.fetch(self.identity(), query="?identity=extra").status_code, 422)
self.assertEqual(self.fetch(None, raw=b" " * (authorization.MAX_REQUEST_BYTES + 1)).status_code, 413)
def test_invalid_key_fails_closed_without_paths_or_key_material(self):
for configured in ("", "relative.key", str(self.root / "missing.key")):
with mock.patch.dict(os.environ, {"ZYT_GROUP_METADATA_KEY_FILE": configured}):
raw = self.raw(self.identity())
response = self.client.post("/api/v2" + authorization.OPTIONS_PATH,
content=raw, headers=self.headers(raw, authorization.OPTIONS_PATH))
self.assertEqual(response.status_code, 503)
self.assertNotIn(str(self.root), response.text)
self.assertNotIn(KEY.hex(), response.text)
if __name__ == "__main__":
faulthandler.dump_traceback_later(90, exit=True)
try:
unittest.main(verbosity=2)
finally:
faulthandler.cancel_dump_traceback_later()