AgentServer/apps/server/src/services/access-policy.ts

// 文件访问策略：全局强制禁止 ∩ Profile 策略 ∪ 基线（docs/design/06-security.md §4，ADR-0010）
import fs from 'node:fs'
import os from 'node:os'
import path from 'node:path'
import picomatch from 'picomatch'
import {
  BUILTIN_GLOBAL_DENY,
  type AppSettings,
  type FileAccessPolicy,
  type KnowledgeRootDto,
  type PathCheck,
  type ResolvedAccessDto,
} from '@agent-server/shared'
import { communitySkillsDir, config, customSkillsDir } from '../config'
import { expandPath, hasGlob, isWithin, realpathLoose } from '../lib/paths'
import { codexHome } from '../providers/codex/cli'

export type AccessOp = 'read' | 'write'

export interface ResolvedAccess {
  mode: FileAccessPolicy['mode']
  workspaceDir: string
  readable: string[]
  writable: string[]
  /** 无条件禁止（内置 + 全局 + Profile），可以是路径或通配 */
  hardDeny: string[]
  /** 默认禁止、但可被更具体的允许路径重新打开（如其他会话的工作区） */
  softDeny: string[]
}

export type AccessDecision = { allowed: true } | { allowed: false; reason: string }

/** restricted 模式下 Bash 沙箱额外屏蔽的大块目录 */
const RESTRICTED_BASH_DENY = () => [config.home, '/srv', '/mnt', '/media', '/root']

export function globalDenyList(settings: AppSettings): string[] {
  // Codex 的登录凭证位于服务专用的 CODEX_HOME（开发环境不在 ~ 下，需要单独列出）
  return [...BUILTIN_GLOBAL_DENY, ...settings.security.globalDeny, path.join(codexHome(), 'auth.json')].map(expandPath)
}

export function resolveAccess(opts: {
  policy: FileAccessPolicy
  knowledgeRoots: KnowledgeRootDto[]
  selectedKnowledgeRootIds: string[]
  workspaceDir: string
  settings: AppSettings
}): ResolvedAccess {
  const { policy, settings } = opts
  const roots = new Map(opts.knowledgeRoots.map(r => [r.id, r]))
  const readable = new Set<string>()
  const writable = new Set<string>()

  // 基线：工作区读写（含 Claude CLI 为该工作目录分配的临时目录）；运行环境与技能只读
  writable.add(opts.workspaceDir)
  writable.add(claudeScratchDir(opts.workspaceDir))
  readable.add(settings.paths.pythonRuntime)
  readable.add(settings.paths.matplotlibrc)
  readable.add(settings.paths.nodeModules)
  readable.add(settings.paths.skillsDir)
  readable.add(communitySkillsDir())
  // 只开放生效目录；版本库（custom-skill-store）在服务数据目录中，不可见
  readable.add(customSkillsDir())

  for (const id of opts.selectedKnowledgeRootIds) {
    const root = roots.get(id)
    if (!root?.enabled) continue
    const target = root.access === 'write' ? writable : readable
    target.add(expandPath(root.path))
  }
  for (const rule of policy.read) {
    const p = ruleToPath(rule, roots)
    if (p) readable.add(p)
  }
  for (const rule of policy.write) {
    const p = ruleToPath(rule, roots)
    if (p) writable.add(p)
  }
  if (policy.mode === 'broad') readable.add('/')

  return {
    mode: policy.mode,
    workspaceDir: opts.workspaceDir,
    readable: [...readable].map(p => realpathLoose(p)),
    writable: [...writable].map(p => realpathLoose(p)),
    hardDeny: [...globalDenyList(settings), ...policy.deny.map(expandPath)],
    softDeny: [realpathLoose(settings.paths.workspaceRoot), realpathLoose(config.dataDir)],
  }
}

/** Claude CLI 的会话临时目录：<tmp>/claude-<uid>/<工作目录路径中非字母数字替换为 -> */
export function claudeScratchDir(workspaceDir: string): string {
  return path.join(os.tmpdir(), `claude-${process.getuid?.() ?? 0}`, workspaceDir.replace(/[^a-zA-Z0-9]/g, '-'))
}

function ruleToPath(rule: FileAccessPolicy['read'][number], roots: Map<string, KnowledgeRootDto>): string | null {
  if ('path' in rule) return expandPath(rule.path)
  const root = roots.get(rule.knowledgeRootId)
  return root?.enabled ? expandPath(root.path) : null
}

const matcherCache = new Map<string, (p: string) => boolean>()

/** 路径是否命中全局禁止项（内置 + 设置中追加的）；后台文件管理也遵守 */
export function isGloballyDenied(target: string, settings: AppSettings): boolean {
  const real = realpathLoose(target)
  return globalDenyList(settings).some(e => matchesEntry(real, e))
}

function matchesEntry(target: string, entry: string): boolean {
  if (!hasGlob(entry)) return isWithin(target, entry)
  let m = matcherCache.get(entry)
  if (!m) {
    // 相对模式在任何位置匹配：不含 / 时按文件名匹配，否则匹配去掉开头 / 的路径
    const relative = !path.isAbsolute(entry)
    const pm = picomatch(entry, { dot: true, basename: relative && !entry.includes('/') })
    m = relative ? (p: string) => pm(p.slice(1)) : pm
    matcherCache.set(entry, m)
  }
  // 通配命中目录本身或其任一祖先，都视为命中
  for (let p = target; ; p = path.dirname(p)) {
    if (m(p)) return true
    if (path.dirname(p) === p) return false
  }
}

