aps-agent/scripts/control-plane.test.mjs

/**
 * A4 控制面用例：身份、策略、凭据、确认、审计五块一起验收（A4.1–A4.6）。
 *
 * 这些用例刻意走 HTTP 与 SSE 的公开接口，而不是直接调内部函数——控制面的价值就在于
 * 「谁都绕不过去」，只有从外面打进来才算数。
 */
import assert from 'node:assert/strict';
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { test } from 'node:test';

import { createFakeProvider } from '@aps/provider-sdk/testing';

const subject = { userId: 'u-a4', tenantId: 't-a4' };
const otherSubject = { userId: 'u-other', tenantId: 't-a4' };

/** 真实系统都要先绑账号（A4.1）：包一层把 requiresIdentity 打开 */
function identityProvider(id) {
  const base = createFakeProvider(id);
  return {
    stats: base.stats,
    describe: async () => ({ ...(await base.describe()), requiresIdentity: true }),
    invoke: (context, call) => base.invoke(context, call),
    health: () => base.health(),
    preview: (context, call) => base.preview(context, call),
  };
}

function tmpHome() {
  const dir = mkdtempSync(join(tmpdir(), 'aps-a4-'));
  process.env.APS_CREDENTIAL_KEY_FILE = join(dir, 'secret.key');
  return dir;
}

async function startPlatform(options = {}) {
  delete process.env.APS_DEV_PROVIDER;
  const { startServer } = await import('../server/dist/index.js');
  return startServer({
    host: '127.0.0.1',
    port: 0,
    version: 'test',
    engine: 'fake',
    databasePath: options.databasePath ?? ':memory:',
    ...(options.capabilityVisibility === undefined ? {} : { capabilityVisibility: options.capabilityVisibility }),
    ...(options.approvalTtlMs === undefined ? {} : { approvalTtlMs: options.approvalTtlMs }),
  });
}

const headersFor = (who) => ({ 'content-type': 'application/json', 'x-aps-user': who.userId, 'x-aps-tenant': who.tenantId });

async function createSession(server, who = subject) {
  const response = await fetch(`${server.url}/v1/sessions`, {
    method: 'POST',
    headers: headersFor(who),
    body: JSON.stringify({ scope: 'project', projectId: 'p-default', title: 'A4 用例' }),
  });
  assert.equal(response.status, 200, '建会话应成功');
  return (await response.json()).session.sessionId;
}

/** 打开 SSE 并收集事件（用完 close()） */
async function openStream(server, sessionId, who = subject) {
  const controller = new AbortController();
  const events = [];
  const stream = await fetch(`${server.url}/v1/sessions/${sessionId}/events`, { headers: { accept: 'text/event-stream', 'x-aps-user': who.userId }, signal: controller.signal });
  assert.equal(stream.status, 200, '事件流要能连上（会话归属要对）');
  void (async () => {
    const decoder = new TextDecoder();
    let buffer = '';
    try {
      for await (const chunk of stream.body) {
        buffer += decoder.decode(chunk, { stream: true });
        let index = buffer.indexOf('\n\n');
        while (index >= 0) {
          const frame = buffer.slice(0, index);
          buffer = buffer.slice(index + 2);
          const line = frame.split('\n').find((item) => item.startsWith('data: '));
          if (line) events.push(JSON.parse(line.slice(6)));
          index = buffer.indexOf('\n\n');
        }
      }
    } catch {
      /* aborted */
    }
  })();
  return {
    events,
    async waitFor(predicate, timeoutMs = 3000) {
      const deadline = Date.now() + timeoutMs;
      while (Date.now() < deadline) {
        if (events.some(predicate)) return true;
        await new Promise((resolve) => setTimeout(resolve, 15));
      }
      return false;
    },
    close() {
      controller.abort();
    },
  };
}

