25960901307

Committed 16 May 2026 11:29AM UTC coverage: 76.887% (+0.2%) from 76.64%

Build # 25960901307

Build Type

push

github

Committed by

et-nik

Commit Message

audit logs tests

Coverage Stats

45395 of 59041 relevant lines covered (76.89%)

33895.16 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.19

/internal/api/ws/attach/handler.go

package attach

import (
        "context"
        "encoding/json"
        "log/slog"
        "net/http"
        "os"
        "path/filepath"
        "time"

        "github.com/coder/websocket"
        "github.com/gameap/gameap/internal/api/base"
        serversbase "github.com/gameap/gameap/internal/api/servers/base"
        "github.com/gameap/gameap/internal/daemon"
        "github.com/gameap/gameap/internal/domain"
        "github.com/gameap/gameap/internal/filters"
        "github.com/gameap/gameap/internal/grpc/handlers"
        "github.com/gameap/gameap/internal/grpc/session"
        "github.com/gameap/gameap/internal/pubsub/channels"
        "github.com/gameap/gameap/internal/repositories"
        "github.com/gameap/gameap/internal/ws"
        "github.com/gameap/gameap/pkg/api"
        "github.com/gameap/gameap/pkg/auth"
        "github.com/gameap/gameap/pkg/idgen"
        "github.com/gameap/gameap/pkg/proto"
        "github.com/pkg/errors"
)

const (
        typeAttachInput   = "attach.input"
        typeAttachOutput  = "attach.output"
        typeAttachDetach  = "attach.detach"
        typeAttachStarted = "attach.started"

        legacyPollInterval = 500 * time.Millisecond
)

type daemonCommands interface {
        ExecuteCommand(
                ctx context.Context,
                node *domain.Node,
                command string,
                opts ...daemon.CommandServiceOption,
        ) (*daemon.CommandResult, error)
}

type fileService interface {
        Download(ctx context.Context, node *domain.Node, filePath string) ([]byte, error)
        Upload(
                ctx context.Context, node *domain.Node, filePath string,
                content []byte, perms os.FileMode, owner daemon.OwnerOptions,
        ) error
}

type Handler struct {
        serverFinder   *serversbase.ServerFinder
        abilityChecker *serversbase.AbilityChecker
        nodeRepo       repositories.NodeRepository
        hub            *ws.Hub
        originPatterns []string
        registry       *session.Registry
        attachHandler  *handlers.AttachHandler
        daemonCommands daemonCommands
        fileService    fileService
        responder      base.Responder
        logger         *slog.Logger
}

func NewHandler(
        serverRepo repositories.ServerRepository,
        nodeRepo repositories.NodeRepository,
        rbac base.RBAC,
        hub *ws.Hub,
        originPatterns []string,
        registry *session.Registry,
        attachHandler *handlers.AttachHandler,
        daemonCommands daemonCommands,
        fileService fileService,
        responder base.Responder,
) *Handler {
        return &Handler{
                serverFinder:   serversbase.NewServerFinder(serverRepo, rbac),
                abilityChecker: serversbase.NewAbilityChecker(rbac),
                nodeRepo:       nodeRepo,
                hub:            hub,
                originPatterns: originPatterns,
                registry:       registry,
                attachHandler:  attachHandler,
                daemonCommands: daemonCommands,
                fileService:    fileService,
                responder:      responder,
                logger:         slog.Default(),
        }
}

