From ab4f6830bc4d65be886ac0c5b4d2fe2cc52fc052 Mon Sep 17 00:00:00 2001 From: Fabricio Voznika Date: Mon, 13 Jun 2022 10:21:34 -0700 Subject: [PATCH] Add tool to save remote trace sessions This tool allows testing of trace servers without the need to setup runsc and workloads during testing. More details in the readme file. Updates #4805 PiperOrigin-RevId: 454641010 --- .../seccheck/checkers/remote/remote_test.go | 6 +- .../seccheck/checkers/remote/server/BUILD | 19 ++ .../seccheck/checkers/remote/server/server.go | 245 ++++++++++++++++++ .../seccheck/checkers/remote/test/BUILD | 6 +- .../seccheck/checkers/remote/test/server.go | 211 +++------------ runsc/container/trace_test.go | 8 +- test/trace/trace_test.go | 2 +- tools/tracereplay/BUILD | 36 +++ tools/tracereplay/README.md | 77 ++++++ tools/tracereplay/main/BUILD | 15 ++ tools/tracereplay/main/main.go | 152 +++++++++++ tools/tracereplay/replay.go | 133 ++++++++++ tools/tracereplay/save.go | 111 ++++++++ tools/tracereplay/testdata/client-0001 | Bin 0 -> 804 bytes tools/tracereplay/tracereplay.go | 83 ++++++ tools/tracereplay/tracereplay_test.go | 76 ++++++ 16 files changed, 997 insertions(+), 183 deletions(-) create mode 100644 pkg/sentry/seccheck/checkers/remote/server/BUILD create mode 100644 pkg/sentry/seccheck/checkers/remote/server/server.go create mode 100644 tools/tracereplay/BUILD create mode 100644 tools/tracereplay/README.md create mode 100644 tools/tracereplay/main/BUILD create mode 100644 tools/tracereplay/main/main.go create mode 100644 tools/tracereplay/replay.go create mode 100644 tools/tracereplay/save.go create mode 100644 tools/tracereplay/testdata/client-0001 create mode 100644 tools/tracereplay/tracereplay.go create mode 100644 tools/tracereplay/tracereplay_test.go diff --git a/pkg/sentry/seccheck/checkers/remote/remote_test.go b/pkg/sentry/seccheck/checkers/remote/remote_test.go index 61ae35605..57aa2b80f 100644 --- a/pkg/sentry/seccheck/checkers/remote/remote_test.go +++ b/pkg/sentry/seccheck/checkers/remote/remote_test.go @@ -115,7 +115,7 @@ func TestBasic(t *testing.T) { } defer server.Close() - endpoint, err := setup(server.Path) + endpoint, err := setup(server.Endpoint) if err != nil { t.Fatalf("setup(): %v", err) } @@ -163,7 +163,7 @@ func TestVersionUnsupported(t *testing.T) { server.SetVersion(0) - _, err = setup(server.Path) + _, err = setup(server.Endpoint) if err == nil || !strings.Contains(err.Error(), "remote version") { t.Fatalf("Wrong error: %v", err) } @@ -178,7 +178,7 @@ func TestVersionNewer(t *testing.T) { server.SetVersion(wire.CurrentVersion + 10) - endpoint, err := setup(server.Path) + endpoint, err := setup(server.Endpoint) if err != nil { t.Fatalf("setup(): %v", err) } diff --git a/pkg/sentry/seccheck/checkers/remote/server/BUILD b/pkg/sentry/seccheck/checkers/remote/server/BUILD new file mode 100644 index 000000000..cefced63d --- /dev/null +++ b/pkg/sentry/seccheck/checkers/remote/server/BUILD @@ -0,0 +1,19 @@ +load("//tools:defs.bzl", "go_library") + +package(licenses = ["notice"]) + +go_library( + name = "server", + srcs = ["server.go"], + visibility = ["//:sandbox"], + deps = [ + "//pkg/cleanup", + "//pkg/log", + "//pkg/sentry/seccheck/checkers/remote/wire", + "//pkg/sentry/seccheck/points:points_go_proto", + "//pkg/sync", + "//pkg/unet", + "@org_golang_google_protobuf//proto:go_default_library", + "@org_golang_x_sys//unix:go_default_library", + ], +) diff --git a/pkg/sentry/seccheck/checkers/remote/server/server.go b/pkg/sentry/seccheck/checkers/remote/server/server.go new file mode 100644 index 000000000..cbfbd59e1 --- /dev/null +++ b/pkg/sentry/seccheck/checkers/remote/server/server.go @@ -0,0 +1,245 @@ +// Copyright 2022 The gVisor Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package server provides a common server implementation that can connect with +// remote.Remote. +package server + +import ( + "errors" + "fmt" + "io" + "os" + + "golang.org/x/sys/unix" + "google.golang.org/protobuf/proto" + "gvisor.dev/gvisor/pkg/cleanup" + "gvisor.dev/gvisor/pkg/log" + "gvisor.dev/gvisor/pkg/sentry/seccheck/checkers/remote/wire" + pb "gvisor.dev/gvisor/pkg/sentry/seccheck/points/points_go_proto" + "gvisor.dev/gvisor/pkg/sync" + "gvisor.dev/gvisor/pkg/unet" +) + +// ClientHandler is used to interface with client that connect to the server. +type ClientHandler interface { + // NewClient is called when a new client connects to the server. It returns + // a handler that will be bound to the client. + NewClient() (MessageHandler, error) +} + +// MessageHandler is used to process messages from a client. +type MessageHandler interface { + // Message processes a single message. raw contains the entire unparsed + // message. hdr is the parser message header and payload is the unparsed + // message data. + Message(raw []byte, hdr wire.Header, payload []byte) error + + // Version returns what wire version of the protocol is supported. + Version() uint32 + + // Close closes the handler. + Close() +} + +type client struct { + socket *unet.Socket + handler MessageHandler +} + +func (c client) close() { + _ = c.socket.Close() + c.handler.Close() +} + +// CommonServer provides common functionality to connect and process messages +// from different clients. Implementors decide how clients and messages are +// handled, e.g. counting messages for testing. +type CommonServer struct { + // Endpoint is the path to the socket that the server listens to. + Endpoint string + + socket *unet.ServerSocket + + handler ClientHandler + + cond sync.Cond + + // +checklocks:cond.L + clients []client +} + +// Init initializes the server. It must be called before it is used. +func (s *CommonServer) Init(path string, handler ClientHandler) { + s.Endpoint = path + s.handler = handler + s.cond = sync.Cond{L: &sync.Mutex{}} +} + +// Start creates the socket file and listens for new connections. +func (s *CommonServer) Start() error { + socket, err := unix.Socket(unix.AF_UNIX, unix.SOCK_SEQPACKET, 0) + if err != nil { + return fmt.Errorf("socket(AF_UNIX, SOCK_SEQPACKET, 0): %w", err) + } + cu := cleanup.Make(func() { + _ = unix.Close(socket) + }) + defer cu.Clean() + + sa := &unix.SockaddrUnix{Name: s.Endpoint} + if err := unix.Bind(socket, sa); err != nil { + return fmt.Errorf("bind(%q): %w", s.Endpoint, err) + } + + s.socket, err = unet.NewServerSocket(socket) + if err != nil { + return err + } + cu.Add(func() { s.socket.Close() }) + + if err := s.socket.Listen(); err != nil { + return err + } + + go s.run() + cu.Release() + return nil +} + +func (s *CommonServer) run() { + for { + socket, err := s.socket.Accept() + if err != nil { + // EBADF returns when the socket closes. + if !errors.Is(err, unix.EBADF) { + log.Warningf("socket.Accept(): %v", err) + } + return + } + msgHandler, err := s.handler.NewClient() + if err != nil { + log.Warningf("handler.NewClient: %v", err) + return + } + client := client{ + socket: socket, + handler: msgHandler, + } + s.cond.L.Lock() + s.clients = append(s.clients, client) + s.cond.Broadcast() + s.cond.L.Unlock() + + if err := s.handshake(client); err != nil { + log.Warningf(err.Error()) + s.closeClient(client) + continue + } + go s.handleClient(client) + } +} + +// handshake performs version exchange with client. See common.proto for details +// about the protocol. +func (s *CommonServer) handshake(client client) error { + var in [1024]byte + read, err := client.socket.Read(in[:]) + if err != nil { + return fmt.Errorf("reading handshake message: %w", err) + } + hsIn := pb.Handshake{} + if err := proto.Unmarshal(in[:read], &hsIn); err != nil { + return fmt.Errorf("unmarshalling handshake message: %w", err) + } + if hsIn.Version != wire.CurrentVersion { + return fmt.Errorf("wrong version number, want: %d, got, %d", wire.CurrentVersion, hsIn.Version) + } + + hsOut := pb.Handshake{Version: client.handler.Version()} + out, err := proto.Marshal(&hsOut) + if err != nil { + return fmt.Errorf("marshalling handshake message: %w", err) + } + if _, err := client.socket.Write(out); err != nil { + return fmt.Errorf("sending handshake message: %w", err) + } + return nil +} + +func (s *CommonServer) handleClient(client client) { + defer s.closeClient(client) + + var buf = make([]byte, 1024*1024) + for { + read, err := client.socket.Read(buf) + if err != nil { + if errors.Is(err, io.EOF) || errors.Is(err, unix.EBADF) { + // Both errors indicate that the socket has been closed. + return + } + panic(err) + } + if read < wire.HeaderStructSize { + panic("message too small") + } + hdr := wire.Header{} + hdr.UnmarshalUnsafe(buf[0:wire.HeaderStructSize]) + if read < int(hdr.HeaderSize) { + panic(fmt.Sprintf("message truncated, header size: %d, read: %d", hdr.HeaderSize, read)) + } + if err := client.handler.Message(buf[:read], hdr, buf[hdr.HeaderSize:read]); err != nil { + panic(err) + } + } +} + +func (s *CommonServer) closeClient(client client) { + client.close() + + // Stop tracking this client. + s.cond.L.Lock() + for i, c := range s.clients { + if c == client { + s.clients = append(s.clients[:i], s.clients[i+1:]...) + break + } + } + s.cond.Broadcast() + s.cond.L.Unlock() +} + +// Close stops listening and closes all connections. +func (s *CommonServer) Close() { + if s.socket != nil { + _ = s.socket.Close() + } + s.cond.L.Lock() + for _, client := range s.clients { + client.close() + } + s.clients = nil + s.cond.Broadcast() + s.cond.L.Unlock() + _ = os.Remove(s.Endpoint) +} + +// WaitForNoClients waits until the number of clients connected reaches 0. +func (s *CommonServer) WaitForNoClients() { + s.cond.L.Lock() + defer s.cond.L.Unlock() + for len(s.clients) > 0 { + s.cond.Wait() + } +} diff --git a/pkg/sentry/seccheck/checkers/remote/test/BUILD b/pkg/sentry/seccheck/checkers/remote/test/BUILD index 2e3fb6666..1d02d47b5 100644 --- a/pkg/sentry/seccheck/checkers/remote/test/BUILD +++ b/pkg/sentry/seccheck/checkers/remote/test/BUILD @@ -8,13 +8,9 @@ go_library( srcs = ["server.go"], visibility = ["//:sandbox"], deps = [ - "//pkg/cleanup", - "//pkg/log", + "//pkg/sentry/seccheck/checkers/remote/server", "//pkg/sentry/seccheck/checkers/remote/wire", "//pkg/sentry/seccheck/points:points_go_proto", "//pkg/sync", - "//pkg/unet", - "@org_golang_google_protobuf//proto:go_default_library", - "@org_golang_x_sys//unix:go_default_library", ], ) diff --git a/pkg/sentry/seccheck/checkers/remote/test/server.go b/pkg/sentry/seccheck/checkers/remote/test/server.go index bbe687c3e..e73443883 100644 --- a/pkg/sentry/seccheck/checkers/remote/test/server.go +++ b/pkg/sentry/seccheck/checkers/remote/test/server.go @@ -16,33 +16,23 @@ package test import ( - "errors" - "fmt" "io/ioutil" "os" "path/filepath" - "golang.org/x/sys/unix" - "google.golang.org/protobuf/proto" - "gvisor.dev/gvisor/pkg/cleanup" - "gvisor.dev/gvisor/pkg/log" + "gvisor.dev/gvisor/pkg/sentry/seccheck/checkers/remote/server" "gvisor.dev/gvisor/pkg/sentry/seccheck/checkers/remote/wire" pb "gvisor.dev/gvisor/pkg/sentry/seccheck/points/points_go_proto" "gvisor.dev/gvisor/pkg/sync" - "gvisor.dev/gvisor/pkg/unet" ) // Server is the counterpart to the checkers.Remote. It receives connections // remote checkers and stores all points that it receives. type Server struct { - Path string - socket *unet.ServerSocket + server.CommonServer cond sync.Cond - // +checklocks:cond.L - clients []*unet.Socket - // +checklocks:cond.L points []Message @@ -67,147 +57,21 @@ func NewServer() (*Server, error) { if err != nil { return nil, err } - server, err := newServerPath(filepath.Join(dir, "remote.sock")) - if err != nil { - _ = os.RemoveAll(dir) - return nil, err - } - return server, nil -} - -func newServerPath(path string) (*Server, error) { - socket, err := unix.Socket(unix.AF_UNIX, unix.SOCK_SEQPACKET, 0) - if err != nil { - return nil, fmt.Errorf("socket(AF_UNIX, SOCK_SEQPACKET, 0): %w", err) - } - cu := cleanup.Make(func() { - _ = unix.Close(socket) - }) - defer cu.Clean() - - sa := &unix.SockaddrUnix{Name: path} - if err := unix.Bind(socket, sa); err != nil { - return nil, fmt.Errorf("bind(%q): %w", path, err) - } - - ss, err := unet.NewServerSocket(socket) - if err != nil { - return nil, err - } - cu.Add(func() { ss.Close() }) - - if err := ss.Listen(); err != nil { - return nil, err - } - - server := &Server{ - Path: path, - socket: ss, + s := &Server{ version: wire.CurrentVersion, cond: sync.Cond{L: &sync.Mutex{}}, } - go server.run() - cu.Release() - return server, nil + s.CommonServer.Init(filepath.Join(dir, "remote.sock"), s) + if err := s.CommonServer.Start(); err != nil { + _ = os.RemoveAll(dir) + return nil, err + } + return s, nil } -func (s *Server) run() { - for { - client, err := s.socket.Accept() - if err != nil { - // EBADF returns when the socket closes. - if !errors.Is(err, unix.EBADF) { - log.Warningf("socket.Accept(): %v", err) - } - return - } - if err := s.handshake(client); err != nil { - log.Warningf(err.Error()) - _ = client.Close() - continue - } - s.cond.L.Lock() - s.clients = append(s.clients, client) - s.cond.Broadcast() - s.cond.L.Unlock() - go s.handleClient(client) - } -} - -// handshake performs version exchange with client. See common.proto for details -// about the protocol. -func (s *Server) handshake(client *unet.Socket) error { - var in [1024]byte - read, err := client.Read(in[:]) - if err != nil { - return fmt.Errorf("reading handshake message: %w", err) - } - hsIn := pb.Handshake{} - if err := proto.Unmarshal(in[:read], &hsIn); err != nil { - return fmt.Errorf("unmarshalling handshake message: %w", err) - } - if hsIn.Version != wire.CurrentVersion { - return fmt.Errorf("wrong version number, want: %d, got, %d", wire.CurrentVersion, hsIn.Version) - } - - s.mu.Lock() - v := s.version - s.mu.Unlock() - hsOut := pb.Handshake{Version: v} - out, err := proto.Marshal(&hsOut) - if err != nil { - return fmt.Errorf("marshalling handshake message: %w", err) - } - if _, err := client.Write(out); err != nil { - return fmt.Errorf("sending handshake message: %w", err) - } - return nil -} - -func (s *Server) handleClient(client *unet.Socket) { - defer func() { - s.cond.L.Lock() - for i, c := range s.clients { - if c == client { - s.clients = append(s.clients[:i], s.clients[i+1:]...) - break - } - } - s.cond.Broadcast() - s.cond.L.Unlock() - _ = client.Close() - }() - - var buf = make([]byte, 1024*1024) - for { - read, err := client.Read(buf) - if err != nil { - return - } - if read == 0 { - return - } - if read < wire.HeaderStructSize { - panic("invalid message") - } - hdr := wire.Header{} - hdr.UnmarshalUnsafe(buf[0:wire.HeaderStructSize]) - if read < int(hdr.HeaderSize) { - panic(fmt.Sprintf("message truncated, header size: %d, readL %d", hdr.HeaderSize, read)) - } - - msgSize := read - int(hdr.HeaderSize) - msg := Message{ - MsgType: pb.MessageType(hdr.MessageType), - Msg: make([]byte, msgSize), - } - copy(msg.Msg, buf[hdr.HeaderSize:read]) - - s.cond.L.Lock() - s.points = append(s.points, msg) - s.cond.Broadcast() - s.cond.L.Unlock() - } +// NewClient returns a new MessageHandler to process messages. +func (s *Server) NewClient() (server.MessageHandler, error) { + return &msgHandler{owner: s}, nil } // Count return the number of points it has received. @@ -236,19 +100,6 @@ func (s *Server) GetPoints() []Message { return cpy } -// Close stops listenning and closes all connections. -func (s *Server) Close() { - _ = s.socket.Close() - s.cond.L.Lock() - for _, client := range s.clients { - _ = client.Close() - } - s.clients = nil - s.cond.Broadcast() - s.cond.L.Unlock() - _ = os.Remove(s.Path) -} - // WaitForCount waits for the number of points to reach the desired number. func (s *Server) WaitForCount(count int) { s.cond.L.Lock() @@ -259,18 +110,38 @@ func (s *Server) WaitForCount(count int) { return } -// WaitForNoClients waits until the number of clients connected reaches 0. -func (s *Server) WaitForNoClients() { - s.cond.L.Lock() - defer s.cond.L.Unlock() - for len(s.clients) > 0 { - s.cond.Wait() - } -} - // SetVersion sets the version to be used in handshake. func (s *Server) SetVersion(newVersion uint32) { s.mu.Lock() defer s.mu.Unlock() s.version = newVersion } + +type msgHandler struct { + owner *Server +} + +// Message stores the message type and payload. +func (m *msgHandler) Message(_ []byte, hdr wire.Header, payload []byte) error { + msg := Message{ + MsgType: pb.MessageType(hdr.MessageType), + Msg: make([]byte, len(payload)), + } + copy(msg.Msg, payload) + + m.owner.cond.L.Lock() + defer m.owner.cond.L.Unlock() + m.owner.points = append(m.owner.points, msg) + m.owner.cond.Broadcast() + return nil +} + +// Version returns the wire version supported or overriden by SetVersion. +func (m *msgHandler) Version() uint32 { + m.owner.mu.Lock() + defer m.owner.mu.Unlock() + return m.owner.version +} + +// Close implements server.MessageHandler. +func (m *msgHandler) Close() {} diff --git a/runsc/container/trace_test.go b/runsc/container/trace_test.go index 9a98af330..a2df2d5a5 100644 --- a/runsc/container/trace_test.go +++ b/runsc/container/trace_test.go @@ -69,7 +69,7 @@ func TestTraceStartup(t *testing.T) { ContextFields: []string{"container_id"}, }, }, - Sinks: []seccheck.SinkConfig{remoteSinkConfig(server.Path)}, + Sinks: []seccheck.SinkConfig{remoteSinkConfig(server.Endpoint)}, }, } encoder := json.NewEncoder(podInitConfig) @@ -148,7 +148,7 @@ func TestTraceLifecycle(t *testing.T) { ContextFields: []string{"container_id"}, }, }, - Sinks: []seccheck.SinkConfig{remoteSinkConfig(server.Path)}, + Sinks: []seccheck.SinkConfig{remoteSinkConfig(server.Endpoint)}, } if err := cont.Sandbox.CreateTraceSession(&session, false); err != nil { t.Fatalf("CreateTraceSession(): %v", err) @@ -248,7 +248,7 @@ func TestTraceForceCreate(t *testing.T) { Points: []seccheck.PointConfig{ {Name: "sentry/exit_notify_parent"}, }, - Sinks: []seccheck.SinkConfig{remoteSinkConfig(server.Path)}, + Sinks: []seccheck.SinkConfig{remoteSinkConfig(server.Endpoint)}, } if err := cont.Sandbox.CreateTraceSession(&session, false); err != nil { t.Fatalf("CreateTraceSession(): %v", err) @@ -277,7 +277,7 @@ func TestTraceForceCreate(t *testing.T) { Points: []seccheck.PointConfig{ {Name: "sentry/task_exit"}, }, - Sinks: []seccheck.SinkConfig{remoteSinkConfig(server.Path)}, + Sinks: []seccheck.SinkConfig{remoteSinkConfig(server.Endpoint)}, } if err := cont.Sandbox.CreateTraceSession(&session, true); err != nil { t.Fatalf("CreateTraceSession(force): %v", err) diff --git a/test/trace/trace_test.go b/test/trace/trace_test.go index cd5de6ef8..76f370270 100644 --- a/test/trace/trace_test.go +++ b/test/trace/trace_test.go @@ -48,7 +48,7 @@ func TestAll(t *testing.T) { if err != nil { t.Fatal(err) } - cfg, err := buildPodConfig(runsc, server.Path) + cfg, err := buildPodConfig(runsc, server.Endpoint) if err != nil { t.Fatal(err) } diff --git a/tools/tracereplay/BUILD b/tools/tracereplay/BUILD new file mode 100644 index 000000000..54d131568 --- /dev/null +++ b/tools/tracereplay/BUILD @@ -0,0 +1,36 @@ +load("//tools:defs.bzl", "go_library", "go_test") + +package(licenses = ["notice"]) + +go_library( + name = "tracereplay", + srcs = [ + "replay.go", + "save.go", + "tracereplay.go", + ], + visibility = [ + "//tools/tracereplay:__subpackages__", + ], + deps = [ + "//pkg/atomicbitops", + "//pkg/log", + "//pkg/sentry/seccheck/checkers/remote/server", + "//pkg/sentry/seccheck/checkers/remote/wire", + "//pkg/sentry/seccheck/points:points_go_proto", + "@org_golang_google_protobuf//proto:go_default_library", + "@org_golang_x_sys//unix:go_default_library", + ], +) + +go_test( + name = "tracereplay_test", + srcs = ["tracereplay_test.go"], + data = [ + "testdata/client-0001", + ], + library = ":tracereplay", + deps = [ + "//pkg/test/testutil", + ], +) diff --git a/tools/tracereplay/README.md b/tools/tracereplay/README.md new file mode 100644 index 000000000..964f89666 --- /dev/null +++ b/tools/tracereplay/README.md @@ -0,0 +1,77 @@ +# What is it? + +The `tracereplay` tool can save `runsc trace` sessions to a file, and later +replay the same sequence of messages. This can be used to run tests that rely on +the messages without the need to setup runsc, configure trace sessions, and run +specific workloads. + +# How to use it? + +The `tracereplay save` command starts a server that listens to new connections +from runsc and creates a trace file for each runsc instance that connects to it. +The command below starts a server on listening on `/tmp/gvisor_events.sock` and +writes trace files to `/tmp/trace` directory: + +```shell +$ tracereplay save --endpoint=/tmp/gvisor_events.sock --out=/tmp/trace +``` + +When you execute runsc configured with a trace session using the `remote` sink +connecting to `/tmp/gvisor_events.sock`, all messages will be saved to a file +under `/tmp/trace`. For example, if you run the following commands, runsc will +connect to the server above and all trace points triggered by the workload will +be stored in the save file: + +```shell +$ cat > /tmp/pod_init.json < id: "runsc-865139" cwd: "/home/fvoznika" args: "/bin/true" +Connection closed +``` diff --git a/tools/tracereplay/main/BUILD b/tools/tracereplay/main/BUILD new file mode 100644 index 000000000..1e114b7da --- /dev/null +++ b/tools/tracereplay/main/BUILD @@ -0,0 +1,15 @@ +load("//tools:defs.bzl", "go_binary") + +package(licenses = ["notice"]) + +go_binary( + name = "tracereplay", + srcs = [ + "main.go", + ], + deps = [ + "//runsc/flag", + "//tools/tracereplay", + "@com_github_google_subcommands//:go_default_library", + ], +) diff --git a/tools/tracereplay/main/main.go b/tools/tracereplay/main/main.go new file mode 100644 index 000000000..979aeefc6 --- /dev/null +++ b/tools/tracereplay/main/main.go @@ -0,0 +1,152 @@ +// Copyright 2022 The gVisor Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package main implements a tool that can save and replay messages from +// issued from remote.Remote. +package main + +import ( + "context" + "fmt" + "os" + "os/signal" + + "github.com/google/subcommands" + "gvisor.dev/gvisor/runsc/flag" + "gvisor.dev/gvisor/tools/tracereplay" +) + +func main() { + subcommands.Register(subcommands.HelpCommand(), "") + subcommands.Register(subcommands.FlagsCommand(), "") + subcommands.Register(&saveCmd{}, "") + subcommands.Register(&replayCmd{}, "") + flag.CommandLine.Parse(os.Args[1:]) + os.Exit(int(subcommands.Execute(context.Background()))) +} + +// saveCmd implements subcommands.Command for the "save" command. +type saveCmd struct { + endpoint string + out string + prefix string +} + +// Name implements subcommands.Command. +func (*saveCmd) Name() string { + return "save" +} + +// Synopsis implements subcommands.Command. +func (*saveCmd) Synopsis() string { + return "save trace sessions to files" +} + +// Usage implements subcommands.Command. +func (*saveCmd) Usage() string { + return `save [flags] - save trace sessions to files +` +} + +// SetFlags implements subcommands.Command. +func (c *saveCmd) SetFlags(f *flag.FlagSet) { + f.StringVar(&c.endpoint, "endpoint", "", "path to trace server endpoint to connect") + f.StringVar(&c.out, "out", "./replay", "path to a directory where trace files will be saved") + f.StringVar(&c.prefix, "prefix", "client-", "name to be prefixed to each trace file") +} + +// Execute implements subcommands.Command. +func (c *saveCmd) Execute(_ context.Context, f *flag.FlagSet, args ...interface{}) subcommands.ExitStatus { + if f.NArg() > 0 { + fmt.Fprintf(os.Stderr, "unexpected argument: %s\n", f.Args()) + return subcommands.ExitUsageError + } + if len(c.endpoint) == 0 { + fmt.Fprintf(os.Stderr, "--endpoint is required\n") + return subcommands.ExitUsageError + } + _ = os.Remove(c.endpoint) + + server := tracereplay.NewSave(c.endpoint, c.out, c.prefix) + defer server.Close() + + if err := server.Start(); err != nil { + fmt.Fprintf(os.Stderr, "starting server: %v\n", err) + return subcommands.ExitFailure + } + + ch := make(chan os.Signal) + signal.Notify(ch, os.Interrupt) + + done := make(chan struct{}) + go func() { + <-ch + fmt.Printf("Ctrl-C pressed, stopping.\n") + done <- struct{}{} + }() + + fmt.Printf("Listening on %q. Press ctrl-C to stop...\n", c.endpoint) + <-done + return subcommands.ExitSuccess +} + +// replayCmd implements subcommands.Command for the "replay" command. +type replayCmd struct { + endpoint string + in string +} + +// Name implements subcommands.Command. +func (*replayCmd) Name() string { + return "replay" +} + +// Synopsis implements subcommands.Command. +func (*replayCmd) Synopsis() string { + return "replay a trace session from a file" +} + +// Usage implements subcommands.Command. +func (*replayCmd) Usage() string { + return `replay [flags] - replay a trace session from a file +` +} + +// SetFlags implements subcommands.Command. +func (c *replayCmd) SetFlags(f *flag.FlagSet) { + f.StringVar(&c.endpoint, "endpoint", "", "path to trace server endpoint to connect") + f.StringVar(&c.in, "in", "", "path to trace file containing messages to be replayed") +} + +// Execute implements subcommands.Command. +func (c *replayCmd) Execute(_ context.Context, f *flag.FlagSet, _ ...interface{}) subcommands.ExitStatus { + if f.NArg() > 0 { + fmt.Fprintf(os.Stderr, "unexpected argument: %s\n", f.Args()) + return subcommands.ExitUsageError + } + if len(c.in) == 0 { + fmt.Fprintf(os.Stderr, "--in is required\n") + return subcommands.ExitUsageError + } + + r := tracereplay.Replay{ + Endpoint: c.endpoint, + In: c.in, + } + if err := r.Execute(); err != nil { + fmt.Fprintln(os.Stderr, err) + return subcommands.ExitFailure + } + return subcommands.ExitSuccess +} diff --git a/tools/tracereplay/replay.go b/tools/tracereplay/replay.go new file mode 100644 index 000000000..3a2fc7b16 --- /dev/null +++ b/tools/tracereplay/replay.go @@ -0,0 +1,133 @@ +// Copyright 2022 The gVisor Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package tracereplay + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "os" + + "golang.org/x/sys/unix" + "google.golang.org/protobuf/proto" + "gvisor.dev/gvisor/pkg/log" + pb "gvisor.dev/gvisor/pkg/sentry/seccheck/points/points_go_proto" +) + +// Replay implements the functionality required for the "replay" command. +type Replay struct { + Endpoint string + In string +} + +// Execute connects to the remote endpoint and replays all messages stored in +// the `In` file. +func (r *Replay) Execute() error { + socket, err := connect(r.Endpoint) + if err != nil { + return err + } + defer socket.Close() + + f, err := os.Open(r.In) + if err != nil { + return err + } + defer f.Close() + + hdr := make([]byte, len(signature)) + if err := readFull(f, hdr); err != nil { + return err + } + if string(hdr) != signature { + return fmt.Errorf("%q is not a replay file", r.In) + } + + cfgJSON, err := readWithSize(f) + if err != nil { + return err + } + cfg := Config{} + if err := json.Unmarshal(cfgJSON, &cfg); err != nil { + return err + } + if err := handshake(socket, cfg.Version); err != nil { + return err + } + fmt.Printf("Handshake completed\n") + + for count := 1; ; count++ { + bytes, err := readWithSize(f) + if err != nil { + if errors.Is(err, io.EOF) { + break + } + return err + } + fmt.Printf("\rReplaying message: %d", count) + if _, err := socket.Write(bytes); err != nil { + return err + } + } + fmt.Printf("\nDone\n") + return nil +} + +func connect(endpoint string) (*os.File, error) { + log.Debugf("Connecting to %q", endpoint) + socket, err := unix.Socket(unix.AF_UNIX, unix.SOCK_SEQPACKET, 0) + if err != nil { + return nil, fmt.Errorf("socket(AF_UNIX, SOCK_SEQPACKET, 0): %w", err) + } + f := os.NewFile(uintptr(socket), endpoint) + + addr := unix.SockaddrUnix{Name: endpoint} + if err := unix.Connect(int(f.Fd()), &addr); err != nil { + _ = f.Close() + return nil, fmt.Errorf("connect(%q): %w", endpoint, err) + } + return f, nil +} + +// See common.proto for details about the handshake protocol. +func handshake(socket *os.File, version uint32) error { + hsOut := pb.Handshake{Version: version} + out, err := proto.Marshal(&hsOut) + if err != nil { + return err + } + if _, err := socket.Write(out); err != nil { + return fmt.Errorf("sending handshake message: %w", err) + } + + in := make([]byte, 10240) + read, err := socket.Read(in) + if err != nil && !errors.Is(err, io.EOF) { + return fmt.Errorf("reading handshake message: %w", err) + } + // Protect against the handshake becoming larger than the buffer. + if read == len(in) { + return fmt.Errorf("handshake message too big") + } + hsIn := pb.Handshake{} + if err := proto.Unmarshal(in[:read], &hsIn); err != nil { + return fmt.Errorf("unmarshalling handshake message: %w", err) + } + + // Just validate that the message can unmarshall and accept any version from + // the server. Will try to replay and see what happens... + return nil +} diff --git a/tools/tracereplay/save.go b/tools/tracereplay/save.go new file mode 100644 index 000000000..b12a00a62 --- /dev/null +++ b/tools/tracereplay/save.go @@ -0,0 +1,111 @@ +// Copyright 2022 The gVisor Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package tracereplay + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + + "gvisor.dev/gvisor/pkg/atomicbitops" + "gvisor.dev/gvisor/pkg/sentry/seccheck/checkers/remote/server" + "gvisor.dev/gvisor/pkg/sentry/seccheck/checkers/remote/wire" +) + +// Save implements the functionality required for the "save" command. +type Save struct { + server.CommonServer + dir string + prefix string + clientCount atomicbitops.Uint64 +} + +var _ server.ClientHandler = (*Save)(nil) + +// NewSave creates a new Save instance. +func NewSave(endpoint, dir, prefix string) *Save { + s := &Save{dir: dir, prefix: prefix} + s.CommonServer.Init(endpoint, s) + return s +} + +// Start starts the server. +func (s *Save) Start() error { + if err := os.MkdirAll(s.dir, 0755); err != nil { + return err + } + return s.CommonServer.Start() +} + +// NewClient creates a new file for the client and writes messages to it. +// +// The file format starts with a string signature to make it easy to check that +// it's a trace file. The signature is followed by a JSON configuration that +// contains information required to process the file. Next, there are a sequence +// of messages. Both JSON and messages are prefixed by an uint64 with their +// size. +// +// Ex: +// signature Config JSON [message]* +func (s *Save) NewClient() (server.MessageHandler, error) { + seq := s.clientCount.Add(1) + filename := filepath.Join(s.dir, fmt.Sprintf("%s%04d", s.prefix, seq)) + fmt.Printf("New client connected, writing to: %q\n", filename) + + out, err := os.Create(filename) + if err != nil { + return nil, err + } + if _, err := out.Write([]byte(signature)); err != nil { + return nil, err + } + + handler := &msgHandler{out: out} + + cfg, err := json.Marshal(Config{Version: handler.Version()}) + if err != nil { + return nil, err + } + if err := writeWithSize(out, cfg); err != nil { + return nil, err + } + + return handler, nil +} + +type msgHandler struct { + out *os.File + messageCount atomicbitops.Uint64 +} + +var _ server.MessageHandler = (*msgHandler)(nil) + +// Version implements server.MessageHandler. +func (m *msgHandler) Version() uint32 { + return wire.CurrentVersion +} + +// Message saves the message to the client file. +func (m *msgHandler) Message(raw []byte, _ wire.Header, _ []byte) error { + m.messageCount.Add(1) + return writeWithSize(m.out, raw) +} + +// Close closes the client file. +func (m *msgHandler) Close() { + fmt.Printf("Closing client, wrote %d messages to %q\n", m.messageCount.Load(), m.out.Name()) + _ = m.out.Close() +} diff --git a/tools/tracereplay/testdata/client-0001 b/tools/tracereplay/testdata/client-0001 new file mode 100644 index 0000000000000000000000000000000000000000..cdcc767302a7522b9e7e357fc9080921c426f52a GIT binary patch literal 804 zcmXRZN=!~IN-fAqtW-$L%t_^CfPiYHvecsD%=|nhE5q6dh!_J010$Fvb46iu3i7 z6O%JiEgBjW48ST8dW4ubn2FJ6jiwJxkGy_PW|Dq|g;~6riEd72UTKAHdR{3=Fd1aM z8OUmTG$XKBO^h*(aAP>ZEG|EeWqYn}-`lkOmzV&f5yfGwz^tVNayWM1=o;x68|oG3 z8ysL00$UFXN{B;*m^luxDKHxxU@L^nuz^`zJ{)`Af8GOjA{89SsKIFP;pqG0J9dj& pusC!+g~fOPV=34=kdxpRKbzLDmr523F>$ac;I^CtZaFKM1pt{E;IRMz literal 0 HcmV?d00001 diff --git a/tools/tracereplay/tracereplay.go b/tools/tracereplay/tracereplay.go new file mode 100644 index 000000000..eef449c05 --- /dev/null +++ b/tools/tracereplay/tracereplay.go @@ -0,0 +1,83 @@ +// Copyright 2022 The gVisor Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package tracereplay implements a tool that can save and replay messages +// issued from remote.Remote. +package tracereplay + +import ( + "encoding/binary" + "fmt" + "io" + "os" +) + +const signature = "tracereplay file" + +func writeSize(w io.Writer, val int) error { + var bin [8]byte + binary.LittleEndian.PutUint64(bin[:], uint64(val)) + _, err := w.Write(bin[:]) + return err +} + +func readSize(r io.Reader) (int, error) { + var bin [8]byte + if read, err := r.Read(bin[:]); err != nil { + return 0, err + } else if read != 8 { + return 0, fmt.Errorf("truncated read (%d bytes)", read) + } + size := int(binary.LittleEndian.Uint64(bin[:])) + // Prevent returning a too large size to avoid OOMs. + if size > 1024*1024 { + return 0, fmt.Errorf("size is too big: %d", size) + } + return size, nil +} + +func writeWithSize(f *os.File, buf []byte) error { + if err := writeSize(f, len(buf)); err != nil { + return err + } + _, err := f.Write(buf) + return err +} + +func readWithSize(r io.Reader) ([]byte, error) { + size, err := readSize(r) + if err != nil { + return nil, err + } + bytes := make([]byte, size) + if err := readFull(r, bytes); err != nil { + return nil, err + } + return bytes, nil +} + +func readFull(r io.Reader, dest []byte) error { + if read, err := r.Read(dest); err != nil { + return err + } else if read < len(dest) { + return fmt.Errorf("truncated read. Read %d bytes, expected %d bytes", read, len(dest)) + } + return nil +} + +// Config contains information required to replay messages from a file. +type Config struct { + // Version is the wire format saved in the file. + Version uint32 `json:"version"` +} diff --git a/tools/tracereplay/tracereplay_test.go b/tools/tracereplay/tracereplay_test.go new file mode 100644 index 000000000..f3756b001 --- /dev/null +++ b/tools/tracereplay/tracereplay_test.go @@ -0,0 +1,76 @@ +// Copyright 2022 The gVisor Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package tracereplay + +import ( + "bytes" + "os" + "path/filepath" + "testing" + + "gvisor.dev/gvisor/pkg/test/testutil" +) + +// TestBasic uses a pre-generated file that is replayed into a save process. +// Then verifies that the generated file looks exactly the same as the original. +// In other words, it is doing `replay original | save new`, then checking if +// `original == new`. +func TestBasic(t *testing.T) { + dir, err := os.MkdirTemp(testutil.TmpDir(), "tracereplay") + if err != nil { + t.Fatal(err) + } + endpoint := filepath.Join(dir, "tracereplay.sock") + + // Start a new save server to store the replayed file. This tests that save + // communicates with clients correctly and generates a valid file. + s := NewSave(endpoint, filepath.Join(dir, "out"), "test-") + defer s.Close() + + if err := s.Start(); err != nil { + t.Fatal(err) + } + + // Then replay the re-generated file. This tests that replay can connect to + // a server and process the generated file. + r := Replay{} + r.Endpoint = endpoint + + const testdata = "tools/tracereplay/testdata/client-0001" + r.In, err = testutil.FindFile(testdata) + if err != nil { + t.Fatalf("FindFile(%q): %v", testdata, err) + } + + if err := r.Execute(); err != nil { + t.Fatal(err) + } + + // Wait until all messages are processed and client disconnects. + s.WaitForNoClients() + + // The generated file must be an exact copy of the original file. + want, err := os.ReadFile(r.In) + if err != nil { + t.Fatal(err) + } + got, err := os.ReadFile(filepath.Join(dir, "out", "test-0001")) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(want, got) { + t.Errorf("files don't match\nwant: %s\ngot: %s", want, got) + } +}