test('A4.1 身份：一个用户绑多个系统账号，停用一个不影响另一个', async () => {
  const dir = tmpHome();
  const server = await startPlatform();
  try {
    const aps = identityProvider('aps');
    await server.service.registerProvider(aps);
    await server.service.registerProvider(identityProvider('erp'));
    const identity = server.service.controlPlane.identity;

    identity.bind({ userId: subject.userId, tenantId: subject.tenantId, providerId: 'aps', accountId: 'E1001' });
    identity.bind({ userId: subject.userId, tenantId: subject.tenantId, providerId: 'erp', accountId: 'R2002' });
    assert.deepEqual(identity.list(subject).map((binding) => binding.providerId), ['aps', 'erp']);

    // 一次只读调用：Provider 拿到的是映射后的账号，不是平台用户
    const apsCall = await server.service.invoke({ subject, sessionId: 's-a4', capabilityId: 'aps.echo', params: { text: 'hi' } });
    assert.equal(apsCall.status, 'succeeded');
    assert.equal(aps.stats.contexts.at(-1).accountId, 'E1001');

    // 停用 aps：aps 被拒，erp 照常
    identity.disable(subject, 'aps');
    const denied = await server.service.invoke({ subject, sessionId: 's-a4', capabilityId: 'aps.echo', params: { text: 'hi' } });
    assert.equal(denied.status, 'failed');
    assert.equal(denied.errorKind, 'forbidden');
    const erpCall = await server.service.invoke({ subject, sessionId: 's-a4', capabilityId: 'erp.echo', params: { text: 'hi' } });
    assert.equal(erpCall.status, 'succeeded', '停用 aps 不该影响 erp 的登录态');
  } finally {
    await server.close();
    rmSync(dir, { recursive: true, force: true });
    delete process.env.APS_CREDENTIAL_KEY_FILE;
  }
});

test('A4.1/A4.2 未绑定身份默认拒绝，并且留下 policy.denied 审计', async () => {
  const dir = tmpHome();
  const server = await startPlatform();
  try {
    await server.service.registerProvider(identityProvider('aps'));
    const receipt = await server.service.invoke({ subject, sessionId: 's-a4', capabilityId: 'aps.echo', params: { text: 'hi' } });
    assert.equal(receipt.status, 'failed');
    assert.equal(receipt.errorKind, 'forbidden');
    assert.match(String(receipt.result.message), /绑定/);

    const trace = server.service.controlPlane.audit.trace(receipt.executionId);
    assert.equal(trace.length, 1);
    assert.equal(trace[0].event, 'policy.denied');
    assert.equal(trace[0].decision, 'policy.identity_unbound');
    assert.equal(trace[0].actorUserId, subject.userId);
  } finally {
    await server.close();
    rmSync(dir, { recursive: true, force: true });
    delete process.env.APS_CREDENTIAL_KEY_FILE;
  }
});

test('A4.2 策略：可见的读放行、写要确认、个人会话写直接拒', async () => {
  const dir = tmpHome();
  const server = await startPlatform();
  try {
    await server.service.registerProvider(identityProvider('aps'));
    server.service.controlPlane.identity.bind({ userId: subject.userId, tenantId: subject.tenantId, providerId: 'aps', accountId: 'E1001' });

    const read = await server.service.invoke({ subject, sessionId: 's-a4', projectId: 'p-default', capabilityId: 'aps.echo', params: { text: 'hi' } });
    assert.equal(read.status, 'succeeded');

    const write = await server.service.invoke({ subject, sessionId: 's-a4', projectId: 'p-default', capabilityId: 'aps.write', params: { value: 1 } });
    assert.equal(write.status, 'pending-confirmation', '写操作先开确认卡');

    // ADR-0003：个人会话只读，写操作必须在项目会话里
    const personal = await server.service.invoke({ subject, sessionId: 's-a4', capabilityId: 'aps.write', params: { value: 1 } });
    assert.equal(personal.status, 'failed');
    assert.match(String(personal.result.message), /项目会话/);
  } finally {
    await server.close();
    rmSync(dir, { recursive: true, force: true });
    delete process.env.APS_CREDENTIAL_KEY_FILE;
  }
});

