diff --git a/agent/agent.go b/agent/agent.go index 4dd9072bc1..21918bbc52 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -43,6 +43,7 @@ import ( "github.com/coder/coder/v2/agent/agentscripts" "github.com/coder/coder/v2/agent/agentsocket" "github.com/coder/coder/v2/agent/agentssh" + "github.com/coder/coder/v2/agent/boundarylogproxy" "github.com/coder/coder/v2/agent/proto" "github.com/coder/coder/v2/agent/proto/resourcesmonitor" "github.com/coder/coder/v2/agent/reconnectingpty" @@ -102,6 +103,7 @@ type Options struct { Clock quartz.Clock SocketServerEnabled bool SocketPath string // Path for the agent socket server socket + BoundaryLogProxySocketPath string } type Client interface { @@ -205,10 +207,11 @@ func New(options Options) Agent { metrics: newAgentMetrics(prometheusRegistry), execer: options.Execer, - devcontainers: options.Devcontainers, - containerAPIOptions: options.DevcontainerAPIOptions, - socketPath: options.SocketPath, - socketServerEnabled: options.SocketServerEnabled, + devcontainers: options.Devcontainers, + containerAPIOptions: options.DevcontainerAPIOptions, + socketPath: options.SocketPath, + socketServerEnabled: options.SocketServerEnabled, + boundaryLogProxySocketPath: options.BoundaryLogProxySocketPath, } // Initially, we have a closed channel, reflecting the fact that we are not initially connected. // Each time we connect we replace the channel (while holding the closeMutex) with a new one @@ -277,6 +280,11 @@ type agent struct { logSender *agentsdk.LogSender + // boundaryLogProxy is a socket server that forwards boundary audit logs to coderd. + // It may be nil if there is a problem starting the server. + boundaryLogProxy *boundarylogproxy.Server + boundaryLogProxySocketPath string + prometheusRegistry *prometheus.Registry // metrics are prometheus registered metrics that will be collected and // labeled in Coder with the agent + workspace. @@ -371,6 +379,7 @@ func (a *agent) init() { ) a.initSocketServer() + a.startBoundaryLogProxyServer() go a.runLoop() } @@ -395,6 +404,19 @@ func (a *agent) initSocketServer() { a.logger.Debug(a.hardCtx, "socket server started", slog.F("path", a.socketPath)) } +// startBoundaryLogProxyServer starts the boundary log proxy socket server. +func (a *agent) startBoundaryLogProxyServer() { + proxy := boundarylogproxy.NewServer(a.logger, a.boundaryLogProxySocketPath) + if err := proxy.Start(); err != nil { + a.logger.Warn(a.hardCtx, "failed to start boundary log proxy", slog.Error(err)) + return + } + + a.boundaryLogProxy = proxy + a.logger.Info(a.hardCtx, "boundary log proxy server started", + slog.F("socket_path", a.boundaryLogProxySocketPath)) +} + // runLoop attempts to start the agent in a retry loop. // Coder may be offline temporarily, a connection issue // may be happening, but regardless after the intermittent @@ -1012,6 +1034,15 @@ func (a *agent) run() (retErr error) { return err }) + // Forward boundary audit logs to coderd if boundary log forwarding is enabled. + // These are audit logs so they should continue during graceful shutdown. + if a.boundaryLogProxy != nil { + proxyFunc := func(ctx context.Context, aAPI proto.DRPCAgentClient27) error { + return a.boundaryLogProxy.RunForwarder(ctx, aAPI) + } + connMan.startAgentAPI("boundary log proxy", gracefulShutdownBehaviorRemain, proxyFunc) + } + // part of graceful shut down is reporting the final lifecycle states, e.g "ShuttingDown" so the // lifecycle reporting has to be via gracefulShutdownBehaviorRemain connMan.startAgentAPI("report lifecycle", gracefulShutdownBehaviorRemain, a.reportLifecycle) @@ -1982,6 +2013,13 @@ func (a *agent) Close() error { a.logger.Error(a.hardCtx, "container API close", slog.Error(err)) } + if a.boundaryLogProxy != nil { + err = a.boundaryLogProxy.Close() + if err != nil { + a.logger.Warn(context.Background(), "close boundary log proxy", slog.Error(err)) + } + } + // Wait for the graceful shutdown to complete, but don't wait forever so // that we don't break user expectations. go func() { diff --git a/agent/boundary_logs_test.go b/agent/boundary_logs_test.go new file mode 100644 index 0000000000..63bf21fe25 --- /dev/null +++ b/agent/boundary_logs_test.go @@ -0,0 +1,165 @@ +//go:build linux || darwin + +package agent_test + +import ( + "context" + "net" + "path/filepath" + "sync" + "testing" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/timestamppb" + + "cdr.dev/slog" + + "github.com/coder/coder/v2/agent/boundarylogproxy" + "github.com/coder/coder/v2/agent/boundarylogproxy/codec" + agentproto "github.com/coder/coder/v2/agent/proto" + "github.com/coder/coder/v2/coderd/agentapi" + "github.com/coder/coder/v2/testutil" +) + +// logSink captures structured log entries for testing. +type logSink struct { + mu sync.Mutex + entries []slog.SinkEntry +} + +func (s *logSink) LogEntry(_ context.Context, e slog.SinkEntry) { + s.mu.Lock() + defer s.mu.Unlock() + s.entries = append(s.entries, e) +} + +func (*logSink) Sync() {} + +func (s *logSink) getEntries() []slog.SinkEntry { + s.mu.Lock() + defer s.mu.Unlock() + return append([]slog.SinkEntry{}, s.entries...) +} + +// getField returns the value of a field by name from a slog.Map. +func getField(fields slog.Map, name string) interface{} { + for _, f := range fields { + if f.Name == name { + return f.Value + } + } + return nil +} + +func sendBoundaryLogsRequest(t *testing.T, conn net.Conn, req *agentproto.ReportBoundaryLogsRequest) { + t.Helper() + + data, err := proto.Marshal(req) + require.NoError(t, err) + + err = codec.WriteFrame(conn, codec.TagV1, data) + require.NoError(t, err) +} + +// TestBoundaryLogs_EndToEnd is an end-to-end test that sends a protobuf +// message over the agent's unix socket (as boundary would) and verifies +// it is ultimately logged by coderd with the correct structured fields. +func TestBoundaryLogs_EndToEnd(t *testing.T) { + t.Parallel() + + socketPath := filepath.Join(testutil.TempDirUnixSocket(t), "boundary.sock") + srv := boundarylogproxy.NewServer(testutil.Logger(t), socketPath) + + err := srv.Start() + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, srv.Close()) }) + + sink := &logSink{} + logger := slog.Make(sink) + workspaceID := uuid.New() + reporter := &agentapi.BoundaryLogsAPI{ + Log: logger, + WorkspaceID: workspaceID, + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + forwarderDone := make(chan error, 1) + go func() { + forwarderDone <- srv.RunForwarder(ctx, reporter) + }() + + conn, err := net.Dial("unix", socketPath) + require.NoError(t, err) + defer conn.Close() + + // Allowed HTTP request. + req := &agentproto.ReportBoundaryLogsRequest{ + Logs: []*agentproto.BoundaryLog{ + { + Allowed: true, + Time: timestamppb.Now(), + Resource: &agentproto.BoundaryLog_HttpRequest_{ + HttpRequest: &agentproto.BoundaryLog_HttpRequest{ + Method: "GET", + Url: "https://example.com/allowed", + MatchedRule: "*.example.com", + }, + }, + }, + }, + } + sendBoundaryLogsRequest(t, conn, req) + + require.Eventually(t, func() bool { + return len(sink.getEntries()) >= 1 + }, testutil.WaitShort, testutil.IntervalFast) + + entries := sink.getEntries() + require.Len(t, entries, 1) + entry := entries[0] + require.Equal(t, slog.LevelInfo, entry.Level) + require.Equal(t, "boundary_request", entry.Message) + require.Equal(t, "allow", getField(entry.Fields, "decision")) + require.Equal(t, workspaceID.String(), getField(entry.Fields, "workspace_id")) + require.Equal(t, "GET", getField(entry.Fields, "http_method")) + require.Equal(t, "https://example.com/allowed", getField(entry.Fields, "http_url")) + require.Equal(t, "*.example.com", getField(entry.Fields, "matched_rule")) + + // Denied HTTP request. + req2 := &agentproto.ReportBoundaryLogsRequest{ + Logs: []*agentproto.BoundaryLog{ + { + Allowed: false, + Time: timestamppb.Now(), + Resource: &agentproto.BoundaryLog_HttpRequest_{ + HttpRequest: &agentproto.BoundaryLog_HttpRequest{ + Method: "POST", + Url: "https://blocked.com/denied", + }, + }, + }, + }, + } + sendBoundaryLogsRequest(t, conn, req2) + + require.Eventually(t, func() bool { + return len(sink.getEntries()) >= 2 + }, testutil.WaitShort, testutil.IntervalFast) + + entries = sink.getEntries() + entry = entries[1] + require.Len(t, entries, 2) + require.Equal(t, slog.LevelInfo, entry.Level) + require.Equal(t, "boundary_request", entry.Message) + require.Equal(t, "deny", getField(entry.Fields, "decision")) + require.Equal(t, workspaceID.String(), getField(entry.Fields, "workspace_id")) + require.Equal(t, "POST", getField(entry.Fields, "http_method")) + require.Equal(t, "https://blocked.com/denied", getField(entry.Fields, "http_url")) + require.Equal(t, nil, getField(entry.Fields, "matched_rule")) + + cancel() + <-forwarderDone +} diff --git a/agent/boundarylogproxy/codec/codec.go b/agent/boundarylogproxy/codec/codec.go new file mode 100644 index 0000000000..cda876c64d --- /dev/null +++ b/agent/boundarylogproxy/codec/codec.go @@ -0,0 +1,127 @@ +// Package codec implements the wire format for agent <-> boundary communication. +// +// Wire Format: +// - 8 bits: big-endian tag +// - 24 bits: big-endian length of the protobuf data (bit usage depends on tag) +// - length bytes: encoded protobuf data +// +// Note that while there are 24 bits available for the length, the actual maximum +// length depends on the tag. For TagV1, only 15 bits are used (MaxMessageSizeV1). +package codec + +import ( + "encoding/binary" + "io" + + "golang.org/x/xerrors" +) + +type Tag uint8 + +const ( + // TagV1 identifies the first revision of the protocol. This version has a maximum + // data length of MaxMessageSizeV1. + TagV1 Tag = 1 +) + +const ( + // DataLength is the number of bits used for the length of encoded protobuf data. + DataLength = 24 + + // tagLength is the number of bits used for the tag. + tagLength = 8 + + // MaxMessageSizeV1 is the maximum size of the encoded protobuf messages sent + // over the wire for the TagV1 tag. While the wire format allows 24 bits for + // length, TagV1 only uses 15 bits. + MaxMessageSizeV1 uint32 = 1 << 15 +) + +var ( + // ErrMessageTooLarge is returned when the message exceeds the maximum size + // allowed for the tag. + ErrMessageTooLarge = xerrors.New("message too large") + // ErrUnsupportedTag is returned when an unrecognized tag is encountered. + ErrUnsupportedTag = xerrors.New("unsupported tag") +) + +// WriteFrame writes a framed message with the given tag and data. The data +// must not exceed 2^DataLength in length. +func WriteFrame(w io.Writer, tag Tag, data []byte) error { + var maxSize uint32 + switch tag { + case TagV1: + maxSize = MaxMessageSizeV1 + default: + return xerrors.Errorf("%w: %d", ErrUnsupportedTag, tag) + } + + if len(data) > int(maxSize) { + return xerrors.Errorf("%w for tag %d: %d > %d", ErrMessageTooLarge, tag, len(data), maxSize) + } + + var header uint32 + //nolint:gosec // The length check above ensures there's no overflow. + header |= uint32(len(data)) + header |= uint32(tag) << DataLength + + if err := binary.Write(w, binary.BigEndian, header); err != nil { + return xerrors.Errorf("write header error: %w", err) + } + if _, err := w.Write(data); err != nil { + return xerrors.Errorf("write data error: %w", err) + } + + return nil +} + +// ReadFrame reads a framed message, returning the decoded tag and data. If the +// message size exceeds MaxMessageSizeV1, ErrMessageTooLarge is returned. The +// provided buf is used if it has sufficient capacity; otherwise a new buffer is +// allocated. To reuse the buffer across calls, pass in the returned data slice: +// +// buf := make([]byte, initialSize) +// for { +// _, buf, _ = ReadFrame(r, buf) +// } +func ReadFrame(r io.Reader, buf []byte) (Tag, []byte, error) { + var header uint32 + if err := binary.Read(r, binary.BigEndian, &header); err != nil { + return 0, nil, xerrors.Errorf("read header error: %w", err) + } + + const lengthMask = (1 << DataLength) - 1 + length := header & lengthMask + const tagMask = (1 << tagLength) - 1 // 0xFF + shifted := (header >> DataLength) & tagMask + if shifted > tagMask { + // This is really only here to satisfy the gosec linter. We know from above that + // shifted <= tagMask. + return 0, nil, xerrors.Errorf("invalid tag: %d", shifted) + } + tag := Tag(shifted) + + var maxSize uint32 + switch tag { + case TagV1: + maxSize = MaxMessageSizeV1 + default: + return 0, nil, xerrors.Errorf("%w: %d", ErrUnsupportedTag, tag) + } + + if length > maxSize { + return 0, nil, ErrMessageTooLarge + } + + if cap(buf) < int(length) { + buf = make([]byte, length) + } else { + buf = buf[:length:cap(buf)] + } + + if _, err := io.ReadFull(r, buf[:length]); err != nil { + return 0, nil, xerrors.Errorf("read full error: %w", err) + } + + return tag, buf[:length], nil +} diff --git a/agent/boundarylogproxy/codec/codec_test.go b/agent/boundarylogproxy/codec/codec_test.go new file mode 100644 index 0000000000..4ca719f2d0 --- /dev/null +++ b/agent/boundarylogproxy/codec/codec_test.go @@ -0,0 +1,145 @@ +package codec_test + +import ( + "bytes" + "encoding/binary" + "io" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/coder/coder/v2/agent/boundarylogproxy/codec" +) + +func TestRoundTrip(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + tag codec.Tag + data []byte + }{ + { + name: "empty data", + tag: codec.TagV1, + data: []byte{}, + }, + { + name: "simple data", + tag: codec.TagV1, + data: []byte("hello world"), + }, + { + name: "binary data", + tag: codec.TagV1, + data: []byte{0x00, 0x01, 0x02, 0xff, 0xfe}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + var buf bytes.Buffer + err := codec.WriteFrame(&buf, tt.tag, tt.data) + require.NoError(t, err) + + readBuf := make([]byte, codec.MaxMessageSizeV1) + tag, data, err := codec.ReadFrame(&buf, readBuf) + require.NoError(t, err) + require.Equal(t, tt.tag, tag) + require.Equal(t, tt.data, data) + }) + } +} + +func TestReadFrameTooLarge(t *testing.T) { + t.Parallel() + + // Hand construct a header that indicates the message size exceeds the maximum + // message size for codec.TagV1 by one. We just write the header to buf because + // we expect codec.ReadFrame to bail out when reading the invalid length. + header := uint32(codec.TagV1)< "/tmp)"). + byt = bytes.ReplaceAll(byt, []byte("/tmp/)"), []byte("/tmp)")) + for _, r := range []struct { old string new string @@ -145,7 +156,6 @@ func normalizeGoldenFile(t *testing.T, byt []byte) []byte { {"\r\n", "\n"}, {`~\.cache\coder`, "~/.cache/coder"}, {`C:\Users\RUNNER~1\AppData\Local\Temp`, "/tmp"}, - {os.TempDir(), "/tmp"}, } { byt = bytes.ReplaceAll(byt, []byte(r.old), []byte(r.new)) } diff --git a/cli/testdata/coder_agent_--help.golden b/cli/testdata/coder_agent_--help.golden index d262c0d0c7..16e4680547 100644 --- a/cli/testdata/coder_agent_--help.golden +++ b/cli/testdata/coder_agent_--help.golden @@ -39,6 +39,10 @@ OPTIONS: --block-file-transfer bool, $CODER_AGENT_BLOCK_FILE_TRANSFER (default: false) Block file transfer using known applications: nc,rsync,scp,sftp. + --boundary-log-proxy-socket-path string, $CODER_AGENT_BOUNDARY_LOG_PROXY_SOCKET_PATH (default: /tmp/boundary-audit.sock) + The path for the boundary log proxy server Unix socket. Boundary + should write audit logs to this socket. + --debug-address string, $CODER_AGENT_DEBUG_ADDRESS (default: 127.0.0.1:2113) The bind address to serve a debug HTTP server. diff --git a/coderd/agentapi/api.go b/coderd/agentapi/api.go index aec9c0dfa6..fd44873a8f 100644 --- a/coderd/agentapi/api.go +++ b/coderd/agentapi/api.go @@ -220,7 +220,10 @@ func New(opts Options, workspace database.Workspace) *API { Database: opts.Database, } - api.BoundaryLogsAPI = &BoundaryLogsAPI{} + api.BoundaryLogsAPI = &BoundaryLogsAPI{ + Log: opts.Log, + WorkspaceID: opts.WorkspaceID, + } // Start background cache refresh loop to handle workspace changes // like prebuild claims where owner_id and other fields may be modified in the DB. diff --git a/coderd/agentapi/boundary_logs.go b/coderd/agentapi/boundary_logs.go index 3f4f02007b..1f2cf99526 100644 --- a/coderd/agentapi/boundary_logs.go +++ b/coderd/agentapi/boundary_logs.go @@ -2,14 +2,60 @@ package agentapi import ( "context" + "time" - "golang.org/x/xerrors" + "github.com/google/uuid" + + "cdr.dev/slog" agentproto "github.com/coder/coder/v2/agent/proto" ) -type BoundaryLogsAPI struct{} +type BoundaryLogsAPI struct { + Log slog.Logger + WorkspaceID uuid.UUID +} -func (*BoundaryLogsAPI) ReportBoundaryLogs(context.Context, *agentproto.ReportBoundaryLogsRequest) (*agentproto.ReportBoundaryLogsResponse, error) { - return nil, xerrors.New("not implemented") +func (a *BoundaryLogsAPI) ReportBoundaryLogs(ctx context.Context, req *agentproto.ReportBoundaryLogsRequest) (*agentproto.ReportBoundaryLogsResponse, error) { + for _, l := range req.Logs { + var logTime time.Time + if l.Time != nil { + logTime = l.Time.AsTime() + } + + switch r := l.Resource.(type) { + case *agentproto.BoundaryLog_HttpRequest_: + if r.HttpRequest == nil { + a.Log.Warn(ctx, "empty http request resource", + slog.F("workspace_id", a.WorkspaceID.String())) + continue + } + + fields := []slog.Field{ + slog.F("decision", allowBoolToString(l.Allowed)), + slog.F("workspace_id", a.WorkspaceID.String()), + slog.F("http_method", r.HttpRequest.Method), + slog.F("http_url", r.HttpRequest.Url), + slog.F("event_time", logTime.Format(time.RFC3339Nano)), + } + if l.Allowed { + fields = append(fields, slog.F("matched_rule", r.HttpRequest.MatchedRule)) + } + + a.Log.With(fields...).Info(ctx, "boundary_request") + default: + a.Log.Warn(ctx, "unknown resource type", + slog.F("workspace_id", a.WorkspaceID.String())) + } + } + + return &agentproto.ReportBoundaryLogsResponse{}, nil +} + +//nolint:revive // This stringifies the boolean argument. +func allowBoolToString(b bool) string { + if b { + return "allow" + } + return "deny" }