AgentServer/apps/server/test/access-policy.test.ts

import fs from 'node:fs'
import os from 'node:os'
import path from 'node:path'
import { describe, expect, it } from 'vitest'
import type { FileAccessPolicy, KnowledgeRootDto } from '@agent-server/shared'
import { config } from '../src/config'
import { checkAccess, checkKnowledgePath, resolveAccess, toClaudeSandbox } from '../src/services/access-policy'
import { defaultSettings } from '../src/services/settings'

const settings = defaultSettings()
const ws = path.join(settings.paths.workspaceRoot, 'ses_a')
const other = path.join(settings.paths.workspaceRoot, 'ses_b')
fs.mkdirSync(path.join(ws, 'uploads'), { recursive: true })
fs.mkdirSync(other, { recursive: true })
const kbDir = fs.mkdtempSync(path.join(os.tmpdir(), 'kb-'))
const kb: KnowledgeRootDto = {
  id: 'kb_1', name: '资料', path: kbDir, description: '', access: 'read', enabled: true,
  check: { exists: true, readable: true, isDirectory: true, denied: null, underHome: false, message: null },
}

const policy = (p: Partial<FileAccessPolicy> = {}): FileAccessPolicy => ({ mode: 'restricted', read: [], write: [], deny: [], ...p })
const resolve = (p: Partial<FileAccessPolicy> = {}, kbIds: string[] = []) =>
  resolveAccess({ policy: policy(p), knowledgeRoots: [kb], selectedKnowledgeRootIds: kbIds, workspaceDir: ws, settings })

describe('checkAccess', () => {
  it('允许读写自己的工作区', () => {
    const a = resolve()
    expect(checkAccess(a, path.join(ws, 'uploads/x.csv'), 'read').allowed).toBe(true)
    expect(checkAccess(a, path.join(ws, 'outputs/new.docx'), 'write').allowed).toBe(true)
  })

  it('restricted 模式拒绝工作区以外的读取', () => {
    expect(checkAccess(resolve(), '/etc/hostname', 'read').allowed).toBe(false)
  })

  it('始终拒绝 CLI 凭证，即使是 broad 模式', () => {
    const a = resolve({ mode: 'broad' })
    expect(checkAccess(a, path.join(config.home, '.claude/.credentials.json'), 'read').allowed).toBe(false)
    expect(checkAccess(a, path.join(config.home, '.codex/auth.json'), 'read').allowed).toBe(false)
    expect(checkAccess(a, path.join(config.home, '.ssh/id_ed25519'), 'read').allowed).toBe(false)
  })

  it('broad 模式可读系统文件，但不能读其他会话的工作区和服务数据库', () => {
    const a = resolve({ mode: 'broad' })
    expect(checkAccess(a, '/etc/hostname', 'read').allowed).toBe(true)
    expect(checkAccess(a, path.join(other, 'secret.txt'), 'read').allowed).toBe(false)
    expect(checkAccess(a, path.join(config.dataDir, 'data/agent.db'), 'read').allowed).toBe(false)
    expect(checkAccess(a, '/etc/hostname', 'write').allowed).toBe(false)
  })

  it('运行环境只读', () => {
    const a = resolve()
    expect(checkAccess(a, path.join(settings.paths.pythonRuntime, 'bin/python'), 'read').allowed).toBe(true)
    expect(checkAccess(a, path.join(settings.paths.pythonRuntime, 'bin/python'), 'write').allowed).toBe(false)
  })

  it('资料库只有在 Profile 启用后才可读', () => {
    expect(checkAccess(resolve(), path.join(kbDir, 'a.pdf'), 'read').allowed).toBe(false)
    expect(checkAccess(resolve({}, ['kb_1']), path.join(kbDir, 'a.pdf'), 'read').allowed).toBe(true)
    expect(checkAccess(resolve({}, ['kb_1']), path.join(kbDir, 'a.pdf'), 'write').allowed).toBe(false)
  })

  it('Profile 的通配禁止项生效', () => {
    const a = resolve({ deny: ['**/*.env'] })
    expect(checkAccess(a, path.join(ws, 'config/.env'), 'read').allowed).toBe(false)
    expect(checkAccess(a, path.join(ws, 'data.csv'), 'read').allowed).toBe(true)
  })

  it('拒绝通过软链接逃逸工作区', () => {
    const link = path.join(ws, 'escape')
    fs.rmSync(link, { force: true })
    fs.symlinkSync('/etc', link)
    expect(checkAccess(resolve(), path.join(link, 'hostname'), 'read').allowed).toBe(false)
  })

  it('拒绝 .. 路径穿越', () => {
    expect(checkAccess(resolve(), path.join(ws, '../ses_b/secret.txt'), 'read').allowed).toBe(false)
  })
})

describe('toClaudeSandbox', () => {
  it('restricted 模式屏蔽 home 并重新开放工作区', () => {
    const sb = toClaudeSandbox(resolve(), settings)
    expect(sb.failIfUnavailable).toBe(true)
    expect(sb.allowUnsandboxedCommands).toBe(false)
    expect(sb.filesystem.denyRead).toContain(config.home)
    expect(sb.filesystem.allowRead).toContain(fs.realpathSync(ws))
    expect(sb.filesystem.allowWrite).toContain(fs.realpathSync(ws))
    expect(sb.filesystem.allowWrite).not.toContain(config.home)
  })
})

describe('checkKnowledgePath', () => {
  it('拒绝包含凭证的目录', () => {
    expect(checkKnowledgePath('~', settings).denied).not.toBeNull()
    expect(checkKnowledgePath('~/.ssh', settings).denied).not.toBeNull()
  })
  it('拒绝工作区目录', () => {
    expect(checkKnowledgePath(ws, settings).message).toContain('工作区')
  })
  it('接受普通目录', () => {
    const c = checkKnowledgePath(kbDir, settings)
    expect(c.exists && c.readable && c.isDirectory && !c.denied).toBe(true)
  })
})