test('A4.2 平台的允许不覆盖业务系统的拒绝：forbidden 原样落地、不重试', async () => {
  const dir = tmpHome();
  const server = await startPlatform();
  try {
    const aps = identityProvider('aps');
    await server.service.registerProvider(aps);
    server.service.controlPlane.identity.bind({ userId: subject.userId, tenantId: subject.tenantId, providerId: 'aps', accountId: 'E1001' });

    const receipt = await server.service.invoke({
      subject,
      sessionId: 's-a4',
      projectId: 'p-default',
      capabilityId: 'aps.declare-error',
      params: { kind: 'forbidden' },
    });
    assert.equal(receipt.status, 'failed');
    assert.equal(receipt.errorKind, 'forbidden');
    assert.equal(aps.stats.calls['aps.declare-error'], 1, '业务系统的拒绝不该被平台重试');

    const trace = server.service.controlPlane.audit.trace(receipt.executionId);
    assert.ok(trace.some((entry) => entry.event === 'provider.denied'), '审计里要记下是业务系统拒绝的');
    assert.ok(!trace.some((entry) => entry.event === 'execution.finished' && entry.decision === 'succeeded'));
  } finally {
    await server.close();
    rmSync(dir, { recursive: true, force: true });
    delete process.env.APS_CREDENTIAL_KEY_FILE;
  }
});

test('A4.3 凭据：库里只有密文，执行期经句柄取用，参数与日志里都没有明文', async () => {
  const dir = tmpHome();
  const server = await startPlatform();
  const secret = 'aps-token-9f2a1c';
  try {
    const aps = identityProvider('aps');
    await server.service.registerProvider(aps);
    const control = server.service.controlPlane;
    control.identity.bind({ userId: subject.userId, tenantId: subject.tenantId, providerId: 'aps', accountId: 'E1001' });
    control.credentials.put(subject, 'aps', { token: secret });

    const sealed = control.credentials.sealedFor(subject, 'aps');
    assert.ok(sealed && sealed.ciphertext.length > 0, '凭据要加密后落库');
    assert.ok(!JSON.stringify(sealed).includes(secret), '密文里不能出现明文');
    assert.equal(control.credentials.redact(`token=${secret} 已加载`), 'token=*** 已加载');

    // Provider 能通过句柄拿到；参数里没有凭据（handle 是上下文里的对象，不是参数）
    const receipt = await server.service.invoke({ subject, sessionId: 's-a4', projectId: 'p-default', capabilityId: 'aps.echo', params: { text: 'hi' } });
    assert.equal(receipt.status, 'succeeded');
    const lastContext = aps.stats.contexts.at(-1);
    assert.equal(lastContext.hasCredentials, true, '有配置凭据时要签发一次性句柄');
    assert.ok(!JSON.stringify(lastContext.params).includes(secret), '凭据不能混进工具参数');

    const record = server.service.core.executions.get(receipt.executionId);
    assert.ok(!JSON.stringify(record).includes(secret), '执行记录里不能有明文');
    // 授权用完即废
    assert.equal(control.credentials.activeGrants(), 0, '执行结束后一次性授权应被撤销');
  } finally {
    await server.close();
    rmSync(dir, { recursive: true, force: true });
    delete process.env.APS_CREDENTIAL_KEY_FILE;
  }
});

