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, mounterDir } from '../src/config'
import { checkAccess, 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 kbRoot = (over: Partial<KnowledgeRootDto>): KnowledgeRootDto => ({
  id: 'kb_1', name: '资料', kind: 'managed', path: kbDir, description: '', access: 'read', enabled: true, url: null, local: null, remote: null,
  status: { state: 'ok', message: null, since: null }, collectionIds: [], ...over,
})
const mountDir = fs.mkdtempSync(path.join(os.tmpdir(), 'kb-mnt-'))
const roots = [
  kbRoot({}),
  kbRoot({ id: 'kb_rw', path: path.join(kbDir, 'rw'), access: 'write' }),
  // 本机目录即使数据被改成 write 也只读
  kbRoot({ id: 'kb_local', kind: 'local', path: mountDir, access: 'write', local: { path: '/home/u/docs', uuid: null, relPath: null } }),
  kbRoot({ id: 'kb_link', kind: 'link', path: '', url: 'http://wiki.local/a' }),
]

const policy = (p: Partial<FileAccessPolicy> = {}): FileAccessPolicy => ({ mode: 'restricted', read: [], write: [], deny: [], ...p })
const resolve = (p: Partial<FileAccessPolicy> = {}, kbIds: string[] = []) =>
  resolveAccess({ policy: policy(p), knowledgeRoots: roots, 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('只有托管目录可以写；本机目录只读；网络链接不开放任何路径', () => {
    const a = resolve({}, ['kb_rw', 'kb_local', 'kb_link'])
    expect(checkAccess(a, path.join(kbDir, 'rw/new.md'), 'write').allowed).toBe(true)
    expect(checkAccess(a, path.join(mountDir, 'a.md'), 'read').allowed).toBe(true)
    expect(checkAccess(a, path.join(mountDir, 'a.md'), 'write').allowed).toBe(false)
    expect(a.readable).not.toContain('')
    // 按资料库设置的写规则也只对可写的托管目录生效
    const rules = resolve({ write: [{ knowledgeRootId: 'kb_local' }, { knowledgeRootId: 'kb_1' }, { knowledgeRootId: 'kb_rw' }], read: [{ knowledgeRootId: 'kb_link' }] })
    expect(checkAccess(rules, path.join(mountDir, 'a.md'), 'write').allowed).toBe(false)
    expect(checkAccess(rules, path.join(kbDir, 'a.md'), 'write').allowed).toBe(false)
    expect(checkAccess(rules, path.join(kbDir, 'rw/x.md'), 'write').allowed).toBe(true)
  })

  it('挂载服务的目录（socket、密钥、凭据）始终禁止，即使是 broad 模式', () => {
    expect(checkAccess(resolve({ mode: 'broad' }), path.join(mounterDir(), 'token'), 'read').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)
  })
})