/** 沙箱配置只接受锚定路径：把相对通配挂到每个可访问根目录下 */
function anchorGlobs(entries: string[], roots: string[]): string[] {
  return entries.flatMap(e => (path.isAbsolute(e) ? [e] : roots.map(r => path.join(r, e.startsWith('**/') ? e : `**/${e}`))))
}

/** 返回 list 中包含 target 的最具体（最长）路径；通配项不参与 */
function mostSpecific(target: string, list: string[]): string | null {
  let best: string | null = null
  for (const p of list) if (!hasGlob(p) && isWithin(target, p) && (!best || p.length > best.length)) best = p
  return best
}

/**
 * 服务管理的技能生效目录（第三方、自定义）：位于全局禁止的 {dataDir}/data 之内，但必须只读可见，
 * 否则 Agent 读不到技能的参考文档、运行不了脚本。只放开这两个目录，data 下的其他内容（数据库、版本库、MCP 数据）仍然禁止
 */
export const managedSkillRoots = () => [communitySkillsDir(), customSkillsDir()].map(realpathLoose)

export function checkAccess(access: ResolvedAccess, rawPath: string, op: AccessOp): AccessDecision {
  const target = realpathLoose(rawPath)
  if (op === 'read' && managedSkillRoots().some(r => isWithin(target, r))) return { allowed: true }
  for (const entry of access.hardDeny) {
    if (matchesEntry(target, entry)) return { allowed: false, reason: `路径被禁止访问：${entry}` }
  }

  const allowList = op === 'write' ? access.writable : [...access.readable, ...access.writable]
  const allow = mostSpecific(target, allowList)
  if (!allow) {
    return { allowed: false, reason: op === 'write' ? `不允许写入工作区以外的路径：${target}` : `路径不在可读范围内：${target}` }
  }
  const soft = mostSpecific(target, access.softDeny)
  if (soft && soft.length > allow.length) return { allowed: false, reason: `路径被禁止访问：${soft}` }
  return { allowed: true }
}

/** 把合并后的策略映射为 Claude Bash 沙箱设置（06 §4.2） */
export function toClaudeSandbox(access: ResolvedAccess, settings: AppSettings) {
  const readable = access.readable.filter(p => p !== '/')
  const hardDeny = anchorGlobs(access.hardDeny, [...readable, ...access.writable])
  const denyRead = [...hardDeny, ...(access.mode === 'restricted' ? RESTRICTED_BASH_DENY() : access.softDeny)]
  return {
    enabled: true,
    failIfUnavailable: true,
    allowUnsandboxedCommands: false,
    autoAllowBashIfSandboxed: true,
    filesystem: {
      // 注意：沙箱禁止创建 Unix socket，LibreOffice 无法在 Bash 中运行，改由 convert_document 工具提供（services/office.ts）
      allowWrite: access.writable,
      denyWrite: hardDeny,
      denyRead,
      allowRead: [...readable, ...access.writable],
    },
    network: { allowedDomains: settings.network.allowedDomains },
  }
}

export function toResolvedDto(access: ResolvedAccess): ResolvedAccessDto {
  return {
    mode: access.mode,
    readable: access.readable,
    writable: access.writable,
    denied: [...access.hardDeny, ...access.softDeny.map(p => `${p}（其他会话 / 服务数据，工作区除外）`)],
  }
}

/** 通用路径检查（系统设置中的路径）：只看是否存在、可读 */
export function checkPathBasic(raw: string): PathCheck {
  const p = expandPath(raw)
  const result: PathCheck = { exists: false, readable: false, isDirectory: false, denied: null, underHome: isWithin(p, config.home), message: null }
  try {
    const st = fs.statSync(p)
    result.exists = true
    result.isDirectory = st.isDirectory()
    fs.accessSync(p, fs.constants.R_OK)
    result.readable = true
  } catch {
    result.message = result.exists ? '服务进程没有读取权限' : '路径不存在'
  }
  return result
}

export function checkKnowledgePath(raw: string, settings: AppSettings): PathCheck {
  const p = expandPath(raw)
  const result: PathCheck = { exists: false, readable: false, isDirectory: false, denied: null, underHome: isWithin(p, config.home), message: null }
  for (const entry of globalDenyList(settings)) {
    if (matchesEntry(p, entry) || (!hasGlob(entry) && isWithin(entry, p))) {
      result.denied = entry
      result.message = `与全局禁止项冲突：${entry}`
      return result
    }
  }
  if (isWithin(p, realpathLoose(settings.paths.workspaceRoot))) {
    result.message = '不能把会话工作区目录设为资料库'
    return result
  }
  try {
    const st = fs.statSync(p)
    result.exists = true
    result.isDirectory = st.isDirectory()
    fs.accessSync(p, fs.constants.R_OK)
    result.readable = true
  } catch {
    result.message ??= result.exists ? '服务进程没有读取权限' : '路径不存在'
  }
  if (result.exists && !result.isDirectory) result.message = '资料库必须是目录'
  if (!result.message && result.underHome) {
    result.message = `位于 home 目录下；以 systemd 加固方式部署时需要在 unit 中追加 BindReadOnlyPaths=${p}`
  }
  return result
}