package server import ( "context" "errors" "fmt" "net" "time" "perms-system-server/internal/consts" authHelper "perms-system-server/internal/logic/auth" pub "perms-system-server/internal/logic/pub" "perms-system-server/internal/middleware" userModel "perms-system-server/internal/model/user" "perms-system-server/internal/svc" "perms-system-server/pb" "github.com/zeromicro/go-zero/core/limit" "github.com/zeromicro/go-zero/core/logx" "golang.org/x/crypto/bcrypt" "google.golang.org/grpc/codes" "google.golang.org/grpc/peer" "google.golang.org/grpc/status" ) // unknownPeerBucket 当无法解析对端 IP 时共享的限流桶 key。 // 生产环境 gRPC-over-TCP 必然有 peer.Addr,正常流量不会落到这里;此常量仅为 in-process/socket // 等边缘路径兜底,避免 M-7 审计指出的"按 p.Addr.String() 取完整 host:port 导致限流形同虚设"的 // 随端口漂移问题。共享同一个 key 会放大 DoS 面(所有未知 peer 共用一个计数器),但在此类路径 // 不走真实业务流量的前提下收益足够。 const unknownPeerBucket = "unknown" // extractClientIP 从 gRPC context 中提取对端 IP。显式剥离端口号(M-7):gRPC 的 p.Addr.String() // 形如 "1.2.3.4:54321",端口每次连接都变,若直接作为限流 key 相当于没限流。 // 解析失败返回 error,由上层按场景决定是 fail-close(RefreshToken 敏感路径)还是降级到 unknown 桶 // (VerifyToken 契约层约束不允许返回 error)。 func extractClientIP(ctx context.Context) (string, error) { p, ok := peer.FromContext(ctx) if !ok || p == nil || p.Addr == nil { return "", errors.New("peer not identifiable") } host, _, err := net.SplitHostPort(p.Addr.String()) if err != nil || host == "" { return "", errors.New("peer address invalid") } return host, nil } // PermServer 权限管理系统 gRPC 服务实现,供接入产品的服务端调用。 type PermServer struct { svcCtx *svc.ServiceContext pb.UnimplementedPermServiceServer } func NewPermServer(svcCtx *svc.ServiceContext) *PermServer { return &PermServer{svcCtx: svcCtx} } // SyncPermissions 同步权限声明。产品服务端通过 appKey/appSecret 认证后批量同步权限定义(新增/更新/禁用不在列表中的权限)。 func (s *PermServer) SyncPermissions(ctx context.Context, req *pb.SyncPermissionsReq) (*pb.SyncPermissionsResp, error) { items := make([]pub.SyncPermItem, len(req.Perms)) for i, p := range req.Perms { items[i] = pub.SyncPermItem{Code: p.Code, Name: p.Name, Remark: p.Remark} } result, err := pub.ExecuteSyncPerms(ctx, s.svcCtx, req.AppKey, req.AppSecret, items) if err != nil { if se, ok := err.(*pub.SyncPermsError); ok { // 审计 M-2:404 是 tx 内 LockByCodeTx 命中 sqlx.ErrNotFound(产品行被并发删除) // 的语义,先于 400/401 前的前置校验放行后才可能出现。接入方 SDK 对 NotFound 一般 // 配置"按业务未命中处理/不重试",若在这里落到 default 分支被统一成 codes.Internal, // 接入方会把一次正常的"产品不存在"当作系统故障 page 值班,扭曲重试与告警语义。 switch se.Code { case 400: return nil, status.Error(codes.InvalidArgument, se.Message) case 401: return nil, status.Error(codes.Unauthenticated, se.Message) case 403: return nil, status.Error(codes.PermissionDenied, se.Message) case 404: return nil, status.Error(codes.NotFound, se.Message) case 409: return nil, status.Error(codes.Aborted, se.Message) default: return nil, status.Error(codes.Internal, se.Message) } } return nil, status.Error(codes.Internal, "同步权限失败") } return &pb.SyncPermissionsResp{Added: result.Added, Updated: result.Updated, Disabled: result.Disabled}, nil } // Login 产品端登录。产品成员通过用户名密码 + productCode 登录,返回 JWT 令牌对及用户权限信息。受 IP 维度限流保护。 func (s *PermServer) Login(ctx context.Context, req *pb.LoginReq) (*pb.LoginResp, error) { clientIP, ipErr := extractClientIP(ctx) if ipErr != nil { // 审计 M-7 的核心修复是"把 host:port 剥成 host,避免端口漂移让限流失效"; // 生产环境 gRPC 必有 peer,这里走不到;in-process/单测等边缘路径回落到共享 unknown 桶, // 上层仍会继续执行用户名级的 UsernameLoginLimit,不会造成防护真空。 clientIP = unknownPeerBucket } if s.svcCtx.GrpcLoginLimiter != nil { code, _ := s.svcCtx.GrpcLoginLimiter.Take(fmt.Sprintf("grpc:login:%s", clientIP)) if code == limit.OverQuota { return nil, status.Error(codes.ResourceExhausted, "请求过于频繁,请稍后再试") } } if req.ProductCode == "" { return nil, status.Error(codes.InvalidArgument, "productCode不能为空") } result, err := pub.ValidateProductLogin(ctx, s.svcCtx, req.Username, req.Password, req.ProductCode, clientIP) if err != nil { if le, ok := err.(*pub.LoginError); ok { switch le.Code { case 400: return nil, status.Error(codes.InvalidArgument, le.Message) case 401: return nil, status.Error(codes.Unauthenticated, le.Message) case 403: return nil, status.Error(codes.PermissionDenied, le.Message) case 429: return nil, status.Error(codes.ResourceExhausted, le.Message) } } return nil, status.Error(codes.Internal, "登录失败") } ud := result.UserDetails return &pb.LoginResp{ AccessToken: result.AccessToken, RefreshToken: result.RefreshToken, Expires: time.Now().Unix() + s.svcCtx.Config.Auth.AccessExpire, UserId: ud.UserId, Username: ud.Username, Nickname: ud.Nickname, MemberType: ud.MemberType, Perms: ud.Perms, }, nil } // RefreshToken 刷新令牌。使用有效的 refreshToken 换取新的令牌对,同时原子 CAS 递增 tokenVersion // 使旧令牌即时失效(单会话轮转)。受 IP 维度限流保护,防止签名爆破和并发刷新被用于会话劫持。 func (s *PermServer) RefreshToken(ctx context.Context, req *pb.RefreshTokenReq) (*pb.RefreshTokenResp, error) { clientIP, ipErr := extractClientIP(ctx) if ipErr != nil { // 和 Login 相同,IP 解析失败走共享 unknown 桶;后续 CAS(IncrementTokenVersionIfMatch) // 和 per-user TokenOpLimiter 仍然兜底 session 劫持路径。 clientIP = unknownPeerBucket } if s.svcCtx.GrpcRefreshLimiter != nil { code, _ := s.svcCtx.GrpcRefreshLimiter.Take(fmt.Sprintf("grpc:refresh:%s", clientIP)) if code == limit.OverQuota { return nil, status.Error(codes.ResourceExhausted, "请求过于频繁,请稍后再试") } } claims, err := authHelper.ParseRefreshToken(req.RefreshToken, s.svcCtx.Config.Auth.RefreshSecret) if err != nil { return nil, status.Error(codes.Unauthenticated, "refreshToken无效或已过期") } productCode := claims.ProductCode if req.ProductCode != "" && req.ProductCode != productCode { return nil, status.Error(codes.InvalidArgument, "刷新令牌不允许切换产品") } ud, err := s.svcCtx.UserDetailsLoader.Load(ctx, claims.UserId, productCode) if err != nil { // 与"用户已删除"区分:基础设施短时不可用走 Unavailable,token 不作废让客户端重试 // (见审计 M-1)。 return nil, status.Error(codes.Unavailable, "服务暂时不可用,请稍后重试") } if ud.Username == "" { return nil, status.Error(codes.Unauthenticated, "用户不存在或已被删除") } if ud.Status != consts.StatusEnabled { return nil, status.Error(codes.PermissionDenied, "账号已被冻结") } if productCode != "" && ud.ProductStatus != consts.StatusEnabled { return nil, status.Error(codes.PermissionDenied, "该产品已被禁用") } if productCode != "" && !ud.IsSuperAdmin && ud.MemberType == "" { return nil, status.Error(codes.PermissionDenied, "您已不是该产品的成员") } if claims.TokenVersion != ud.TokenVersion { return nil, status.Error(codes.Unauthenticated, "登录状态已失效,请重新登录") } if s.svcCtx.TokenOpLimiter != nil { code, _ := s.svcCtx.TokenOpLimiter.Take(fmt.Sprintf("grpc-refresh-u:%d", claims.UserId)) if code == limit.OverQuota { return nil, status.Error(codes.ResourceExhausted, "刷新操作过于频繁,请稍后再试") } } // 审计 M-3:CAS 推进 tokenVersion 和签新令牌必须全部成功才能响应客户端,否则会出现 // tokenVersion 已+1 但客户端仍拿着旧 refreshToken → 下一次刷新必 401 被强制登出 // 的"非预期登出"事件(会污染会话劫持告警)。改为"先试签 → 再 CAS": // (a) 拿 claims.TokenVersion+1 预试签发 access/refresh;签名若失败(HMAC 只有 OOM 等 // 极端情况才会失败)直接 500,DB 状态完全不动。 // (b) 两个 token 都成功后再做 IncrementTokenVersionIfMatch 做并发唯一赢家 CAS;CAS 失败走 // 原来的 401/500 分支,客户端拿着的旧 refreshToken 仍然有效。 // (c) CAS 赢家在返回前 Clean 缓存,保证 caller 下一次 Load 读到的是 DB 最新 tokenVersion。 // 注意:由于 CAS 的新 version 一定等于 claims.TokenVersion + 1(见 IncrementTokenVersionIfMatch // 的 UPDATE 语义),这里直接按 claims.TokenVersion+1 预签即可,CAS 成功返回的 newVersion // 只用于 assert。 predictedVersion := claims.TokenVersion + 1 accessToken, err := authHelper.GenerateAccessToken( s.svcCtx.Config.Auth.AccessSecret, s.svcCtx.Config.Auth.AccessExpire, ud.UserId, ud.Username, ud.ProductCode, ud.MemberType, predictedVersion, ) if err != nil { return nil, status.Error(codes.Internal, "生成token失败") } newRefreshToken, err := authHelper.GenerateRefreshTokenWithExpiry( s.svcCtx.Config.Auth.RefreshSecret, claims.ExpiresAt.Time, ud.UserId, ud.ProductCode, predictedVersion, ) if err != nil { return nil, status.Error(codes.Internal, "生成token失败") } newVersion, err := s.svcCtx.SysUserModel.IncrementTokenVersionIfMatch(ctx, claims.UserId, ud.Username, claims.TokenVersion) if err != nil { if errors.Is(err, userModel.ErrTokenVersionMismatch) { return nil, status.Error(codes.Unauthenticated, "登录状态已失效,请重新登录") } return nil, status.Error(codes.Internal, "刷新token失败") } if newVersion != predictedVersion { // 防御:CAS 成功时服务端约定 +1,实际不一致说明上游 SQL 实现漂移。告警后直接要求重登, // 保证客户端不会被发一个 tokenVersion 对不上的 token。 logx.WithContext(ctx).Errorw("refresh token version prediction mismatch", logx.Field("audit", "refresh_token_version_mismatch"), logx.Field("userId", claims.UserId), logx.Field("claimed", claims.TokenVersion), logx.Field("predicted", predictedVersion), logx.Field("actual", newVersion), ) return nil, status.Error(codes.Unauthenticated, "登录状态已失效,请重新登录") } s.svcCtx.UserDetailsLoader.Clean(ctx, claims.UserId) return &pb.RefreshTokenResp{ AccessToken: accessToken, RefreshToken: newRefreshToken, Expires: time.Now().Unix() + s.svcCtx.Config.Auth.AccessExpire, }, nil } // VerifyToken 验证令牌。校验 accessToken 的有效性(签名、过期、用户状态、产品状态、成员资格、tokenVersion), // 有效时返回用户身份和权限信息。受 IP 维度限流保护,防止下游被攻破后把权限中心当作 token oracle 做爆破。 // // 注意:本方法对外契约是"任何畸形/非法 token 都只返回 Valid=false,不返回 gRPC 错误"(见 fuzz 契约测试), // 因此 IP 解析失败时不能走 fail-close,改为降级到共享 "unknown" 限流桶——仍然有限速,但不破坏上游产品网关 // 的稳定错误分类;真正过载时用 ResourceExhausted 响应。 func (s *PermServer) VerifyToken(ctx context.Context, req *pb.VerifyTokenReq) (*pb.VerifyTokenResp, error) { clientIP, ipErr := extractClientIP(ctx) if ipErr != nil { clientIP = "unknown" } if s.svcCtx.GrpcVerifyLimiter != nil { code, _ := s.svcCtx.GrpcVerifyLimiter.Take(fmt.Sprintf("grpc:verify:%s", clientIP)) if code == limit.OverQuota { return nil, status.Error(codes.ResourceExhausted, "请求过于频繁,请稍后再试") } } token, err := middleware.ParseWithHMAC(req.AccessToken, s.svcCtx.Config.Auth.AccessSecret, &middleware.Claims{}) if err != nil || !token.Valid { logx.WithContext(ctx).Infof("verifyToken fail reason=invalid_token") return &pb.VerifyTokenResp{Valid: false}, nil } claims, ok := token.Claims.(*middleware.Claims) if !ok || claims.TokenType != consts.TokenTypeAccess { logx.WithContext(ctx).Infof("verifyToken fail reason=bad_claims") return &pb.VerifyTokenResp{Valid: false}, nil } ud, err := s.svcCtx.UserDetailsLoader.Load(ctx, claims.UserId, claims.ProductCode) if err != nil { // VerifyToken 的对外契约是"任何 token 问题只回 Valid=false,不抛 gRPC 错误"。但基础设施 // 故障不属于"token 问题"——同化为 Valid=false 会让下游把合法用户踢出登录(见审计 M-1)。 // 走 Unavailable,由下游按瞬时故障重试而不是据此清 token。 logx.WithContext(ctx).Errorf("verifyToken: load user details failed: %v", err) return nil, status.Error(codes.Unavailable, "服务暂时不可用,请稍后重试") } if ud.Username == "" { logx.WithContext(ctx).Infof("verifyToken fail userId=%d reason=user_not_found", claims.UserId) return &pb.VerifyTokenResp{Valid: false}, nil } if ud.Status != consts.StatusEnabled { logx.WithContext(ctx).Infof("verifyToken fail userId=%d reason=user_disabled", claims.UserId) return &pb.VerifyTokenResp{Valid: false}, nil } if claims.TokenVersion != ud.TokenVersion { logx.WithContext(ctx).Infof("verifyToken fail userId=%d reason=token_version_mismatch", claims.UserId) return &pb.VerifyTokenResp{Valid: false}, nil } if claims.ProductCode != "" && ud.ProductStatus != consts.StatusEnabled { logx.WithContext(ctx).Infof("verifyToken fail userId=%d reason=product_disabled product=%s", claims.UserId, claims.ProductCode) return &pb.VerifyTokenResp{Valid: false}, nil } if claims.ProductCode != "" && !ud.IsSuperAdmin && ud.MemberType == "" { logx.WithContext(ctx).Infof("verifyToken fail userId=%d reason=not_member product=%s", claims.UserId, claims.ProductCode) return &pb.VerifyTokenResp{Valid: false}, nil } return &pb.VerifyTokenResp{ Valid: true, UserId: ud.UserId, Username: ud.Username, MemberType: ud.MemberType, Perms: ud.Perms, ProductCode: claims.ProductCode, }, nil } // GetUserPerms 查询用户权限。产品服务端通过 appKey/appSecret 认证后查询指定用户在该产品下的成员类型和权限列表,用于产品侧的权限网关判定。 func (s *PermServer) GetUserPerms(ctx context.Context, req *pb.GetUserPermsReq) (*pb.GetUserPermsResp, error) { product, err := s.svcCtx.SysProductModel.FindOneByAppKey(ctx, req.AppKey) if err != nil { return nil, status.Error(codes.Unauthenticated, "无效的appKey") } if err := bcrypt.CompareHashAndPassword([]byte(product.AppSecret), []byte(req.AppSecret)); err != nil { return nil, status.Error(codes.Unauthenticated, "appSecret验证失败") } if product.Status != consts.StatusEnabled { return nil, status.Error(codes.PermissionDenied, "产品已被禁用") } if product.Code != req.ProductCode { return nil, status.Error(codes.InvalidArgument, "appKey与productCode不匹配") } ud, err := s.svcCtx.UserDetailsLoader.Load(ctx, req.UserId, req.ProductCode) if err != nil { return nil, status.Error(codes.Unavailable, "服务暂时不可用,请稍后重试") } if ud.Username == "" { return nil, status.Error(codes.NotFound, "用户不存在") } if ud.Status != consts.StatusEnabled { return nil, status.Error(codes.PermissionDenied, "用户已被冻结") } if !ud.IsSuperAdmin && ud.MemberType == "" { return nil, status.Error(codes.PermissionDenied, "用户不是该产品的有效成员") } return &pb.GetUserPermsResp{ MemberType: ud.MemberType, Perms: ud.Perms, }, nil }