test('A4.4 确认卡：写操作等确认、确认后执行、事件齐全、重放不重复执行', async () => {
  const dir = tmpHome();
  const server = await startPlatform();
  try {
    const aps = identityProvider('aps');
    await server.service.registerProvider(aps);
    server.service.controlPlane.identity.bind({ userId: subject.userId, tenantId: subject.tenantId, providerId: 'aps', accountId: 'E1001' });
    const sessionId = await createSession(server);
    const stream = await openStream(server, sessionId);

    const receipt = await server.service.invokeFromSession(sessionId, { capabilityId: 'aps.write', params: { value: 3 }, idempotencyKey: 'idem-a4-write-1' });
    assert.equal(receipt.status, 'pending-confirmation');
    assert.ok(receipt.approvalId);
    assert.ok(await stream.waitFor((event) => event.type === 'approval.requested'), '要推确认卡给界面');
    assert.equal(aps.stats.calls['aps.write'], undefined, '没确认前不能动业务系统');

    // 待确认中的执行也要能按 executionId 查到，且只有本人能查
    const pendingView = await fetch(`${server.url}/v1/executions/${receipt.executionId}`, { headers: headersFor(subject) });
    assert.equal((await pendingView.json()).status, 'pending-confirmation');
    const stolenView = await fetch(`${server.url}/v1/executions/${receipt.executionId}`, { headers: headersFor(otherSubject) });
    assert.equal(stolenView.status, 403, '执行记录只有本人能查');

    const decision = await fetch(`${server.url}/v1/approvals/${receipt.approvalId}`, {
      method: 'POST',
      headers: headersFor(subject),
      body: JSON.stringify({ decision: 'approve', revision: 7, idempotencyKey: 'idem-a4-approve-1' }),
    });
    assert.equal(decision.status, 200);
    const decided = await decision.json();
    assert.equal(decided.decision, 'approved');
    assert.equal(decided.receipt.status, 'succeeded');
    assert.equal(aps.stats.calls['aps.write'], 1, '确认后只执行一次');
    assert.ok(await stream.waitFor((event) => event.type === 'approval.resolved'), '要推确认结果');
    assert.ok(await stream.waitFor((event) => event.type === 'execution.updated'), '要推执行状态');

    // 换一个幂等键再确认：仍不重复执行
    const again = await fetch(`${server.url}/v1/approvals/${receipt.approvalId}`, {
      method: 'POST',
      headers: headersFor(subject),
      body: JSON.stringify({ decision: 'approve', revision: 7, idempotencyKey: 'idem-a4-approve-2' }),
    });
    assert.equal(again.status, 200);
    assert.equal((await again.json()).receipt.executionId, decided.receipt.executionId);
    assert.equal(aps.stats.calls['aps.write'], 1, '同一写请求重试不重复生效');
    stream.close();
  } finally {
    await server.close();
    rmSync(dir, { recursive: true, force: true });
    delete process.env.APS_CREDENTIAL_KEY_FILE;
  }
});

test('A4.4 版本变了确认即失效；过期确认不再执行', async () => {
  const dir = tmpHome();
  const server = await startPlatform({ approvalTtlMs: 30 });
  try {
    const aps = identityProvider('aps');
    await server.service.registerProvider(aps);
    server.service.controlPlane.identity.bind({ userId: subject.userId, tenantId: subject.tenantId, providerId: 'aps', accountId: 'E1001' });
    const sessionId = await createSession(server);

    const first = await server.service.invokeFromSession(sessionId, { capabilityId: 'aps.write', params: { value: 1 }, idempotencyKey: 'idem-a4-version-1' });
    const mismatch = await fetch(`${server.url}/v1/approvals/${first.approvalId}`, {
      method: 'POST',
      headers: headersFor(subject),
      body: JSON.stringify({ decision: 'approve', revision: 999, idempotencyKey: 'idem-a4-version-2' }),
    });
    assert.equal((await mismatch.json()).decision, 'expired', '目标版本变了确认就作废');
    assert.equal(aps.stats.calls['aps.write'], undefined);

    const second = await server.service.invokeFromSession(sessionId, { capabilityId: 'aps.write', params: { value: 2 }, idempotencyKey: 'idem-a4-version-3' });
    await new Promise((resolve) => setTimeout(resolve, 60));
    const expired = await fetch(`${server.url}/v1/approvals/${second.approvalId}`, {
      method: 'POST',
      headers: headersFor(subject),
      body: JSON.stringify({ decision: 'approve', revision: 7, idempotencyKey: 'idem-a4-version-4' }),
    });
    assert.equal((await expired.json()).decision, 'expired', '过期确认要提示重新确认');
    assert.equal(aps.stats.calls['aps.write'], undefined);
  } finally {
    await server.close();
    rmSync(dir, { recursive: true, force: true });
    delete process.env.APS_CREDENTIAL_KEY_FILE;
  }
});