func (h *Handler) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
        ctx := r.Context()

        sess := auth.SessionFromContext(ctx)
        if !sess.IsAuthenticated() {
                h.responder.WriteError(ctx, rw, api.NewError(http.StatusUnauthorized, "user not authenticated"))

                return
        }

        input := api.NewInputReader(r)

        serverID, err := input.ReadUint("server")
        if err != nil {
                h.responder.WriteError(ctx, rw, api.WrapHTTPError(
                        errors.WithMessage(err, "invalid server id"),
                        http.StatusBadRequest,
                ))

                return
        }

        server, err := h.serverFinder.FindUserServer(ctx, sess.User, serverID)
        if err != nil {
                h.responder.WriteError(ctx, rw, err)

                return
        }

        if err = h.abilityChecker.CheckOrError(
                ctx,
                sess.User.ID,
                server.ID,
                []domain.AbilityName{domain.AbilityNameGameServerConsoleView},
        ); err != nil {
                h.responder.WriteError(ctx, rw, err)

                return
        }

        node, err := h.findNode(ctx, server.DSID)
        if err != nil {
                h.responder.WriteError(ctx, rw, err)

                return
        }

        nodeID := uint64(server.DSID)

        conn, err := ws.Accept(rw, r, &websocket.AcceptOptions{
                OriginPatterns: h.originPatterns,
        })
        if err != nil {
                h.logger.Warn("websocket accept failed", "error", err)

                return
        }

        canSend := h.canSendCommands(ctx, sess.User, server)

        if h.registry.IsConnectedAnywhere(nodeID) {
                sessionID := idgen.New()
                h.runAttachSession(ctx, conn, server, node, sessionID, sess.User, canSend)
        } else {
                h.runLegacyMode(ctx, conn, server, node, sess.User, canSend)
        }
}

func (h *Handler) runAttachSession(
        ctx context.Context,
        conn *websocket.Conn,
        server *domain.Server,
        node *domain.Node,
        sessionID string,
        user *domain.User,
        canSend bool,
) {
        client := ws.NewClient(ctx, conn, h.hub, nil, h.logger)

        startedTopic := ws.ChannelToTopic(channels.BuildRealtimeAttachStartedChannel(sessionID))
        outputTopic := ws.ChannelToTopic(channels.BuildRealtimeAttachOutputChannel(sessionID))
        closedTopic := ws.ChannelToTopic(channels.BuildRealtimeAttachClosedChannel(sessionID))

        h.hub.Register(client, startedTopic, outputTopic, closedTopic)
        h.attachHandler.TrackAttachSession(sessionID, uint64(server.ID))

        nodeID := uint64(node.ID)

        msgHandler := h.newMessageHandler(ctx, client, server, node, user, sessionID, canSend)
        client.SetMessageHandler(msgHandler)

        if err := h.registry.SendAttachRequest(ctx, nodeID, &proto.AttachRequest{
                SessionId: sessionID,
                ServerId:  uint64(server.ID),
        }); err != nil {
                h.logger.Warn("failed to send attach request",
                        "server_id", server.ID,
                        "session_id", sessionID,
                        "error", err,
                )
                client.SendMessage(ws.NewErrorMessage("failed to attach to server console"))
        }

        client.Run()

        _ = h.registry.SendAttachDetach(context.Background(), nodeID, &proto.AttachDetach{
                SessionId: sessionID,
                Reason:    "client disconnected",
        })
        h.attachHandler.UntrackAttachSession(sessionID)
}

func (h *Handler) newMessageHandler(
        ctx context.Context,
        client *ws.Client,
        server *domain.Server,
        node *domain.Node,
        user *domain.User,
        sessionID string,
        canSend bool,
) ws.MessageHandler {
        nodeID := uint64(node.ID)

        return func(_ context.Context, msg *ws.InboundMessage) {
                switch msg.Type {
                case typeAttachInput:
                        if !canSend {
                                client.SendMessage(ws.NewErrorMessage("permission denied: cannot send input"))

                                return
                        }

                        if err := h.abilityChecker.CheckOrError(
                                ctx,
                                user.ID,
                                server.ID,
                                []domain.AbilityName{domain.AbilityNameGameServerConsoleSend},
                        ); err != nil {
                                client.SendMessage(ws.NewErrorMessage("permission denied: cannot send input"))

                                return
                        }

                        var payload attachInputPayload
                        if err := json.Unmarshal(msg.Payload, &payload); err != nil {
                                return
                        }

                        if err := h.registry.SendAttachInput(ctx, nodeID, &proto.AttachInput{
                                SessionId: sessionID,
                                Data:      payload.Data,
                        }); err != nil {
                                h.logger.Warn("failed to send attach input",
                                        "session_id", sessionID,
                                        "error", err,
                                )
                        }

                case typeAttachDetach:
                        _ = h.registry.SendAttachDetach(ctx, nodeID, &proto.AttachDetach{
                                SessionId: sessionID,
                                Reason:    "user detached",
                        })
                        client.Close()
                }
        }
}

