chore!: route connection logs to new table (#18340)

### Breaking Change (changelog note): > User connections to workspaces, and the opening of workspace apps or ports will no longer create entries in the audit log. Those events will now be included in the 'Connection Log'. Please see the 'Connection Log' page in the dashboard, and the Connection Log [documentation](https://coder.com/docs/admin/monitoring/connection-logs) for details. Those with permission to view the Audit Log will also be able to view the Connection Log. The new Connection Log has the same licensing restrictions as the Audit Log, and requires a Premium Coder deployment. ### Context This is the first PR of a few for moving connection events out of the audit log, and into a new database table and web UI page called the 'Connection Log'. This PR: - Creates the new table - Adds and tests queries for inserting and reading, including reading with an RBAC filter. - Implements the corresponding RBAC changes, such that anyone who can view the audit log can read from the table - Implements, under the enterprise package, a `ConnectionLogger` abstraction to replace the `Auditor` abstraction for these logs. (No-op'd in AGPL, like the `Auditor`) - Routes SSH connection and Workspace App events into the new `ConnectionLogger` - Updates all existing tests to check the values of the `ConnectionLogger` instead of the `Auditor`. Future PRs: - Add filtering to the query - Add an enterprise endpoint to query the new table - Write a query to delete old events from the audit log, call it from dbpurge. - Implement a table in the Web UI for viewing connection logs. > [!NOTE] > The PRs in this stack obviously won't be (completely) atomic. Whilst they'll each pass CI, the stack is designed to be merged all at once. I'm splitting them up for the sake of those reviewing, and so changes can be reviewed as early as possible. Despite this, it's really hard to make this PR any smaller than it already is. I'll be keeping it in draft until it's actually ready to merge.
2026-09-24 15:04:27 +08:00 · 2025-07-15 14:36:06 +10:00
parent 43b0bb7f61
commit 08e17a07fc
54 changed files with 2199 additions and 493 deletions
@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"net"
 	"sort"
 	"testing"
 	"time"
@@ -13,6 +14,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/lib/pq"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/sqlc-dev/pqtype"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"

@@ -2085,6 +2087,447 @@ func auditOnlyIDs[T database.AuditLog | database.GetAuditLogsOffsetRow](logs []T
 	return ids
 }

+func TestGetAuthorizedConnectionLogsOffset(t *testing.T) {
+	t.Parallel()
+
+	var allLogs []database.ConnectionLog
+	db, _ := dbtestutil.NewDB(t)
+	authz := rbac.NewAuthorizer(prometheus.NewRegistry())
+	authDb := dbauthz.New(db, authz, slogtest.Make(t, &slogtest.Options{}), coderdtest.AccessControlStorePointer())
+
+	orgA := dbfake.Organization(t, db).Do()
+	orgB := dbfake.Organization(t, db).Do()
+
+	user := dbgen.User(t, db, database.User{})
+
+	tpl := dbgen.Template(t, db, database.Template{
+		OrganizationID: orgA.Org.ID,
+		CreatedBy:      user.ID,
+	})
+
+	wsID := uuid.New()
+	createTemplateVersion(t, db, tpl, tvArgs{
+		WorkspaceTransition: database.WorkspaceTransitionStart,
+		Status:              database.ProvisionerJobStatusSucceeded,
+		CreateWorkspace:     true,
+		WorkspaceID:         wsID,
+	})
+
+	// This map is a simple way to insert a given number of organizations
+	// and audit logs for each organization.
+	// map[orgID][]ConnectionLogID
+	orgConnectionLogs := map[uuid.UUID][]uuid.UUID{
+		orgA.Org.ID: {uuid.New(), uuid.New()},
+		orgB.Org.ID: {uuid.New(), uuid.New()},
+	}
+	orgIDs := make([]uuid.UUID, 0, len(orgConnectionLogs))
+	for orgID := range orgConnectionLogs {
+		orgIDs = append(orgIDs, orgID)
+	}
+	for orgID, ids := range orgConnectionLogs {
+		for _, id := range ids {
+			allLogs = append(allLogs, dbgen.ConnectionLog(t, authDb, database.UpsertConnectionLogParams{
+				WorkspaceID:      wsID,
+				WorkspaceOwnerID: user.ID,
+				ID:               id,
+				OrganizationID:   orgID,
+			}))
+		}
+	}
+
+	// Now fetch all the logs
+	ctx := testutil.Context(t, testutil.WaitLong)
+	auditorRole, err := rbac.RoleByName(rbac.RoleAuditor())
+	require.NoError(t, err)
+
+	memberRole, err := rbac.RoleByName(rbac.RoleMember())
+	require.NoError(t, err)
+
+	orgAuditorRoles := func(t *testing.T, orgID uuid.UUID) rbac.Role {
+		t.Helper()
+
+		role, err := rbac.RoleByName(rbac.ScopedRoleOrgAuditor(orgID))
+		require.NoError(t, err)
+		return role
+	}
+
+	t.Run("NoAccess", func(t *testing.T) {
+		t.Parallel()
+
+		// Given: A user who is a member of 0 organizations
+		memberCtx := dbauthz.As(ctx, rbac.Subject{
+			FriendlyName: "member",
+			ID:           uuid.NewString(),
+			Roles:        rbac.Roles{memberRole},
+			Scope:        rbac.ScopeAll,
+		})
+
+		// When: The user queries for connection logs
+		logs, err := authDb.GetConnectionLogsOffset(memberCtx, database.GetConnectionLogsOffsetParams{})
+		require.NoError(t, err)
+		// Then: No logs returned
+		require.Len(t, logs, 0, "no logs should be returned")
+	})
+
+	t.Run("SiteWideAuditor", func(t *testing.T) {
+		t.Parallel()
+
+		// Given: A site wide auditor
+		siteAuditorCtx := dbauthz.As(ctx, rbac.Subject{
+			FriendlyName: "owner",
+			ID:           uuid.NewString(),
+			Roles:        rbac.Roles{auditorRole},
+			Scope:        rbac.ScopeAll,
+		})
+
+		// When: the auditor queries for connection logs
+		logs, err := authDb.GetConnectionLogsOffset(siteAuditorCtx, database.GetConnectionLogsOffsetParams{})
+		require.NoError(t, err)
+		// Then: All logs are returned
+		require.ElementsMatch(t, connectionOnlyIDs(allLogs), connectionOnlyIDs(logs))
+	})
+
+	t.Run("SingleOrgAuditor", func(t *testing.T) {
+		t.Parallel()
+
+		orgID := orgIDs[0]
+		// Given: An organization scoped auditor
+		orgAuditCtx := dbauthz.As(ctx, rbac.Subject{
+			FriendlyName: "org-auditor",
+			ID:           uuid.NewString(),
+			Roles:        rbac.Roles{orgAuditorRoles(t, orgID)},
+			Scope:        rbac.ScopeAll,
+		})
+
+		// When: The auditor queries for connection logs
+		logs, err := authDb.GetConnectionLogsOffset(orgAuditCtx, database.GetConnectionLogsOffsetParams{})
+		require.NoError(t, err)
+		// Then: Only the logs for the organization are returned
+		require.ElementsMatch(t, orgConnectionLogs[orgID], connectionOnlyIDs(logs))
+	})
+
+	t.Run("TwoOrgAuditors", func(t *testing.T) {
+		t.Parallel()
+
+		first := orgIDs[0]
+		second := orgIDs[1]
+		// Given: A user who is an auditor for two organizations
+		multiOrgAuditCtx := dbauthz.As(ctx, rbac.Subject{
+			FriendlyName: "org-auditor",
+			ID:           uuid.NewString(),
+			Roles:        rbac.Roles{orgAuditorRoles(t, first), orgAuditorRoles(t, second)},
+			Scope:        rbac.ScopeAll,
+		})
+
+		// When: The user queries for connection logs
+		logs, err := authDb.GetConnectionLogsOffset(multiOrgAuditCtx, database.GetConnectionLogsOffsetParams{})
+		require.NoError(t, err)
+		// Then: All logs for both organizations are returned
+		require.ElementsMatch(t, append(orgConnectionLogs[first], orgConnectionLogs[second]...), connectionOnlyIDs(logs))
+	})
+
+	t.Run("ErroneousOrg", func(t *testing.T) {
+		t.Parallel()
+
+		// Given: A user who is an auditor for an organization that has 0 logs
+		userCtx := dbauthz.As(ctx, rbac.Subject{
+			FriendlyName: "org-auditor",
+			ID:           uuid.NewString(),
+			Roles:        rbac.Roles{orgAuditorRoles(t, uuid.New())},
+			Scope:        rbac.ScopeAll,
+		})
+
+		// When: The user queries for audit logs
+		logs, err := authDb.GetConnectionLogsOffset(userCtx, database.GetConnectionLogsOffsetParams{})
+		require.NoError(t, err)
+		// Then: No logs are returned
+		require.Len(t, logs, 0, "no logs should be returned")
+	})
+}
+
+func connectionOnlyIDs[T database.ConnectionLog | database.GetConnectionLogsOffsetRow](logs []T) []uuid.UUID {
+	ids := make([]uuid.UUID, 0, len(logs))
+	for _, log := range logs {
+		switch log := any(log).(type) {
+		case database.ConnectionLog:
+			ids = append(ids, log.ID)
+		case database.GetConnectionLogsOffsetRow:
+			ids = append(ids, log.ConnectionLog.ID)
+		default:
+			panic("unreachable")
+		}
+	}
+	return ids
+}
+
+func TestUpsertConnectionLog(t *testing.T) {
+	t.Parallel()
+	createWorkspace := func(t *testing.T, db database.Store) database.WorkspaceTable {
+		u := dbgen.User(t, db, database.User{})
+		o := dbgen.Organization(t, db, database.Organization{})
+		tpl := dbgen.Template(t, db, database.Template{
+			OrganizationID: o.ID,
+			CreatedBy:      u.ID,
+		})
+		return dbgen.Workspace(t, db, database.WorkspaceTable{
+			ID:               uuid.New(),
+			OwnerID:          u.ID,
+			OrganizationID:   o.ID,
+			AutomaticUpdates: database.AutomaticUpdatesNever,
+			TemplateID:       tpl.ID,
+		})
+	}
+
+	t.Run("ConnectThenDisconnect", func(t *testing.T) {
+		t.Parallel()
+		db, _ := dbtestutil.NewDB(t)
+		ctx := context.Background()
+
+		ws := createWorkspace(t, db)
+
+		connectionID := uuid.New()
+		agentName := "test-agent"
+
+		// 1. Insert a 'connect' event.
+		connectTime := dbtime.Now()
+		connectParams := database.UpsertConnectionLogParams{
+			ID:               uuid.New(),
+			Time:             connectTime,
+			OrganizationID:   ws.OrganizationID,
+			WorkspaceOwnerID: ws.OwnerID,
+			WorkspaceID:      ws.ID,
+			WorkspaceName:    ws.Name,
+			AgentName:        agentName,
+			Type:             database.ConnectionTypeSsh,
+			ConnectionID:     uuid.NullUUID{UUID: connectionID, Valid: true},
+			ConnectionStatus: database.ConnectionStatusConnected,
+			Ip: pqtype.Inet{
+				IPNet: net.IPNet{
+					IP:   net.IPv4(127, 0, 0, 1),
+					Mask: net.IPv4Mask(255, 255, 255, 255),
+				},
+				Valid: true,
+			},
+		}
+
+		log1, err := db.UpsertConnectionLog(ctx, connectParams)
+		require.NoError(t, err)
+		require.Equal(t, connectParams.ID, log1.ID)
+		require.False(t, log1.DisconnectTime.Valid, "CloseTime should not be set on connect")
+
+		// Check that one row exists.
+		rows, err := db.GetConnectionLogsOffset(ctx, database.GetConnectionLogsOffsetParams{LimitOpt: 10})
+		require.NoError(t, err)
+		require.Len(t, rows, 1)
+
+		// 2. Insert a 'disconnected' event for the same connection.
+		disconnectTime := connectTime.Add(time.Second)
+		disconnectParams := database.UpsertConnectionLogParams{
+			ConnectionID:     uuid.NullUUID{UUID: connectionID, Valid: true},
+			WorkspaceID:      ws.ID,
+			AgentName:        agentName,
+			ConnectionStatus: database.ConnectionStatusDisconnected,
+
+			// Updated to:
+			Time:             disconnectTime,
+			DisconnectReason: sql.NullString{String: "test disconnect", Valid: true},
+			Code:             sql.NullInt32{Int32: 1, Valid: true},
+
+			// Ignored
+			ID:               uuid.New(),
+			OrganizationID:   ws.OrganizationID,
+			WorkspaceOwnerID: ws.OwnerID,
+			WorkspaceName:    ws.Name,
+			Type:             database.ConnectionTypeSsh,
+			Ip: pqtype.Inet{
+				IPNet: net.IPNet{
+					IP:   net.IPv4(127, 0, 0, 1),
+					Mask: net.IPv4Mask(255, 255, 255, 254),
+				},
+				Valid: true,
+			},
+		}
+
+		log2, err := db.UpsertConnectionLog(ctx, disconnectParams)
+		require.NoError(t, err)
+
+		// Updated
+		require.Equal(t, log1.ID, log2.ID)
+		require.True(t, log2.DisconnectTime.Valid)
+		require.True(t, disconnectTime.Equal(log2.DisconnectTime.Time))
+		require.Equal(t, disconnectParams.DisconnectReason.String, log2.DisconnectReason.String)
+
+		rows, err = db.GetConnectionLogsOffset(ctx, database.GetConnectionLogsOffsetParams{})
+		require.NoError(t, err)
+		require.Len(t, rows, 1)
+	})
+
+	t.Run("ConnectDoesNotUpdate", func(t *testing.T) {
+		t.Parallel()
+		db, _ := dbtestutil.NewDB(t)
+		ctx := context.Background()
+
+		ws := createWorkspace(t, db)
+
+		connectionID := uuid.New()
+		agentName := "test-agent"
+
+		// 1. Insert a 'connect' event.
+		connectTime := dbtime.Now()
+		connectParams := database.UpsertConnectionLogParams{
+			ID:               uuid.New(),
+			Time:             connectTime,
+			OrganizationID:   ws.OrganizationID,
+			WorkspaceOwnerID: ws.OwnerID,
+			WorkspaceID:      ws.ID,
+			WorkspaceName:    ws.Name,
+			AgentName:        agentName,
+			Type:             database.ConnectionTypeSsh,
+			ConnectionID:     uuid.NullUUID{UUID: connectionID, Valid: true},
+			ConnectionStatus: database.ConnectionStatusConnected,
+			Ip: pqtype.Inet{
+				IPNet: net.IPNet{
+					IP:   net.IPv4(127, 0, 0, 1),
+					Mask: net.IPv4Mask(255, 255, 255, 255),
+				},
+				Valid: true,
+			},
+		}
+
+		log, err := db.UpsertConnectionLog(ctx, connectParams)
+		require.NoError(t, err)
+
+		// 2. Insert another 'connect' event for the same connection.
+		connectTime2 := connectTime.Add(time.Second)
+		connectParams2 := database.UpsertConnectionLogParams{
+			ConnectionID:     uuid.NullUUID{UUID: connectionID, Valid: true},
+			WorkspaceID:      ws.ID,
+			AgentName:        agentName,
+			ConnectionStatus: database.ConnectionStatusConnected,
+
+			// Ignored
+			ID:               uuid.New(),
+			Time:             connectTime2,
+			OrganizationID:   ws.OrganizationID,
+			WorkspaceOwnerID: ws.OwnerID,
+			WorkspaceName:    ws.Name,
+			Type:             database.ConnectionTypeSsh,
+			Code:             sql.NullInt32{Int32: 0, Valid: false},
+			Ip: pqtype.Inet{
+				IPNet: net.IPNet{
+					IP:   net.IPv4(127, 0, 0, 1),
+					Mask: net.IPv4Mask(255, 255, 255, 254),
+				},
+				Valid: true,
+			},
+		}
+
+		origLog, err := db.UpsertConnectionLog(ctx, connectParams2)
+		require.NoError(t, err)
+		require.Equal(t, log, origLog, "connect update should be a no-op")
+
+		// Check that still only one row exists.
+		rows, err := db.GetConnectionLogsOffset(ctx, database.GetConnectionLogsOffsetParams{})
+		require.NoError(t, err)
+		require.Len(t, rows, 1)
+		require.Equal(t, log, rows[0].ConnectionLog)
+	})
+
+	t.Run("DisconnectThenConnect", func(t *testing.T) {
+		t.Parallel()
+
+		db, _ := dbtestutil.NewDB(t)
+		ctx := context.Background()
+
+		ws := createWorkspace(t, db)
+
+		connectionID := uuid.New()
+		agentName := "test-agent"
+
+		// Insert just a 'disconect' event
+		disconnectTime := dbtime.Now()
+		disconnectParams := database.UpsertConnectionLogParams{
+			ID:               uuid.New(),
+			Time:             disconnectTime,
+			OrganizationID:   ws.OrganizationID,
+			WorkspaceOwnerID: ws.OwnerID,
+			WorkspaceID:      ws.ID,
+			WorkspaceName:    ws.Name,
+			AgentName:        agentName,
+			Type:             database.ConnectionTypeSsh,
+			ConnectionID:     uuid.NullUUID{UUID: connectionID, Valid: true},
+			ConnectionStatus: database.ConnectionStatusDisconnected,
+			DisconnectReason: sql.NullString{String: "server shutting down", Valid: true},
+			Ip: pqtype.Inet{
+				IPNet: net.IPNet{
+					IP:   net.IPv4(127, 0, 0, 1),
+					Mask: net.IPv4Mask(255, 255, 255, 255),
+				},
+				Valid: true,
+			},
+		}
+
+		_, err := db.UpsertConnectionLog(ctx, disconnectParams)
+		require.NoError(t, err)
+
+		firstRows, err := db.GetConnectionLogsOffset(ctx, database.GetConnectionLogsOffsetParams{})
+		require.NoError(t, err)
+		require.Len(t, firstRows, 1)
+
+		// We expect the connection event to be marked as closed with the start
+		// and close time being the same.
+		require.True(t, firstRows[0].ConnectionLog.DisconnectTime.Valid)
+		require.Equal(t, disconnectTime, firstRows[0].ConnectionLog.DisconnectTime.Time.UTC())
+		require.Equal(t, firstRows[0].ConnectionLog.ConnectTime.UTC(), firstRows[0].ConnectionLog.DisconnectTime.Time.UTC())
+
+		// Now insert a 'connect' event for the same connection.
+		// This should be a no op
+		connectTime := disconnectTime.Add(time.Second)
+		connectParams := database.UpsertConnectionLogParams{
+			ID:               uuid.New(),
+			Time:             connectTime,
+			OrganizationID:   ws.OrganizationID,
+			WorkspaceOwnerID: ws.OwnerID,
+			WorkspaceID:      ws.ID,
+			WorkspaceName:    ws.Name,
+			AgentName:        agentName,
+			Type:             database.ConnectionTypeSsh,
+			ConnectionID:     uuid.NullUUID{UUID: connectionID, Valid: true},
+			ConnectionStatus: database.ConnectionStatusConnected,
+			DisconnectReason: sql.NullString{String: "reconnected", Valid: true},
+			Code:             sql.NullInt32{Int32: 0, Valid: false},
+			Ip: pqtype.Inet{
+				IPNet: net.IPNet{
+					IP:   net.IPv4(127, 0, 0, 1),
+					Mask: net.IPv4Mask(255, 255, 255, 255),
+				},
+				Valid: true,
+			},
+		}
+
+		_, err = db.UpsertConnectionLog(ctx, connectParams)
+		require.NoError(t, err)
+
+		secondRows, err := db.GetConnectionLogsOffset(ctx, database.GetConnectionLogsOffsetParams{})
+		require.NoError(t, err)
+		require.Len(t, secondRows, 1)
+		require.Equal(t, firstRows, secondRows)
+
+		// Upsert a disconnection, which should also be a no op
+		disconnectParams.DisconnectReason = sql.NullString{
+			String: "updated close reason",
+			Valid:  true,
+		}
+		_, err = db.UpsertConnectionLog(ctx, disconnectParams)
+		require.NoError(t, err)
+		thirdRows, err := db.GetConnectionLogsOffset(ctx, database.GetConnectionLogsOffsetParams{})
+		require.NoError(t, err)
+		require.Len(t, secondRows, 1)
+		// The close reason shouldn't be updated
+		require.Equal(t, secondRows, thirdRows)
+	})
+}
+
 type tvArgs struct {
 	Status database.ProvisionerJobStatus
 	// CreateWorkspace is true if we should create a workspace for the template version