Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion action.yml
Original file line number Diff line number Diff line change
Expand Up @@ -283,7 +283,7 @@ runs:
# Stable per-repo identifier; the runner hostname is ephemeral
# so PMG's hostname fallback would create a new "machine" entry
# in SafeDep Cloud for every job.
export_var PMG_CLOUD_ENDPOINT_ID "github-actions/${GH_REPOSITORY}"
export_var PMG_CLOUD_ENDPOINT_ID "gha/${GH_REPOSITORY}"
fi
fi

Expand Down
48 changes: 30 additions & 18 deletions internal/audit/audit.go
Original file line number Diff line number Diff line change
Expand Up @@ -276,25 +276,37 @@ func LogSessionComplete(outcome Outcome, flowType FlowType) {

cfg := config.Get()

LogSessionSummary(SessionData{
PackageManager: s.packageManager,
FlowType: flowType,
Outcome: outcome,
TotalAnalyzed: s.totalAnalyzed,
AllowedCount: s.allowedCount,
BlockedCount: s.blockedCount,
ConfirmedCount: s.confirmedCount,
TrustedSkipped: s.trustedSkipped,
InsecureBypassed: s.insecureBypassed,
CooldownBlockedCount: s.cooldownBlockedCount,
Duration: time.Since(s.startTime),
SandboxEnabled: cfg.Config.Sandbox.Enabled,
ParanoidMode: cfg.Config.Paranoid,
TransitiveEnabled: cfg.Config.Transitive,
})
}

// LogSessionSummary emits a session-complete audit event from explicit session
// data. The persistent proxy daemon uses this because it aggregates run stats in
// a stats collector rather than the per-invocation audit session that
// LogSessionComplete reads from.
func LogSessionSummary(data SessionData) {
if global == nil {
return
}

logEvent(AuditEvent{
Type: EventTypeSessionComplete,
Message: fmt.Sprintf("Session complete: %s", outcome),
SessionData: &SessionData{
PackageManager: s.packageManager,
FlowType: flowType,
Outcome: outcome,
TotalAnalyzed: s.totalAnalyzed,
AllowedCount: s.allowedCount,
BlockedCount: s.blockedCount,
ConfirmedCount: s.confirmedCount,
TrustedSkipped: s.trustedSkipped,
InsecureBypassed: s.insecureBypassed,
CooldownBlockedCount: s.cooldownBlockedCount,
Duration: time.Since(s.startTime),
SandboxEnabled: cfg.Config.Sandbox.Enabled,
ParanoidMode: cfg.Config.Paranoid,
TransitiveEnabled: cfg.Config.Transitive,
},
Type: EventTypeSessionComplete,
Message: fmt.Sprintf("Session complete: %s", data.Outcome),
SessionData: &data,
})
}

Expand Down
31 changes: 31 additions & 0 deletions internal/audit/audit_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -273,6 +273,37 @@ func TestLogSessionCompleteSilentWhenNotInitialized(t *testing.T) {
LogSessionComplete(OutcomeSuccess, FlowTypeGuard)
}

func TestLogSessionSummaryDispatchesEvent(t *testing.T) {
s := &mockSink{}
setGlobal(newAuditor(s))
defer resetGlobal()

// The persistent proxy daemon serves every package manager, so the summary
// carries no single one.
LogSessionSummary(SessionData{
FlowType: FlowTypeProxy,
Outcome: OutcomeBlocked,
TotalAnalyzed: 3,
BlockedCount: 1,
AllowedCount: 2,
})

events := s.getEvents()
require.Len(t, events, 1)
assert.Equal(t, EventTypeSessionComplete, events[0].Type)
require.NotNil(t, events[0].SessionData)
assert.Empty(t, events[0].SessionData.PackageManager)
assert.Equal(t, FlowTypeProxy, events[0].SessionData.FlowType)
assert.Equal(t, OutcomeBlocked, events[0].SessionData.Outcome)
assert.Equal(t, uint32(1), events[0].SessionData.BlockedCount)
}

func TestLogSessionSummarySilentWhenNotInitialized(t *testing.T) {
resetGlobal()
// Should not panic
LogSessionSummary(SessionData{Outcome: OutcomeSuccess})
}