func (h *Handler) findNode(ctx context.Context, nodeID uint) (*domain.Node, error) {
        nodes, err := h.nodeRepo.Find(ctx, &filters.FindNode{
                IDs: []uint{nodeID},
        }, nil, &filters.Pagination{
                Limit: 1,
        })
        if err != nil {
                return nil, errors.WithMessage(err, "failed to find node")
        }

        if len(nodes) == 0 {
                return nil, api.NewNotFoundError("node not found")
        }

        return &nodes[0], nil
}

func (h *Handler) canSendCommands(ctx context.Context, user *domain.User, server *domain.Server) bool {
        err := h.abilityChecker.CheckOrError(
                ctx,
                user.ID,
                server.ID,
                []domain.AbilityName{domain.AbilityNameGameServerConsoleSend},
        )

        return err == nil
}

type attachInputPayload struct {
        Data []byte `json:"data"`
}

type attachStartedPayload struct {
        SessionID string `json:"session_id"`
        ServerID  uint64 `json:"server_id"`
}

type attachOutputPayload struct {
        Data []byte `json:"data"`
}

// Legacy mode for daemons not connected via gRPC.

func (h *Handler) runLegacyMode(
        ctx context.Context,
        conn *websocket.Conn,
        server *domain.Server,
        node *domain.Node,
        user *domain.User,
        canSend bool,
) {
        client := ws.NewClient(ctx, conn, h.hub, nil, h.logger)

        msgHandler := h.newLegacyMessageHandler(ctx, client, server, node, user, canSend)
        client.SetMessageHandler(msgHandler)

        client.SendMessage(ws.NewOutboundMessage(typeAttachStarted, attachStartedPayload{
                SessionID: idgen.New(),
                ServerID:  uint64(server.ID),
        }))

        lastContent := h.sendLegacyHistory(ctx, client, server, node)

        poller := &legacyAttachPoller{
                client:      client,
                fileService: h.fileService,
                node:        node,
                serverDir:   server.Dir,
                logger:      h.logger,
                lastContent: lastContent,
        }
        go poller.run(ctx)

        client.Run()
}

func (h *Handler) newLegacyMessageHandler(
        ctx context.Context,
        client *ws.Client,
        server *domain.Server,
        node *domain.Node,
        user *domain.User,
        canSend bool,
) ws.MessageHandler {
        return func(_ context.Context, msg *ws.InboundMessage) {
                switch msg.Type {
                case typeAttachInput:
                        if !canSend {
                                client.SendMessage(ws.NewErrorMessage("permission denied: cannot send input"))

                                return
                        }

                        if err := h.abilityChecker.CheckOrError(
                                ctx, user.ID, server.ID,
                                []domain.AbilityName{domain.AbilityNameGameServerConsoleSend},
                        ); err != nil {
                                client.SendMessage(ws.NewErrorMessage("permission denied: cannot send input"))

                                return
                        }

                        var payload attachInputPayload
                        if err := json.Unmarshal(msg.Payload, &payload); err != nil {
                                return
                        }

                        h.sendLegacyInput(ctx, client, server, node, string(payload.Data))

                case typeAttachDetach:
                        client.Close()
                }
        }
}

func (h *Handler) sendLegacyInput(
        ctx context.Context, client *ws.Client, server *domain.Server, node *domain.Node, input string,
) {
        if node.ScriptSendCommand != nil && *node.ScriptSendCommand != "" {
                cmd := server.ReplaceServerShortcodes(node, *node.ScriptSendCommand, map[string]string{
                        "command": input,
                })

                if _, err := h.daemonCommands.ExecuteCommand(ctx, node, cmd); err != nil {
                        h.logger.Warn("failed to execute send command script",
                                "server_id", server.ID, "error", err)
                        client.SendMessage(ws.NewErrorMessage("failed to send input"))
                }

                return
        }

        inputPath := filepath.Join(server.Dir, "input.txt")

        err := h.fileService.Upload(
                ctx, node, inputPath, []byte(input), 0o644, daemon.OwnerFromServer(server),
        )
        if err != nil {
                h.logger.Warn("failed to upload input", "server_id", server.ID, "error", err)
                client.SendMessage(ws.NewErrorMessage("failed to send input"))
        }
}