test('A4.4 待确认动作跨进程重启存活', async () => {
  const dir = tmpHome();
  const databasePath = join(dir, 'aps.sqlite');
  const first = await startPlatform({ databasePath });
  let approvalId;
  let sessionId;
  let executionId;
  try {
    await first.service.registerProvider(identityProvider('aps'));
    first.service.controlPlane.identity.bind({ userId: subject.userId, tenantId: subject.tenantId, providerId: 'aps', accountId: 'E1001' });
    sessionId = await createSession(first);
    const receipt = await first.service.invokeFromSession(sessionId, { capabilityId: 'aps.write', params: { value: 5 }, idempotencyKey: 'idem-a4-restart-1' });
    approvalId = receipt.approvalId;
    executionId = receipt.executionId;
  } finally {
    await first.close();
  }

  const second = await startPlatform({ databasePath });
  try {
    await second.service.registerProvider(identityProvider('aps'));
    const pending = second.service.core.approvals.get(approvalId);
    assert.ok(pending, '重启后待确认动作还在（落库，不是内存）');
    assert.equal(pending.status, 'pending-confirmation');

    const decision = await fetch(`${second.url}/v1/approvals/${approvalId}`, {
      method: 'POST',
      headers: headersFor(subject),
      body: JSON.stringify({ decision: 'approve', revision: 7, idempotencyKey: 'idem-a4-restart-2' }),
    });
    assert.equal(decision.status, 200);
    assert.equal((await decision.json()).decision, 'approved');

    const execution = await (await fetch(`${second.url}/v1/executions/${executionId}`, { headers: headersFor(subject) })).json();
    assert.equal(execution.status, 'succeeded', '重启后仍能按原 executionId 执行并落回执');
  } finally {
    await second.close();
    rmSync(dir, { recursive: true, force: true });
    delete process.env.APS_CREDENTIAL_KEY_FILE;
  }
});

test('A4.5 审计：按 executionId 追到人、参数、审批与回执；改一条就断链', async () => {
  const dir = tmpHome();
  const databasePath = join(dir, 'aps.sqlite');
  const server = await startPlatform({ databasePath });
  try {
    await server.service.registerProvider(identityProvider('aps'));
    server.service.controlPlane.identity.bind({ userId: subject.userId, tenantId: subject.tenantId, providerId: 'aps', accountId: 'E1001' });
    const sessionId = await createSession(server);
    const pending = await server.service.invokeFromSession(sessionId, { capabilityId: 'aps.write', params: { value: 8 }, idempotencyKey: 'idem-a4-audit-1' });
    await fetch(`${server.url}/v1/approvals/${pending.approvalId}`, {
      method: 'POST',
      headers: headersFor(subject),
      body: JSON.stringify({ decision: 'approve', revision: 7, idempotencyKey: 'idem-a4-audit-2' }),
    });

    const trace = await (await fetch(`${server.url}/v1/executions/${pending.executionId}/trace`, { headers: headersFor(subject) })).json();
    assert.equal(trace.executionId, pending.executionId);
    assert.equal(trace.chainOk, true);
    assert.equal(trace.approval.decidedBy, subject.userId, '要能追到是谁批准的');
    assert.equal(trace.receipt.status, 'succeeded');
    const events = trace.entries.map((entry) => entry.event);
    assert.ok(events.includes('policy.requires_confirmation'));
    assert.ok(events.includes('approval.requested'));
    assert.ok(events.includes('approval.decided'));
    assert.ok(events.includes('execution.finished'));
    assert.ok(trace.entries.every((entry) => entry.actorUserId === subject.userId));
    assert.ok(trace.entries.some((entry) => typeof entry.paramsDigest === 'string' && entry.paramsDigest.startsWith('sha256:')));

    // 别人看不到这条执行
    const forbidden = await fetch(`${server.url}/v1/executions/${pending.executionId}/trace`, { headers: headersFor(otherSubject) });
    assert.equal(forbidden.status, 403);

    // 直接改库里的审计记录：链校验必须报错
    const { DatabaseSync } = await import('node:sqlite');
    const raw = new DatabaseSync(databasePath);
    raw.prepare('update audit_log set decision = ? where seq = (select min(seq) from audit_log)').run('被改过');
    raw.close();
    const verified = server.service.controlPlane.audit.verify();
    assert.equal(verified.ok, false, '审计记录被改过要能被发现');
    assert.equal(verified.brokenAt, 1);
  } finally {
    await server.close();
    rmSync(dir, { recursive: true, force: true });
    delete process.env.APS_CREDENTIAL_KEY_FILE;
  }
});

