第2080篇:企业AI应用的权限管理——基于角色的LLM访问控制
2026/4/30大约 7 分钟
第2080篇:企业AI应用的权限管理——基于角色的LLM访问控制
适读人群:负责企业AI应用安全的工程师 | 阅读时长:约18分钟 | 核心价值:设计和实现AI功能的细粒度权限控制,确保不同角色只能访问适合其权限的AI能力
公司内部的AI应用有一个常被忽视的安全问题:权限控制。
不同角色的员工,能访问的数据不同,能调用的功能不同。一个实习生和一个CTO,在AI助手中能看到的信息、能执行的操作,应该是完全不同的。
很多团队给所有人用同一套AI服务,没有任何权限区分——这是一个严重的安全隐患。
AI权限的独特挑战
传统RBAC(基于角色的访问控制)在AI场景面临新问题:
问题1:LLM不认识权限概念
你不能直接告诉LLM"这个用户没有权限"——LLM会从训练数据推断,可能绕过这个限制。
问题2:RAG数据的权限粒度
知识库里有不同权限级别的文档,向量检索时如何确保只返回用户有权看的内容?
问题3:Tool调用的权限
不同权限的用户能调用不同的Tool,动态控制比较复杂。
权限模型设计
/**
* AI功能的权限模型
* 三个维度:角色权限、数据访问权限、功能调用权限
*/
@Data
public class AiPermissionModel {
/**
* 角色定义
*/
public enum AiRole {
PUBLIC_USER("公开用户", List.of(AiFeature.BASIC_QA)),
EMPLOYEE("普通员工", List.of(AiFeature.BASIC_QA, AiFeature.INTERNAL_KB,
AiFeature.PERSONAL_DATA)),
MANAGER("管理层", List.of(AiFeature.BASIC_QA, AiFeature.INTERNAL_KB,
AiFeature.PERSONAL_DATA, AiFeature.TEAM_DATA,
AiFeature.REPORT_GENERATION)),
ADMIN("管理员", List.of(AiFeature.values())),
DATA_ANALYST("数据分析师", List.of(AiFeature.BASIC_QA, AiFeature.INTERNAL_KB,
AiFeature.ANALYTICS, AiFeature.SQL_QUERY));
private final String label;
private final List<AiFeature> allowedFeatures;
AiRole(String label, List<AiFeature> features) {
this.label = label;
this.allowedFeatures = features;
}
}
/**
* AI功能列表
*/
public enum AiFeature {
BASIC_QA, // 基础问答
INTERNAL_KB, // 内部知识库访问
PERSONAL_DATA, // 个人数据查询
TEAM_DATA, // 团队数据访问
REPORT_GENERATION, // 报告生成
SQL_QUERY, // SQL查询执行
ANALYTICS, // 数据分析
ADMIN_TOOLS, // 管理工具
CROSS_DEPT_DATA // 跨部门数据
}
/**
* 数据访问级别
*/
public enum DataAccessLevel {
PUBLIC(0), // 所有人可见
INTERNAL(1), // 公司内部
DEPARTMENT(2), // 部门内部
ROLE_BASED(3), // 按角色
CONFIDENTIAL(4), // 机密
SECRET(5); // 最高机密
private final int level;
DataAccessLevel(int level) { this.level = level; }
}
}权限感知的AI服务
/**
* 权限感知的AI服务
* 根据用户权限动态控制能力
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class PermissionAwareAiService {
private final ChatLanguageModel llm;
private final UserPermissionService permissionService;
private final PermissionedVectorStore vectorStore;
private final AiToolRegistry toolRegistry;
/**
* 主要入口:带权限控制的AI对话
*/
public AiResponse chat(String userId, String message, ChatContext context) {
// 1. 获取用户权限
UserPermissions permissions = permissionService.getPermissions(userId);
// 2. 检查基础访问权限
if (!permissions.hasFeature(AiPermissionModel.AiFeature.BASIC_QA)) {
return AiResponse.unauthorized("您没有使用AI助手的权限");
}
// 3. 构建权限受限的上下文
String permissionContext = buildPermissionContext(permissions);
// 4. 权限受限的知识库检索
List<String> allowedContent = vectorStore.searchWithPermissions(
message, permissions.getDataAccessLevel(), 3);
// 5. 动态注入工具(只注入有权限的工具)
List<ToolDefinition> allowedTools = toolRegistry.getToolsForRole(permissions.getRole());
// 6. 构建最终Prompt
String systemPrompt = buildPermissionedSystemPrompt(permissions, allowedContent);
// 7. 调用LLM
String response = callLlmWithPermissions(systemPrompt, message, allowedTools);
// 8. 输出过滤(防止返回超权数据)
String filteredResponse = filterSensitiveContent(response, permissions);
// 审计日志
auditLog(userId, message, filteredResponse, permissions.getRole());
return AiResponse.success(filteredResponse);
}
private String buildPermissionContext(UserPermissions permissions) {
return String.format(
"当前用户角色:%s,数据访问级别:%s,可用功能:%s",
permissions.getRole(),
permissions.getDataAccessLevel(),
permissions.getAllowedFeatures().stream()
.map(Enum::name)
.collect(Collectors.joining(", "))
);
}
private String buildPermissionedSystemPrompt(
UserPermissions permissions,
List<String> allowedContent) {
StringBuilder prompt = new StringBuilder();
prompt.append("你是一个企业内部AI助手。\n\n");
// 角色特定指令
switch (permissions.getRole()) {
case MANAGER:
prompt.append("当前用户是管理层,可以访问团队数据和报告功能。\n");
break;
case DATA_ANALYST:
prompt.append("当前用户是数据分析师,可以执行数据查询和分析。\n");
break;
default:
prompt.append("当前用户是普通员工,只能访问个人数据和公开知识。\n");
}
// 数据访问限制
prompt.append("\n数据访问限制:\n");
prompt.append("- 只能提供当前用户有权查看的数据(").append(permissions.getDataAccessLevel()).append("级别及以下)\n");
prompt.append("- 如果被问及无权访问的数据,说'您没有权限查看该数据'\n");
prompt.append("- 不要推断或猜测无权数据的内容\n");
// 相关知识库内容
if (!allowedContent.isEmpty()) {
prompt.append("\n相关知识库内容(已按权限过滤):\n");
allowedContent.forEach(c -> prompt.append(c).append("\n"));
}
return prompt.toString();
}
private String callLlmWithPermissions(
String systemPrompt,
String userMessage,
List<ToolDefinition> tools) {
// 简化实现:实际需要根据工具列表动态配置LLM
return llm.generate(
SystemMessage.from(systemPrompt),
UserMessage.from(userMessage)
).content().text();
}
private String filterSensitiveContent(String response, UserPermissions permissions) {
// 后处理:检查响应中是否包含超权内容
if (permissions.getDataAccessLevel().ordinal() < AiPermissionModel.DataAccessLevel.CONFIDENTIAL.ordinal()) {
// 检测并脱敏机密信息
response = response.replaceAll("(密码|token|secret|key)[:=]\\s*\\S+",
"$1: [已脱敏]");
}
return response;
}
private void auditLog(String userId, String question, String response,
AiPermissionModel.AiRole role) {
log.info("AI访问审计: userId={}, role={}, questionLen={}, responseLen={}",
userId, role, question.length(), response.length());
}
}权限感知的向量存储
/**
* 带权限控制的向量存储
* 确保用户只能检索到有权限的文档
*/
@Service
@RequiredArgsConstructor
public class PermissionedVectorStore {
private final EmbeddingModel embeddingModel;
private final EmbeddingStore<TextSegment> vectorStore;
/**
* 权限感知的向量搜索
* 通过元数据过滤实现权限控制
*/
public List<String> searchWithPermissions(
String query,
AiPermissionModel.DataAccessLevel maxLevel,
int topK) {
float[] embedding = embeddingModel.embed(query);
// 构建权限过滤条件
// 只返回数据等级 <= 用户最高权限的文档
Filter permissionFilter = buildPermissionFilter(maxLevel);
return vectorStore.search(EmbeddingSearchRequest.builder()
.queryEmbedding(Embedding.from(embedding))
.maxResults(topK)
.minScore(0.7)
.filter(permissionFilter)
.build())
.matches().stream()
.map(m -> m.embedded().text())
.toList();
}
private Filter buildPermissionFilter(AiPermissionModel.DataAccessLevel maxLevel) {
// 允许访问所有数据级别 <= 当前用户权限级别的文档
// PUBLIC(0) INTERNAL(1) DEPARTMENT(2) ROLE_BASED(3) CONFIDENTIAL(4) SECRET(5)
int maxLevelInt = maxLevel.ordinal();
// 构建过滤器:dataLevel <= maxLevelInt
List<Filter> levelFilters = new ArrayList<>();
for (int i = 0; i <= maxLevelInt; i++) {
String levelName = AiPermissionModel.DataAccessLevel.values()[i].name();
levelFilters.add(metadataKey("dataAccessLevel").isEqualTo(levelName));
}
if (levelFilters.size() == 1) return levelFilters.get(0);
return or(levelFilters.toArray(new Filter[0]));
}
/**
* 入库时记录文档的权限级别
*/
public void addDocumentWithPermissions(
String content,
AiPermissionModel.DataAccessLevel accessLevel,
String department,
Map<String, String> additionalMetadata) {
Map<String, String> metadata = new HashMap<>(additionalMetadata);
metadata.put("dataAccessLevel", accessLevel.name());
metadata.put("department", department);
metadata.put("ingestTime", LocalDateTime.now().toString());
TextSegment segment = TextSegment.from(content, Metadata.from(metadata));
Embedding embedding = embeddingModel.embed(content);
vectorStore.add(embedding, segment);
}
}工具调用权限控制
/**
* AI工具的权限注册和控制
* 不同角色只能调用对应权限的工具
*/
@Service
@RequiredArgsConstructor
public class AiToolRegistry {
private final Map<AiPermissionModel.AiRole, List<ToolDefinition>> roleToolMap;
@PostConstruct
public void init() {
// 注册工具和对应的权限
Map<AiPermissionModel.AiRole, List<ToolDefinition>> map = new EnumMap<>(AiPermissionModel.AiRole.class);
// 公共工具(所有角色)
List<ToolDefinition> publicTools = List.of(
new ToolDefinition("getCurrentDate", "获取当前日期", List.of()),
new ToolDefinition("searchPublicKnowledge", "查询公开知识库", List.of())
);
// 员工工具
List<ToolDefinition> employeeTools = new ArrayList<>(publicTools);
employeeTools.add(new ToolDefinition("getMyTasks", "查询个人任务", List.of()));
employeeTools.add(new ToolDefinition("searchInternalKB", "查询内部知识库", List.of()));
// 管理层工具
List<ToolDefinition> managerTools = new ArrayList<>(employeeTools);
managerTools.add(new ToolDefinition("getTeamMetrics", "查询团队指标", List.of()));
managerTools.add(new ToolDefinition("generateReport", "生成报告", List.of()));
// 管理员工具
List<ToolDefinition> adminTools = new ArrayList<>(managerTools);
adminTools.add(new ToolDefinition("getUserActivity", "查询用户活动", List.of()));
adminTools.add(new ToolDefinition("modifySystemConfig", "修改系统配置", List.of()));
map.put(AiPermissionModel.AiRole.PUBLIC_USER, publicTools);
map.put(AiPermissionModel.AiRole.EMPLOYEE, employeeTools);
map.put(AiPermissionModel.AiRole.MANAGER, managerTools);
map.put(AiPermissionModel.AiRole.ADMIN, adminTools);
map.put(AiPermissionModel.AiRole.DATA_ANALYST, new ArrayList<>(employeeTools) {{
add(new ToolDefinition("executeSqlQuery", "执行SQL查询", List.of()));
add(new ToolDefinition("generateDataVisualization", "生成数据可视化", List.of()));
}});
roleToolMap.putAll(map);
}
public List<ToolDefinition> getToolsForRole(AiPermissionModel.AiRole role) {
return roleToolMap.getOrDefault(role, List.of());
}
/**
* 运行时检查工具调用权限
*/
public boolean canUseToolRaw(String userId, String toolName) {
UserPermissions permissions = permissionService.getPermissions(userId);
return getToolsForRole(permissions.getRole()).stream()
.anyMatch(t -> t.name().equals(toolName));
}
@Autowired
private UserPermissionService permissionService;
public record ToolDefinition(String name, String description, List<String> requiredParams) {}
}权限违规检测
/**
* 权限违规监控
* 检测可能的权限绕过尝试
*/
@Component
@Slf4j
public class PermissionViolationDetector {
// 越权尝试的典型模式
private static final List<Pattern> BYPASS_PATTERNS = List.of(
Pattern.compile("忽略.*权限|绕过.*限制"),
Pattern.compile("假设你是.*管理员|以.*身份"),
Pattern.compile("你现在可以.*访问|临时授权"),
Pattern.compile("system.*override|admin.*mode|debug.*mode")
);
/**
* 检测请求是否包含权限绕过尝试
*/
public ViolationCheckResult checkViolationAttempt(String userId, String request) {
for (Pattern pattern : BYPASS_PATTERNS) {
if (pattern.matcher(request.toLowerCase()).find()) {
log.warn("检测到权限绕过尝试: userId={}, request_preview={}",
userId, request.substring(0, Math.min(100, request.length())));
// 记录安全事件
recordSecurityEvent(userId, request, "PERMISSION_BYPASS_ATTEMPT");
return new ViolationCheckResult(true,
"该请求包含可能的权限绕过尝试,已记录并拦截。");
}
}
return new ViolationCheckResult(false, "");
}
private void recordSecurityEvent(String userId, String request, String eventType) {
// 记录到安全事件日志
log.error("安全事件[{}]: userId={}, time={}", eventType, userId, LocalDateTime.now());
}
public record ViolationCheckResult(boolean isViolation, String message) {}
}AI权限管理的核心思路是:在LLM之前就过滤和限制,而不是依赖LLM自己判断权限。
LLM不是权限执行层,它是能力层。权限检查必须在数据检索、工具调用这些环节做好,不能寄希望于Prompt里的"你不能做什么"指令——因为那些指令有可能被绕过。