func (h *Handler) sendLegacyHistory(
        ctx context.Context, client *ws.Client, server *domain.Server, node *domain.Node,
) string {
        if node.ScriptGetConsole != nil && *node.ScriptGetConsole != "" {
                cmd := server.ReplaceServerShortcodes(node, *node.ScriptGetConsole, nil)

                result, err := h.daemonCommands.ExecuteCommand(ctx, node, cmd)
                if err == nil && result.Output != "" {
                        client.SendMessage(ws.NewOutboundMessage(typeAttachOutput, attachOutputPayload{
                                Data: []byte(result.Output),
                        }))

                        return result.Output
                }
        }

        outputPath := filepath.Join(server.Dir, "output.txt")

        content, err := h.fileService.Download(ctx, node, outputPath)
        if err != nil {
                return ""
        }

        const maxBytes = 65536
        if len(content) > maxBytes {
                content = content[len(content)-maxBytes:]
        }

        if len(content) > 0 {
                client.SendMessage(ws.NewOutboundMessage(typeAttachOutput, attachOutputPayload{
                        Data: content,
                }))
        }

        return string(content)
}

type legacyAttachPoller struct {
        client      *ws.Client
        fileService fileService
        node        *domain.Node
        serverDir   string
        logger      *slog.Logger
        lastContent string
}

func (p *legacyAttachPoller) run(ctx context.Context) {
        ticker := time.NewTicker(legacyPollInterval)
        defer ticker.Stop()

        for {
                select {
                case <-ctx.Done():
                        return
                case <-p.client.Done():
                        return
                case <-ticker.C:
                        p.poll(ctx)
                }
        }
}

func (p *legacyAttachPoller) poll(ctx context.Context) {
        outputPath := filepath.Join(p.serverDir, "output.txt")

        content, err := p.fileService.Download(ctx, p.node, outputPath)
        if err != nil {
                p.logger.Debug("legacy attach poll: download failed",
                        "path", outputPath,
                        "error", err,
                )

                return
        }

        currentContent := string(content)
        if currentContent == p.lastContent {
                return
        }

        var diff string
        if len(currentContent) > len(p.lastContent) && currentContent[:len(p.lastContent)] == p.lastContent {
                diff = currentContent[len(p.lastContent):]
        } else {
                diff = currentContent
        }

        p.lastContent = currentContent

        if diff != "" {
                p.client.SendMessage(ws.NewOutboundMessage(typeAttachOutput, attachOutputPayload{
                        Data: []byte(diff),
                }))
        }
}

