diff --git a/go.mod b/go.mod index 8653de2..de1c394 100644 --- a/go.mod +++ b/go.mod @@ -16,7 +16,7 @@ require ( github.com/landlock-lsm/go-landlock v0.7.0 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 github.com/posthog/posthog-go v1.5.12 - github.com/safedep/dry v0.0.0-20260710090004-346776184be5 + github.com/safedep/dry v0.0.0-20260716095238-84cd2b3cd3a4 github.com/safedep/ptyx v0.2.1-0.20260529140457-d1f745842a6a github.com/sony/gobreaker/v2 v2.4.0 github.com/spf13/cobra v1.9.1 diff --git a/go.sum b/go.sum index 9191d01..71f30c1 100644 --- a/go.sum +++ b/go.sum @@ -184,6 +184,8 @@ github.com/safedep/dry v0.0.0-20260710084513-c7378927978f h1:WeminE3k3HbGPhPPsJc github.com/safedep/dry v0.0.0-20260710084513-c7378927978f/go.mod h1:WfJPfXgWWLfgi72PGllhHRUGoj7vh+zJ5cx4RJ4qRdM= github.com/safedep/dry v0.0.0-20260710090004-346776184be5 h1:KDcSwAhNqLudyqn4iJEYacbAc1yf/g613BnpPTi7dm0= github.com/safedep/dry v0.0.0-20260710090004-346776184be5/go.mod h1:WfJPfXgWWLfgi72PGllhHRUGoj7vh+zJ5cx4RJ4qRdM= +github.com/safedep/dry v0.0.0-20260716095238-84cd2b3cd3a4 h1:9u5mj4LOteAm89eN0hg6eJ38kzc1ArfEjRiSXcKB2v8= +github.com/safedep/dry v0.0.0-20260716095238-84cd2b3cd3a4/go.mod h1:OO3Tcxd+SBHRaN42jv2RjgwZWsUa8jFHB0mDMi5HWyY= github.com/safedep/ptyx v0.2.1-0.20260529140457-d1f745842a6a h1:oJu4dgmz/weiU3CMhFKiXd5zwgvPwPsm20MzG/uAt0s= github.com/safedep/ptyx v0.2.1-0.20260529140457-d1f745842a6a/go.mod h1:fyt+PACz6dtEoqsnE0BPPv/lHpuBG/8zkDqeIVcyRY4= github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc= diff --git a/internal/audit/cloud_client.go b/internal/audit/cloud_client.go index 794751d..39d1d7a 100644 --- a/internal/audit/cloud_client.go +++ b/internal/audit/cloud_client.go @@ -124,12 +124,7 @@ func NewSyncClientBundle(cfg *config.RuntimeConfig) (*SyncClientBundle, error) { identity := endpointsync.NewEndpointIdentityResolver(identityOpts...) - toolVersion := appVersion.Version - if toolVersion == "" { - toolVersion = "dev" - } - - syncClient, err := endpointsync.NewSyncClient("pmg", toolVersion, transport, identity, + syncClient, err := endpointsync.NewSyncClient("pmg", pmgToolVersion(), transport, identity, endpointsync.WithWALPath(cfg.CloudSyncDBPath())) if err != nil { if closeErr := cloudClient.Close(); closeErr != nil { @@ -143,3 +138,12 @@ func NewSyncClientBundle(cfg *config.RuntimeConfig) (*SyncClientBundle, error) { cloudClient: cloudClient, }, nil } + +// pmgToolVersion returns the tool version stamped on cloud events, defaulting +// to "dev" for local builds. +func pmgToolVersion() string { + if appVersion.Version == "" { + return "dev" + } + return appVersion.Version +} diff --git a/internal/audit/cloud_sink.go b/internal/audit/cloud_sink.go index 1728f2d..cae920a 100644 --- a/internal/audit/cloud_sink.go +++ b/internal/audit/cloud_sink.go @@ -17,7 +17,7 @@ import ( ) type cloudSink struct { - *SyncClientBundle + emitter *endpointsync.EventEmitterClient invocationID string ciResolver CloudSinkCIResolver command string @@ -25,32 +25,33 @@ type cloudSink struct { } func newCloudSink(cfg *config.RuntimeConfig, ciResolver CloudSinkCIResolver) (*cloudSink, error) { - bundle, err := NewSyncClientBundle(cfg) + emitter, err := endpointsync.NewEventEmitterClient("pmg", pmgToolVersion(), + endpointsync.WithWALPath(cfg.CloudSyncDBPath())) if err != nil { return nil, err } invocationID, err := uuid.NewRandom() if err != nil { - if closeErr := bundle.Close(); closeErr != nil { - log.Warnf("failed to close sync client bundle after invocation ID failure: %v", closeErr) + if closeErr := emitter.Close(); closeErr != nil { + log.Warnf("failed to close event emitter after invocation ID failure: %v", closeErr) } return nil, fmt.Errorf("failed to generate invocation ID: %w", err) } wd, err := os.Getwd() if err != nil { - if closeErr := bundle.Close(); closeErr != nil { - log.Warnf("failed to close sync client bundle after getwd failure: %v", closeErr) + if closeErr := emitter.Close(); closeErr != nil { + log.Warnf("failed to close event emitter after getwd failure: %v", closeErr) } return nil, fmt.Errorf("failed to get working directory: %w", err) } return &cloudSink{ - SyncClientBundle: bundle, - invocationID: invocationID.String(), - ciResolver: ciResolver, - workingDir: wd, + emitter: emitter, + invocationID: invocationID.String(), + ciResolver: ciResolver, + workingDir: wd, }, nil } @@ -66,7 +67,7 @@ func (s *cloudSink) Handle(ctx context.Context, event AuditEvent) error { } for _, pmgEvent := range pmgEvents { - toolEvent, err := s.syncClient.NewEvent() + toolEvent, err := s.emitter.NewEvent() if err != nil { return fmt.Errorf("failed to create tool event: %w", err) } @@ -79,7 +80,7 @@ func (s *cloudSink) Handle(ctx context.Context, event AuditEvent) error { toolEvent.SetInvocationContext(s.buildInvocationContext()) } - if err := s.syncClient.Emit(ctx, toolEvent); err != nil { + if err := s.emitter.Emit(ctx, toolEvent); err != nil { if errors.Is(err, endpointsync.ErrWALFull) { log.Warnf("Cloud sync WAL is full, dropping event: %v", err) return nil @@ -159,7 +160,7 @@ func buildCommand(packageManager string, args []string) string { return packageManager + " " + strings.Join(args, " ") } -// Close delegates to the embedded SyncClientBundle.Close(). +// Close releases the emitter's WAL handle. func (s *cloudSink) Close() error { - return s.SyncClientBundle.Close() + return s.emitter.Close() } diff --git a/internal/audit/cloud_sink_test.go b/internal/audit/cloud_sink_test.go index 815fdfb..55b4416 100644 --- a/internal/audit/cloud_sink_test.go +++ b/internal/audit/cloud_sink_test.go @@ -31,24 +31,38 @@ func (m *mockTransport) Close() error { return nil } -func newTestCloudSink(t *testing.T, transport endpointsync.EventTransport) *cloudSink { +func newTestCloudSink(t *testing.T) (*cloudSink, string) { t.Helper() walPath := t.TempDir() + "/test-sync.db" - identity := endpointsync.NewEndpointIdentityResolver() - syncClient, err := endpointsync.NewSyncClient("pmg", "test", transport, identity, + emitter, err := endpointsync.NewEventEmitterClient("pmg", "test", endpointsync.WithWALPath(walPath)) require.NoError(t, err) return &cloudSink{ - SyncClientBundle: &SyncClientBundle{syncClient: syncClient}, - invocationID: "test-invocation", - workingDir: t.TempDir(), - } + emitter: emitter, + invocationID: "test-invocation", + workingDir: t.TempDir(), + }, walPath +} + +// drainWAL closes the sink (mirroring the real lifecycle, where audit.Close() +// runs before a sync process starts) and syncs the WAL at walPath through a +// SyncClient backed by transport, returning the number of events synced. +func drainWAL(t *testing.T, walPath string, transport endpointsync.EventTransport) int { + t.Helper() + syncClient, err := endpointsync.NewSyncClient("pmg", "test", transport, + endpointsync.NewEndpointIdentityResolver(), endpointsync.WithWALPath(walPath)) + require.NoError(t, err) + defer func() { + require.NoError(t, syncClient.Close()) + }() + + synced, err := syncClient.Sync(context.Background()) + require.NoError(t, err) + return synced } func TestCloudSinkEmitsTranslatableEvents(t *testing.T) { - transport := &mockTransport{} - - sink := newTestCloudSink(t, transport) + sink, _ := newTestCloudSink(t) defer func() { require.NoError(t, sink.Close()) }() @@ -62,9 +76,7 @@ func TestCloudSinkEmitsTranslatableEvents(t *testing.T) { } func TestCloudSinkSkipsUntranslatableEvents(t *testing.T) { - transport := &mockTransport{} - - sink := newTestCloudSink(t, transport) + sink, _ := newTestCloudSink(t) defer func() { require.NoError(t, sink.Close()) }() @@ -79,12 +91,7 @@ func TestCloudSinkSkipsUntranslatableEvents(t *testing.T) { } func TestCloudSinkEmitAndSync(t *testing.T) { - transport := &mockTransport{} - - sink := newTestCloudSink(t, transport) - defer func() { - require.NoError(t, sink.Close()) - }() + sink, walPath := newTestCloudSink(t) ctx := context.Background() err := sink.Handle(ctx, AuditEvent{ @@ -93,21 +100,17 @@ func TestCloudSinkEmitAndSync(t *testing.T) { Message: "blocked malware package", }) require.NoError(t, err) + require.NoError(t, sink.Close()) - synced, err := sink.syncClient.Sync(ctx) - require.NoError(t, err) + transport := &mockTransport{} + synced := drainWAL(t, walPath, transport) assert.Equal(t, 1, synced) assert.Equal(t, 1, len(transport.requests)) } func TestCloudSinkSetsInvocationContextOnSessionComplete(t *testing.T) { - transport := &mockTransport{} - - sink := newTestCloudSink(t, transport) - defer func() { - require.NoError(t, sink.Close()) - }() + sink, walPath := newTestCloudSink(t) ctx := context.Background() @@ -139,8 +142,10 @@ func TestCloudSinkSetsInvocationContextOnSessionComplete(t *testing.T) { }) require.NoError(t, err) - synced, err := sink.syncClient.Sync(ctx) - require.NoError(t, err) + require.NoError(t, sink.Close()) + + transport := &mockTransport{} + synced := drainWAL(t, walPath, transport) assert.Equal(t, 2, synced) require.Equal(t, 1, len(transport.requests)) diff --git a/internal/audit/cloud_translate_test.go b/internal/audit/cloud_translate_test.go index a096de7..536958d 100644 --- a/internal/audit/cloud_translate_test.go +++ b/internal/audit/cloud_translate_test.go @@ -10,8 +10,8 @@ import ( "github.com/stretchr/testify/require" ) -// testCloudSink returns a cloudSink with no real SyncClient, suitable for testing translation. -var testSink = &cloudSink{SyncClientBundle: &SyncClientBundle{}, invocationID: "test-invocation"} +// testSink is a cloudSink with no emitter, suitable for testing translation only. +var testSink = &cloudSink{invocationID: "test-invocation"} func TestTranslateMalwareBlocked(t *testing.T) { event := AuditEvent{