test('A4.6 越权与绕过：不能替别人确认、看不见的能力调不了、停用后一律拒', async () => {
  const dir = tmpHome();
  const server = await startPlatform({ capabilityVisibility: { defaultAllow: false, denyWrite: false, grants: [{ userId: subject.userId, allow: ['aps.*'] }] } });
  try {
    await server.service.registerProvider(identityProvider('aps'));
    server.service.controlPlane.identity.bind({ userId: subject.userId, tenantId: subject.tenantId, providerId: 'aps', accountId: 'E1001' });
    const sessionId = await createSession(server);
    const pending = await server.service.invokeFromSession(sessionId, { capabilityId: 'aps.write', params: { value: 1 }, idempotencyKey: 'idem-a4-sec-1' });

    // 别人的确认卡：不是发起人就不给确认
    const stolen = await fetch(`${server.url}/v1/approvals/${pending.approvalId}`, {
      method: 'POST',
      headers: headersFor(otherSubject),
      body: JSON.stringify({ decision: 'approve', revision: 7, idempotencyKey: 'idem-a4-sec-2' }),
    });
    assert.equal(stolen.status, 403, '确认人只能是会话发起人');

    // 白名单外的身份：能力不可见，调用直接拒
    const outsider = await fetch(`${server.url}/v1/capabilities`, { headers: headersFor(otherSubject) });
    assert.deepEqual((await outsider.json()).capabilities, [], '别人的能力清单是空的');
    const outsiderCall = await server.service.invoke({ subject: otherSubject, sessionId: 's-x', projectId: 'p-default', capabilityId: 'aps.echo', params: { text: 'hi' } });
    assert.equal(outsiderCall.status, 'failed');
    assert.equal(outsiderCall.errorKind, 'forbidden');

    // 越权尝试要留痕
    const trace = server.service.controlPlane.audit.trace(outsiderCall.executionId);
    assert.equal(trace[0].decision, 'policy.capability_not_visible');

    // 绕过尝试：没有内部令牌打不进工具回调
    const bypass = await fetch(`${server.url}/internal/tool-call`, {
      method: 'POST',
      headers: { 'content-type': 'application/json' },
      body: JSON.stringify({ sessionId, capabilityId: 'aps.write', params: { value: 1 } }),
    });
    assert.equal(bypass.status, 401, '工具回调只认平台内部令牌');
  } finally {
    await server.close();
    rmSync(dir, { recursive: true, force: true });
    delete process.env.APS_CREDENTIAL_KEY_FILE;
  }
});

test('A4.6 会话归属：别人的会话发消息与取能力清单都被拒', async () => {
  const dir = tmpHome();
  const server = await startPlatform();
  try {
    await server.service.registerProvider(identityProvider('aps'));
    const sessionId = await createSession(server);
    const send = await fetch(`${server.url}/v1/sessions/${sessionId}/messages`, {
      method: 'POST',
      headers: headersFor(otherSubject),
      body: JSON.stringify({ text: '插一句', idempotencyKey: 'idem-a4-owner-1' }),
    });
    assert.equal(send.status, 403);
    const capabilities = await fetch(`${server.url}/v1/capabilities?sessionId=${sessionId}`, { headers: headersFor(otherSubject) });
    assert.equal(capabilities.status, 403);
  } finally {
    await server.close();
    rmSync(dir, { recursive: true, force: true });
    delete process.env.APS_CREDENTIAL_KEY_FILE;
  }
});