1	package attach
2
3	import (
4	"context"
5	"encoding/json"
6	"log/slog"
7	"net/http"
8	"os"
9	"path/filepath"
10	"time"
11
12	"github.com/coder/websocket"
13	"github.com/gameap/gameap/internal/api/base"
14	serversbase "github.com/gameap/gameap/internal/api/servers/base"
15	"github.com/gameap/gameap/internal/daemon"
16	"github.com/gameap/gameap/internal/domain"
17	"github.com/gameap/gameap/internal/filters"
18	"github.com/gameap/gameap/internal/grpc/handlers"
19	"github.com/gameap/gameap/internal/grpc/session"
20	"github.com/gameap/gameap/internal/pubsub/channels"
21	"github.com/gameap/gameap/internal/repositories"
22	"github.com/gameap/gameap/internal/ws"
23	"github.com/gameap/gameap/pkg/api"
24	"github.com/gameap/gameap/pkg/auth"
25	"github.com/gameap/gameap/pkg/idgen"
26	"github.com/gameap/gameap/pkg/proto"
27	"github.com/pkg/errors"
28	)
29
30	const (
31	typeAttachInput = "attach.input"
32	typeAttachOutput = "attach.output"
33	typeAttachDetach = "attach.detach"
34	typeAttachStarted = "attach.started"
35
36	legacyPollInterval = 500 * time.Millisecond
37	)
38
39	type daemonCommands interface {
40	ExecuteCommand(
41	ctx context.Context,
42	node *domain.Node,
43	command string,
44	opts ...daemon.CommandServiceOption,
45	) (*daemon.CommandResult, error)
46	}
47
48	type fileService interface {
49	Download(ctx context.Context, node *domain.Node, filePath string) ([]byte, error)
50	Upload(
51	ctx context.Context, node *domain.Node, filePath string,
52	content []byte, perms os.FileMode, owner daemon.OwnerOptions,
53	) error
54	}
55
56	type Handler struct {
57	serverFinder *serversbase.ServerFinder
58	abilityChecker *serversbase.AbilityChecker
59	nodeRepo repositories.NodeRepository
60	hub *ws.Hub
61	originPatterns []string
62	registry *session.Registry
63	attachHandler *handlers.AttachHandler
64	daemonCommands daemonCommands
65	fileService fileService
66	responder base.Responder
67	logger *slog.Logger
68	}
69
70	func NewHandler(
71	serverRepo repositories.ServerRepository,
72	nodeRepo repositories.NodeRepository,
73	rbac base.RBAC,
74	hub *ws.Hub,
75	originPatterns []string,
76	registry *session.Registry,
77	attachHandler *handlers.AttachHandler,
78	daemonCommands daemonCommands,
79	fileService fileService,
80	responder base.Responder,
81	) *Handler {	24✔
82	return &Handler{	24✔
83	serverFinder: serversbase.NewServerFinder(serverRepo, rbac),	24✔
84	abilityChecker: serversbase.NewAbilityChecker(rbac),	24✔
85	nodeRepo: nodeRepo,	24✔
86	hub: hub,	24✔
87	originPatterns: originPatterns,	24✔
88	registry: registry,	24✔
89	attachHandler: attachHandler,	24✔
90	daemonCommands: daemonCommands,	24✔
91	fileService: fileService,	24✔
92	responder: responder,	24✔
93	logger: slog.Default(),	24✔
94	}	24✔
95	}	24✔
96
97	func (h Handler) ServeHTTP(rw http.ResponseWriter, r http.Request) {	19✔
98	ctx := r.Context()	19✔
99		19✔
100	sess := auth.SessionFromContext(ctx)	19✔
101	if !sess.IsAuthenticated() {	20✔
102	h.responder.WriteError(ctx, rw, api.NewError(http.StatusUnauthorized, "user not authenticated"))	1✔
103		1✔
104	return	1✔
105	}	1✔
106
107	input := api.NewInputReader(r)	18✔
108		18✔
109	serverID, err := input.ReadUint("server")	18✔
110	if err != nil {	19✔
111	h.responder.WriteError(ctx, rw, api.WrapHTTPError(	1✔
112	errors.WithMessage(err, "invalid server id"),	1✔
113	http.StatusBadRequest,	1✔
114	))	1✔
115		1✔
116	return	1✔
117	}	1✔
118
119	server, err := h.serverFinder.FindUserServer(ctx, sess.User, serverID)	17✔
120	if err != nil {	18✔
121	h.responder.WriteError(ctx, rw, err)	1✔
122		1✔
123	return	1✔
124	}	1✔
125
126	if err = h.abilityChecker.CheckOrError(	16✔
127	ctx,	16✔
128	sess.User.ID,	16✔
129	server.ID,	16✔
130	[]domain.AbilityName{domain.AbilityNameGameServerConsoleView},	16✔
131	); err != nil {	17✔
132	h.responder.WriteError(ctx, rw, err)	1✔
133		1✔
134	return	1✔
135	}	1✔
136
137	node, err := h.findNode(ctx, server.DSID)	15✔
138	if err != nil {	16✔
139	h.responder.WriteError(ctx, rw, err)	1✔
140		1✔
141	return	1✔
142	}	1✔
143
144	nodeID := uint64(server.DSID)	14✔
145		14✔
146	conn, err := ws.Accept(rw, r, &websocket.AcceptOptions{	14✔
147	OriginPatterns: h.originPatterns,	14✔
148	})	14✔
149	if err != nil {	14✔
150	h.logger.Warn("websocket accept failed", "error", err)	×
151		×
152	return	×
153	}	×
154
155	canSend := h.canSendCommands(ctx, sess.User, server)	14✔
156		14✔
157	if h.registry.IsConnectedAnywhere(nodeID) {	20✔
158	sessionID := idgen.New()	6✔
159	h.runAttachSession(ctx, conn, server, node, sessionID, sess.User, canSend)	6✔
160	} else {	14✔
161	h.runLegacyMode(ctx, conn, server, node, sess.User, canSend)	8✔
162	}	8✔
163	}
164
165	func (h *Handler) runAttachSession(
166	ctx context.Context,
167	conn *websocket.Conn,
168	server *domain.Server,
169	node *domain.Node,
170	sessionID string,
171	user *domain.User,
172	canSend bool,
173	) {	6✔
174	client := ws.NewClient(ctx, conn, h.hub, nil, h.logger)	6✔
175		6✔
176	startedTopic := ws.ChannelToTopic(channels.BuildRealtimeAttachStartedChannel(sessionID))	6✔
177	outputTopic := ws.ChannelToTopic(channels.BuildRealtimeAttachOutputChannel(sessionID))	6✔
178	closedTopic := ws.ChannelToTopic(channels.BuildRealtimeAttachClosedChannel(sessionID))	6✔
179		6✔
180	h.hub.Register(client, startedTopic, outputTopic, closedTopic)	6✔
181	h.attachHandler.TrackAttachSession(sessionID, uint64(server.ID))	6✔
182		6✔
183	nodeID := uint64(node.ID)	6✔
184		6✔
185	msgHandler := h.newMessageHandler(ctx, client, server, node, user, sessionID, canSend)	6✔
186	client.SetMessageHandler(msgHandler)	6✔
187		6✔
188	if err := h.registry.SendAttachRequest(ctx, nodeID, &proto.AttachRequest{	6✔
189	SessionId: sessionID,	6✔
190	ServerId: uint64(server.ID),	6✔
191	}); err != nil {	6✔
192	h.logger.Warn("failed to send attach request",	×
193	"server_id", server.ID,	×
194	"session_id", sessionID,	×
195	"error", err,	×
196	)	×
197	client.SendMessage(ws.NewErrorMessage("failed to attach to server console"))	×
198	}	×
199
200	client.Run()	6✔
201		6✔
202	_ = h.registry.SendAttachDetach(context.Background(), nodeID, &proto.AttachDetach{	6✔
203	SessionId: sessionID,	6✔
204	Reason: "client disconnected",	6✔
205	})	6✔
206	h.attachHandler.UntrackAttachSession(sessionID)	6✔
207	}
208
209	func (h *Handler) newMessageHandler(
210	ctx context.Context,
211	client *ws.Client,
212	server *domain.Server,
213	node *domain.Node,
214	user *domain.User,
215	sessionID string,
216	canSend bool,
217	) ws.MessageHandler {	6✔
218	nodeID := uint64(node.ID)	6✔
219		6✔
220	return func(_ context.Context, msg *ws.InboundMessage) {	9✔
221	switch msg.Type {	3✔
222	case typeAttachInput:	2✔
223	if !canSend {	3✔
224	client.SendMessage(ws.NewErrorMessage("permission denied: cannot send input"))	1✔
225		1✔
226	return	1✔
227	}	1✔
228
229	if err := h.abilityChecker.CheckOrError(	1✔
230	ctx,	1✔
231	user.ID,	1✔
232	server.ID,	1✔
233	[]domain.AbilityName{domain.AbilityNameGameServerConsoleSend},	1✔
234	); err != nil {	1✔
235	client.SendMessage(ws.NewErrorMessage("permission denied: cannot send input"))	×
236		×
237	return	×
238	}	×
239
240	var payload attachInputPayload	1✔
241	if err := json.Unmarshal(msg.Payload, &payload); err != nil {	1✔
242	return	×
243	}	×
244
245	if err := h.registry.SendAttachInput(ctx, nodeID, &proto.AttachInput{	1✔
246	SessionId: sessionID,	1✔
247	Data: payload.Data,	1✔
248	}); err != nil {	1✔
249	h.logger.Warn("failed to send attach input",	×
250	"session_id", sessionID,	×
251	"error", err,	×
252	)	×
253	}	×
254
255	case typeAttachDetach:	1✔
256	_ = h.registry.SendAttachDetach(ctx, nodeID, &proto.AttachDetach{	1✔
257	SessionId: sessionID,	1✔
258	Reason: "user detached",	1✔
259	})	1✔
260	client.Close()	1✔
261	}
262	}
263	}
264
265	func (h Handler) findNode(ctx context.Context, nodeID uint) (domain.Node, error) {	15✔
266	nodes, err := h.nodeRepo.Find(ctx, &filters.FindNode{	15✔
267	IDs: []uint{nodeID},	15✔
268	}, nil, &filters.Pagination{	15✔
269	Limit: 1,	15✔
270	})	15✔
271	if err != nil {	15✔
272	return nil, errors.WithMessage(err, "failed to find node")	×
273	}	×
274
275	if len(nodes) == 0 {	16✔
276	return nil, api.NewNotFoundError("node not found")	1✔
277	}	1✔
278
279	return &nodes[0], nil	14✔
280	}
281
282	func (h Handler) canSendCommands(ctx context.Context, user domain.User, server *domain.Server) bool {	14✔
283	err := h.abilityChecker.CheckOrError(	14✔
284	ctx,	14✔
285	user.ID,	14✔
286	server.ID,	14✔
287	[]domain.AbilityName{domain.AbilityNameGameServerConsoleSend},	14✔
288	)	14✔
289		14✔
290	return err == nil	14✔
291	}	14✔
292
293	type attachInputPayload struct {
294	Data []byte `json:"data"`
295	}
296
297	type attachStartedPayload struct {
298	SessionID string `json:"session_id"`
299	ServerID uint64 `json:"server_id"`
300	}
301
302	type attachOutputPayload struct {
303	Data []byte `json:"data"`
304	}
305
306	// Legacy mode for daemons not connected via gRPC.
307
308	func (h *Handler) runLegacyMode(
309	ctx context.Context,
310	conn *websocket.Conn,
311	server *domain.Server,
312	node *domain.Node,
313	user *domain.User,
314	canSend bool,
315	) {	8✔
316	client := ws.NewClient(ctx, conn, h.hub, nil, h.logger)	8✔
317		8✔
318	msgHandler := h.newLegacyMessageHandler(ctx, client, server, node, user, canSend)	8✔
319	client.SetMessageHandler(msgHandler)	8✔
320		8✔
321	client.SendMessage(ws.NewOutboundMessage(typeAttachStarted, attachStartedPayload{	8✔
322	SessionID: idgen.New(),	8✔
323	ServerID: uint64(server.ID),	8✔
324	}))	8✔
325		8✔
326	lastContent := h.sendLegacyHistory(ctx, client, server, node)	8✔
327		8✔
328	poller := &legacyAttachPoller{	8✔
329	client: client,	8✔
330	fileService: h.fileService,	8✔
331	node: node,	8✔
332	serverDir: server.Dir,	8✔
333	logger: h.logger,	8✔
334	lastContent: lastContent,	8✔
335	}	8✔
336	go poller.run(ctx)	8✔
337		8✔
338	client.Run()	8✔
339	}	8✔
340
341	func (h *Handler) newLegacyMessageHandler(
342	ctx context.Context,
343	client *ws.Client,
344	server *domain.Server,
345	node *domain.Node,
346	user *domain.User,
347	canSend bool,
348	) ws.MessageHandler {	8✔
349	return func(_ context.Context, msg *ws.InboundMessage) {	12✔
350	switch msg.Type {	4✔
351	case typeAttachInput:	4✔
352	if !canSend {	5✔
353	client.SendMessage(ws.NewErrorMessage("permission denied: cannot send input"))	1✔
354		1✔
355	return	1✔
356	}	1✔
357
358	if err := h.abilityChecker.CheckOrError(	3✔
359	ctx, user.ID, server.ID,	3✔
360	[]domain.AbilityName{domain.AbilityNameGameServerConsoleSend},	3✔
361	); err != nil {	3✔
362	client.SendMessage(ws.NewErrorMessage("permission denied: cannot send input"))	×
363		×
364	return	×
365	}	×
366
367	var payload attachInputPayload	3✔
368	if err := json.Unmarshal(msg.Payload, &payload); err != nil {	3✔
369	return	×
370	}	×
371
372	h.sendLegacyInput(ctx, client, server, node, string(payload.Data))	3✔
373
374	case typeAttachDetach:	×
375	client.Close()	×
376	}
377	}
378	}
379
380	func (h *Handler) sendLegacyInput(
381	ctx context.Context, client ws.Client, server domain.Server, node *domain.Node, input string,
382	) {	3✔
383	if node.ScriptSendCommand != nil && *node.ScriptSendCommand != "" {	4✔
384	cmd := server.ReplaceServerShortcodes(node, *node.ScriptSendCommand, map[string]string{	1✔
385	"command": input,	1✔
386	})	1✔
387		1✔
388	if _, err := h.daemonCommands.ExecuteCommand(ctx, node, cmd); err != nil {	1✔
389	h.logger.Warn("failed to execute send command script",	×
390	"server_id", server.ID, "error", err)	×
391	client.SendMessage(ws.NewErrorMessage("failed to send input"))	×
392	}	×
393
394	return	1✔
395	}
396
397	inputPath := filepath.Join(server.Dir, "input.txt")	2✔
398		2✔
399	err := h.fileService.Upload(	2✔
400	ctx, node, inputPath, []byte(input), 0o644, daemon.OwnerFromServer(server),	2✔
401	)	2✔
402	if err != nil {	3✔
403	h.logger.Warn("failed to upload input", "server_id", server.ID, "error", err)	1✔
404	client.SendMessage(ws.NewErrorMessage("failed to send input"))	1✔
405	}	1✔
406	}
407
408	func (h *Handler) sendLegacyHistory(
409	ctx context.Context, client ws.Client, server domain.Server, node *domain.Node,
410	) string {	8✔
411	if node.ScriptGetConsole != nil && *node.ScriptGetConsole != "" {	9✔
412	cmd := server.ReplaceServerShortcodes(node, *node.ScriptGetConsole, nil)	1✔
413		1✔
414	result, err := h.daemonCommands.ExecuteCommand(ctx, node, cmd)	1✔
415	if err == nil && result.Output != "" {	2✔
416	client.SendMessage(ws.NewOutboundMessage(typeAttachOutput, attachOutputPayload{	1✔
417	Data: []byte(result.Output),	1✔
418	}))	1✔
419		1✔
420	return result.Output	1✔
421	}	1✔
422	}
423
424	outputPath := filepath.Join(server.Dir, "output.txt")	7✔
425		7✔
426	content, err := h.fileService.Download(ctx, node, outputPath)	7✔
427	if err != nil {	7✔
428	return ""	×
429	}	×
430
431	const maxBytes = 65536	7✔
432	if len(content) > maxBytes {	8✔
433	content = content[len(content)-maxBytes:]	1✔
434	}	1✔
435
436	if len(content) > 0 {	10✔
437	client.SendMessage(ws.NewOutboundMessage(typeAttachOutput, attachOutputPayload{	3✔
438	Data: content,	3✔
439	}))	3✔
440	}	3✔
441
442	return string(content)	7✔
443	}
444
445	type legacyAttachPoller struct {
446	client *ws.Client
447	fileService fileService
448	node *domain.Node
449	serverDir string
450	logger *slog.Logger
451	lastContent string
452	}
453
454	func (p *legacyAttachPoller) run(ctx context.Context) {	8✔
455	ticker := time.NewTicker(legacyPollInterval)	8✔
456	defer ticker.Stop()	8✔
457		8✔
458	for {	18✔
459	select {	10✔
460	case <-ctx.Done():	×
461	return	×
462	case <-p.client.Done():	8✔
463	return	8✔
464	case <-ticker.C:	2✔
465	p.poll(ctx)	2✔
466	}
467	}
468	}
469
470	func (p *legacyAttachPoller) poll(ctx context.Context) {	2✔
471	outputPath := filepath.Join(p.serverDir, "output.txt")	2✔
472		2✔
473	content, err := p.fileService.Download(ctx, p.node, outputPath)	2✔
474	if err != nil {	2✔
475	p.logger.Debug("legacy attach poll: download failed",	×
476	"path", outputPath,	×
477	"error", err,	×
478	)	×
479		×
480	return	×
481	}	×
482
483	currentContent := string(content)	2✔
484	if currentContent == p.lastContent {	3✔
485	return	1✔
486	}	1✔
487
488	var diff string	1✔
489	if len(currentContent) > len(p.lastContent) && currentContent[:len(p.lastContent)] == p.lastContent {	2✔
490	diff = currentContent[len(p.lastContent):]	1✔
491	} else {	1✔
492	diff = currentContent	×
493	}	×
494
495	p.lastContent = currentContent	1✔
496		1✔
497	if diff != "" {	2✔
498	p.client.SendMessage(ws.NewOutboundMessage(typeAttachOutput, attachOutputPayload{	1✔
499	Data: []byte(diff),	1✔
500	}))	1✔
501	}	1✔
502	}

gameap / gameap / 25960901307

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous