权限系统深度解析教程

import "github.com/casbin/casbin/v2"

// 检查权限
func CheckPermission(enforcer *casbin.Enforcer, user string, resource string, action string) bool {
    allowed, err := enforcer.Enforce(user, resource, action)
    if err != nil {
        log.Errorf("check permission failed: %v", err)
        return false
    }
    return allowed
}

// 使用示例
if !CheckPermission(enforcer, "alice", "/admin/v1/article/123", "DELETE") {
    return errors.New("permission denied")
}

// 添加策略
enforcer.AddPolicy("editor", "/admin/v1/article/*", "DELETE")

// 删除策略
enforcer.RemovePolicy("viewer", "/admin/v1/article/*", "POST")

// 重新加载所有策略
enforcer.LoadPolicy()

import "github.com/open-policy-agent/opa/rego"

// 创建查询
query, err := rego.New(
    rego.Query("data.authz.allow"),
    rego.Input(input),
    rego.Load([]string{"./policies"}, nil),
).PrepareForEval(ctx)

// 执行评估
results, err := query.Eval(ctx)
if err != nil {
    return false, err
}

// 检查结果
allowed := len(results) > 0 && len(results[0].Expressions) > 0 && 
           results[0].Expressions[0].Value == true

import "github.com/ory/keto-client-go"

// 检查权限
func CheckAccess(ketoClient *keto.Client, subject string, relation string, object string) bool {
    resp, err := ketoClient.Check(context.Background(), &keto.CheckRequest{
        Namespace: "article",
        Object: object,
        Relation: relation,
        Subject: &keto.Subject{
            Id: subject,
        },
    })
    
    if err != nil {
        log.Errorf("check access failed: %v", err)
        return false
    }
    
    return resp.Allowed
}

// 使用示例
if !CheckAccess(ketoClient, "Alice", "viewer", "article:123") {
    return errors.New("access denied")
}

// 创建用户级 Viewer
viewer := entgo.NewUserViewerContext(ctx, userID, dataScope)

// 在查询时自动应用过滤
articles, err := client.Article.Query().
    Where(article.HasAuthorWith(user.ID(userID))).
    All(viewer)

// 从 Context 获取当前用户的数据权限范围
dataScope := auth.GetDataScopeFromContext(ctx)

switch dataScope {
case identityV1.DataScope_SELF:
    query = query.Where(article.AuthorID(userID))
    
case identityV1.DataScope_UNIT_ONLY:
    query = query.Where(article.HasAuthorWith(
        user.HasOrgUnitsWith(orgunit.ID(orgUnitID)),
    ))
    
case identityV1.DataScope_ALL:
    // 不过滤
    
// ...
}

// 添加角色权限时记录日志
func (s *RoleService) AddPermissions(ctx context.Context, req *AddPermissionsRequest) error {
    // 执行添加操作
    err := s.repo.AddPermissions(ctx, req.RoleId, req.PermissionIds)
    if err != nil {
        return err
    }
    
    // 记录审计日志
    auditLogRepo.Create(ctx, &permissionV1.PermissionAuditLog{
        Action:      "add_permissions",
        RoleId:      req.RoleId,
        PermissionIds: req.PermissionIds,
        OperatorId:  auth.GetUserID(ctx),
        CreatedAt:   time.Now(),
    })
    
    return nil
}

// 获取用户权限（带缓存）
func GetUserPermissions(ctx context.Context, userID uint32) ([]string, error) {
    cacheKey := fmt.Sprintf("user:%d:permissions", userID)
    
    // 尝试从缓存获取
    cached, err := redis.Get(ctx, cacheKey).Result()
    if err == nil {
        var permissions []string
        json.Unmarshal([]byte(cached), &permissions)
        return permissions, nil
    }
    
    // 从数据库查询
    permissions, err := queryPermissionsFromDB(userID)
    if err != nil {
        return nil, err
    }
    
    // 写入缓存（TTL 1 小时）
    data, _ := json.Marshal(permissions)
    redis.Set(ctx, cacheKey, data, time.Hour)
    
    return permissions, nil
}

// ❌ 错误做法：循环检查
for _, resource := range resources {
    if !CheckPermission(user, resource, "view") {
        return errors.New("permission denied")
    }
}

// ✅ 正确做法：一次性获取所有权限
permissions := GetUserPermissions(user)
permissionSet := make(map[string]bool)
for _, p := range permissions {
    permissionSet[p] = true
}

for _, resource := range resources {
    if !permissionSet[resource] {
        return errors.New("permission denied")
    }
}