// TestUIOutcomesMappToAuditOutcomes ensures every ui.ExecutionOutcome has a
// corresponding audit.Outcome constant. If someone adds a new outcome to the
// UI layer without updating the audit package, this test will fail.
Expand Down
43 changes: 38 additions & 5 deletions internal/proxyserver/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,8 @@ func Run(ctx context.Context, cfg *config.RuntimeConfig, statePath, host string,
return fmt.Errorf("proxy already running (pid %d, addr %s) — run 'pmg proxy stop' first", existing.PID, existing.Addr)
}

startTime := time.Now()

caCertPath := certmanager.ProxyCABundlePath(cfg.ConfigDir())
caCert, _, err := flows.SetupCACertificate(cfg.ConfigDir(), caCertPath)
if err != nil {
Expand All @@ -91,12 +93,12 @@ func Run(ctx context.Context, cfg *config.RuntimeConfig, statePath, host string,
}

cache := interceptors.NewInMemoryAnalysisCache()
stats := interceptors.NewAnalysisStatsCollector()
statsCollector := interceptors.NewAnalysisStatsCollector()
confirmationChan := make(chan *interceptors.ConfirmationRequest, 100)
go autoBlockConfirmations(confirmationChan)

factory := interceptors.NewInterceptorFactory(
malysisAnalyzer, cache, stats, confirmationChan, interceptors.InterceptorContext{},
malysisAnalyzer, cache, statsCollector, confirmationChan, interceptors.InterceptorContext{},
)

var interceptorList []pmgproxy.Interceptor
Expand Down Expand Up @@ -158,15 +160,22 @@ func Run(ctx context.Context, cfg *config.RuntimeConfig, statePath, host string,

close(confirmationChan)

// Count is read after drain so a package analyzed at shutdown is not missed.
// Persist it BEFORE the (possibly slow) cloud flush so the blocked count
// Stats are read after drain so a package analyzed at shutdown is not missed.
// Persist the blocked count BEFORE the (possibly slow) cloud flush so it
// survives even if the flush hangs or the daemon is killed mid-flush, which
// keeps `stop --fail-on-violation` correct in those cases.
state.BlockedCount = stats.GetStats().BlockedCount
stats := statsCollector.GetStats()
state.BlockedCount = stats.BlockedCount
if werr := writeState(statePath, state); werr != nil {
log.Warnf("failed to write final proxy state: %v", werr)
}

// Emit the daemon-lifetime session summary before the final flush so it is
// delivered alongside the run's other events. Unlike the per-invocation flow,
// the daemon serves every package manager, so the summary carries no single
// package manager.
logSessionSummary(cfg, stats, time.Since(startTime))

// Halt the periodic sync (waits for any in-flight drain) before the final
// flush, so the two never hold the sync lock at once.
periodicSynced := stopSyncLoop()
Expand All @@ -181,6 +190,30 @@ func Run(ctx context.Context, cfg *config.RuntimeConfig, statePath, host string,
return stopErr
}

// logSessionSummary emits an aggregate session-complete audit event for the
// daemon's lifetime, mapping the proxy stats collector onto SessionData. The
// outcome is blocked when anything was blocked, otherwise success.
func logSessionSummary(cfg *config.RuntimeConfig, stats interceptors.AnalysisStats, duration time.Duration) {
outcome := audit.OutcomeSuccess
if stats.BlockedCount > 0 {
outcome = audit.OutcomeBlocked
}

audit.LogSessionSummary(audit.SessionData{
FlowType: audit.FlowTypeProxy,
Outcome: outcome,
TotalAnalyzed: uint32(stats.TotalAnalyzed),
AllowedCount: uint32(stats.AllowedCount),
BlockedCount: uint32(stats.BlockedCount),
ConfirmedCount: uint32(stats.ConfirmedCount),
CooldownBlockedCount: uint32(stats.CooldownBlockedCount),
Duration: duration,
SandboxEnabled: cfg.Config.Sandbox.Enabled,
ParanoidMode: cfg.Config.Paranoid,
TransitiveEnabled: cfg.Config.Transitive,
})
}

// cloudFlush drains whatever the periodic sync left and returns the outcome
// (total delivered, including periodicSynced). Returns nil when automatic cloud
// delivery is off (cloud or auto-sync disabled) — same gate as the periodic
Expand Down
Loading