mirror of
https://github.com/Buriburizaem0n/nezha_domains.git
synced 2026-09-19 09:40:12 +00:00
fix(security): harden gRPC stream WAF, MCP fs arg validation, and PAT server binding
- ServeRPC now applies real-IP + WAF interceptors to streaming RPCs (RequestTask, IOStream), not just unary calls; without them authHandler.check saw an empty real IP so brute-force BlockIP counters never keyed on a source and the WAF block table was bypassed at the stream entrypoint. Factored real-IP resolution into ctxWithRealIP shared by unary and stream paths. - handleFsRead rejects negative offset/length; handleFsWrite validates if_match_sha256 is 64 hex chars before dispatch. - createAPIToken always verifies each server_id exists (even for admins), so an admin cannot bind a PAT to a not-yet-created server id that a future server would auto-inherit. - Add relay zero-length-chunk rejection regression test. Co-authored-by: cloudcode <cloudcode@users.noreply.github.com>
This commit is contained in:
@@ -120,14 +120,12 @@ func createAPIToken(c *gin.Context) (*model.APITokenCreateResponse, error) {
|
||||
}
|
||||
seenSrv[sid] = struct{}{}
|
||||
deduped = append(deduped, sid)
|
||||
if !callerIsAdmin(c) {
|
||||
server, _ := singleton.ServerShared.Get(sid)
|
||||
if server == nil {
|
||||
return nil, errors.New("server not found")
|
||||
}
|
||||
if !server.HasPermission(c) {
|
||||
return nil, errors.New("permission denied on server")
|
||||
}
|
||||
server, _ := singleton.ServerShared.Get(sid)
|
||||
if server == nil {
|
||||
return nil, errors.New("server not found")
|
||||
}
|
||||
if !callerIsAdmin(c) && !server.HasPermission(c) {
|
||||
return nil, errors.New("permission denied on server")
|
||||
}
|
||||
}
|
||||
req.ServerIDs = deduped
|
||||
|
||||
@@ -233,6 +233,21 @@ func TestCreateAPIToken_AdminCanIncludeAnyServerID(t *testing.T) {
|
||||
require.Equal(t, []uint64{77}, res.ServerIDs)
|
||||
}
|
||||
|
||||
func TestCreateAPIToken_AdminCannotIncludeNonexistentServerID(t *testing.T) {
|
||||
defer setupAPITokenTest(t)()
|
||||
defer installServerForAPIToken(t, 77, 999)() // only server 77 exists
|
||||
|
||||
c := ctxAsUser(1, model.RoleAdmin)
|
||||
bindJSON(c, model.APITokenCreateRequest{
|
||||
Name: "ops-future-bind",
|
||||
Scopes: []string{model.ScopeServerRead},
|
||||
ServerIDs: []uint64{4242}, // never-existed server id
|
||||
})
|
||||
_, err := createAPIToken(c)
|
||||
require.Error(t, err, "admin must not bind a PAT to a nonexistent server_id; a later-created server with that id would auto-inherit the grant")
|
||||
require.Contains(t, err.Error(), "server not found")
|
||||
}
|
||||
|
||||
func TestCreateAPIToken_MemberOwnServerIDIsAccepted(t *testing.T) {
|
||||
defer setupAPITokenTest(t)()
|
||||
defer installServerForAPIToken(t, 55, 10)() // user 10 owns server 55
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package controller
|
||||
|
||||
import (
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"time"
|
||||
@@ -144,6 +145,12 @@ func handleFsRead(c *gin.Context, raw json.RawMessage) (any, error) {
|
||||
if args.Path == "" {
|
||||
return nil, errMCPInvalidArgs("path required")
|
||||
}
|
||||
if args.Offset < 0 {
|
||||
return nil, errMCPInvalidArgs("offset must be >= 0")
|
||||
}
|
||||
if args.Length < 0 {
|
||||
return nil, errMCPInvalidArgs("length must be >= 0")
|
||||
}
|
||||
out, err := rpc.CallAgent(c.Request.Context(), args.ServerID, model.TaskTypeFsRead,
|
||||
model.FsReadRequest{
|
||||
Path: args.Path,
|
||||
@@ -189,6 +196,11 @@ func handleFsWrite(c *gin.Context, raw json.RawMessage) (any, error) {
|
||||
if args.Path == "" {
|
||||
return nil, errMCPInvalidArgs("path required")
|
||||
}
|
||||
if args.IfMatchSHA256 != "" {
|
||||
if _, decErr := hex.DecodeString(args.IfMatchSHA256); decErr != nil || len(args.IfMatchSHA256) != 64 {
|
||||
return nil, errMCPInvalidArgs("if_match_sha256 must be 64 hex chars")
|
||||
}
|
||||
}
|
||||
out, err := rpc.CallAgent(c.Request.Context(), args.ServerID, model.TaskTypeFsWrite,
|
||||
model.FsWriteRequest{
|
||||
Path: args.Path,
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
package controller
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"net/http"
|
||||
"testing"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
|
||||
"github.com/nezhahq/nezha/model"
|
||||
)
|
||||
|
||||
// writeZeroChunkFrame emits a well-formed NZTC chunk header that declares a
|
||||
// zero-length payload. A malicious or buggy agent can emit an unbounded run of
|
||||
// these: each frame is syntactically valid but carries no data, so a relay
|
||||
// that treats them as no-op `continue` never makes progress toward `remaining`
|
||||
// and never reaches the final NZTO frame — pinning a dashboard goroutine, gRPC
|
||||
// stream and spool tmpfile until the client disconnects.
|
||||
func writeZeroChunkFrame(out *bytes.Buffer) {
|
||||
writeChunkFrame(out, nil)
|
||||
}
|
||||
|
||||
// A download that declares size > 0 but then streams zero-length NZTC frames
|
||||
// must be rejected as a protocol violation, not relayed forever. The relay
|
||||
// must not accept a zero-length data frame while it still expects bytes.
|
||||
func TestRelayDownloadFrames_RejectsZeroLengthDataFrames(t *testing.T) {
|
||||
const size = int64(64)
|
||||
|
||||
var src bytes.Buffer
|
||||
// A burst of zero-length chunk frames. With the buggy `continue`, the
|
||||
// loop consumes all of these without decrementing `remaining`; once the
|
||||
// buffer drains it hits EOF on the next ReadFull and returns a *bad
|
||||
// gateway* error — but against a real (blocking) stream the same code
|
||||
// path loops forever. We assert the relay rejects the zero-length frame
|
||||
// the moment it sees one, before draining a long run of them.
|
||||
for i := 0; i < 1000; i++ {
|
||||
writeZeroChunkFrame(&src)
|
||||
}
|
||||
// Even if a valid chunk + final frame follow, the relay must already have
|
||||
// failed on the first zero-length data frame.
|
||||
writeChunkFrame(&src, bytes.Repeat([]byte{'x'}, int(size)))
|
||||
writeOKFrame(&src, uint64(size))
|
||||
|
||||
stream := &fixedSizeFrameStream{buf: src}
|
||||
|
||||
sink := newCountingDiscardWriter()
|
||||
gin.SetMode(gin.TestMode)
|
||||
c, _ := gin.CreateTestContext(sink)
|
||||
|
||||
err := relayDownloadFrames(c, stream, size)
|
||||
if err == nil {
|
||||
t.Fatalf("relayDownloadFrames accepted a stream of zero-length NZTC frames; a zero-length data frame while remaining>0 must be rejected to avoid an unbounded relay loop")
|
||||
}
|
||||
if sink.status == 0 || sink.status == http.StatusOK {
|
||||
t.Fatalf("a rejected transfer must set a non-200 status, got %d", sink.status)
|
||||
}
|
||||
}
|
||||
|
||||
// Sanity: a single zero-length leading frame is just as invalid; the relay
|
||||
// must not silently swallow it as progress.
|
||||
func TestRelayDownloadFrames_SingleZeroLengthFrameRejected(t *testing.T) {
|
||||
const size = int64(8)
|
||||
|
||||
var src bytes.Buffer
|
||||
writeZeroChunkFrame(&src)
|
||||
writeChunkFrame(&src, bytes.Repeat([]byte{'y'}, int(size)))
|
||||
writeOKFrame(&src, uint64(size))
|
||||
|
||||
stream := &fixedSizeFrameStream{buf: src}
|
||||
sink := newCountingDiscardWriter()
|
||||
gin.SetMode(gin.TestMode)
|
||||
c, _ := gin.CreateTestContext(sink)
|
||||
|
||||
if err := relayDownloadFrames(c, stream, size); err == nil {
|
||||
t.Fatalf("relayDownloadFrames must reject a zero-length data frame while remaining>0")
|
||||
}
|
||||
_ = model.MCPFsXferMagicChunk
|
||||
}
|
||||
+54
-15
@@ -30,7 +30,14 @@ func SetMCPKillSwitchObserver(fn func() bool) {
|
||||
}
|
||||
|
||||
func ServeRPC() *grpc.Server {
|
||||
server := grpc.NewServer(grpc.ChainUnaryInterceptor(getRealIp, waf))
|
||||
// Streaming RPCs (RequestTask, IOStream) need the same real-IP + WAF
|
||||
// gate as unary calls; without the stream interceptors authHandler.check
|
||||
// sees an empty real IP, so brute-force BlockIP counters never key on a
|
||||
// source and the WAF block table is bypassed at the stream entrypoint.
|
||||
server := grpc.NewServer(
|
||||
grpc.ChainUnaryInterceptor(getRealIp, waf),
|
||||
grpc.ChainStreamInterceptor(getRealIpStream, wafStream),
|
||||
)
|
||||
rpcService.NezhaHandlerSingleton = rpcService.NewNezhaHandler()
|
||||
// Install the IOStream revocation hook so ServerTransferShared can tear
|
||||
// down terminal/FM/NAT sessions held by the previous owner on every
|
||||
@@ -40,15 +47,7 @@ func ServeRPC() *grpc.Server {
|
||||
return server
|
||||
}
|
||||
|
||||
func waf(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
|
||||
realip, _ := ctx.Value(model.CtxKeyRealIP{}).(string)
|
||||
if err := model.CheckIP(singleton.DB, realip); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return handler(ctx, req)
|
||||
}
|
||||
|
||||
func getRealIp(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
|
||||
func ctxWithRealIP(ctx context.Context) (context.Context, error) {
|
||||
var ip, connectingIp string
|
||||
p, ok := peer.FromContext(ctx)
|
||||
if ok {
|
||||
@@ -60,22 +59,22 @@ func getRealIp(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler
|
||||
ctx = context.WithValue(ctx, model.CtxKeyConnectingIP{}, connectingIp)
|
||||
|
||||
if singleton.Conf.AgentRealIPHeader == "" {
|
||||
return handler(ctx, req)
|
||||
return ctx, nil
|
||||
}
|
||||
|
||||
if singleton.Conf.AgentRealIPHeader == model.ConfigUsePeerIP {
|
||||
if connectingIp == "" {
|
||||
return nil, fmt.Errorf("connecting ip not found")
|
||||
return ctx, fmt.Errorf("connecting ip not found")
|
||||
}
|
||||
} else {
|
||||
vals := metadata.ValueFromIncomingContext(ctx, singleton.Conf.AgentRealIPHeader)
|
||||
if len(vals) == 0 {
|
||||
return nil, fmt.Errorf("real ip header not found")
|
||||
return ctx, fmt.Errorf("real ip header not found")
|
||||
}
|
||||
var err error
|
||||
ip, err = utils.GetIPFromHeader(vals[0])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return ctx, err
|
||||
}
|
||||
}
|
||||
|
||||
@@ -83,10 +82,50 @@ func getRealIp(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler
|
||||
log.Printf("NEZHA>> gRPC Agent Real IP: %s, connecting IP: %s\n", ip, connectingIp)
|
||||
}
|
||||
|
||||
ctx = context.WithValue(ctx, model.CtxKeyRealIP{}, ip)
|
||||
return context.WithValue(ctx, model.CtxKeyRealIP{}, ip), nil
|
||||
}
|
||||
|
||||
func waf(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
|
||||
realip, _ := ctx.Value(model.CtxKeyRealIP{}).(string)
|
||||
if err := model.CheckIP(singleton.DB, realip); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return handler(ctx, req)
|
||||
}
|
||||
|
||||
func getRealIp(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
|
||||
ctx, err := ctxWithRealIP(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return handler(ctx, req)
|
||||
}
|
||||
|
||||
// realIPServerStream overrides Context() so stream handlers and
|
||||
// authHandler.check observe the resolved real IP, like the unary path.
|
||||
type realIPServerStream struct {
|
||||
grpc.ServerStream
|
||||
ctx context.Context
|
||||
}
|
||||
|
||||
func (s *realIPServerStream) Context() context.Context { return s.ctx }
|
||||
|
||||
func getRealIpStream(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
|
||||
ctx, err := ctxWithRealIP(ss.Context())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return handler(srv, &realIPServerStream{ServerStream: ss, ctx: ctx})
|
||||
}
|
||||
|
||||
func wafStream(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
|
||||
realip, _ := ss.Context().Value(model.CtxKeyRealIP{}).(string)
|
||||
if err := model.CheckIP(singleton.DB, realip); err != nil {
|
||||
return err
|
||||
}
|
||||
return handler(srv, ss)
|
||||
}
|
||||
|
||||
func DispatchTask(serviceSentinelDispatchBus <-chan *model.Service) {
|
||||
for task := range serviceSentinelDispatchBus {
|
||||
if task == nil {
|
||||
|
||||
Reference in New Issue
Block a user