From 5abb2d8f481e4af59d1c2ef0fa5518edd27d067c Mon Sep 17 00:00:00 2001 From: rcourtman Date: Fri, 17 Jul 2026 17:36:17 +0100 Subject: [PATCH] refactor(agentexec): dedupe host operation dispatch and receipt-identity validation Contract-Neutral: dupl lint dedupe refactor; no behavior or contract delta (typed host-operation dispatch and receipt-identity validation extracted to shared generic helpers) --- internal/agentexec/apt_codec.go | 69 ++++++++----- internal/agentexec/server.go | 177 ++++++++++++++------------------ 2 files changed, 121 insertions(+), 125 deletions(-) diff --git a/internal/agentexec/apt_codec.go b/internal/agentexec/apt_codec.go index 144d421ca..ed7594fc3 100644 --- a/internal/agentexec/apt_codec.go +++ b/internal/agentexec/apt_codec.go @@ -343,24 +343,47 @@ func ValidateHostUpdateResultForRequest(req HostUpdatePayload, result HostUpdate // bearing results additionally prove the original before-state by deriving the // digest from their matching observation. func ValidateHostUpdateReceiptForIdentity(identity operationreceipt.Identity, result HostUpdateResultPayload) error { + return validateHostAPTReceiptForIdentity( + identity, HostUpdateOperationInstall, result.RequestID, result.ActionID, + func() error { return ValidateHostUpdateResultPayload(&result) }, + result.Verification == HostUpdateVerificationVerified || result.Verification == HostUpdateVerificationFailed, + func() error { + req := HostUpdatePayload{RequestID: identity.AttemptID, ActionID: identity.ActionID, Operation: identity.OperationKind, OperationVersion: identity.OperationVersion, RequestDigest: identity.RequestDigest, ExpectedInventoryHash: result.Before.InventoryHash} + if err := ValidateHostUpdatePayload(&req); err != nil { + return err + } + return ValidateHostUpdateResultForRequest(req, result) + }, + ) +} + +// validateHostAPTReceiptForIdentity is the shared receipt-binding core of the +// per-operation ReceiptForIdentity validators: bind the terminal result to the +// exact admitted identity, validate the result payload, and — only for +// evidence-bearing verifications — re-validate against a request reconstructed +// from the admitted identity. +func validateHostAPTReceiptForIdentity( + identity operationreceipt.Identity, + operationKind string, + resultRequestID, resultActionID string, + validateResultPayload func() error, + evidenceBearing bool, + validateEvidence func() error, +) error { normalized, err := operationreceipt.NormalizeIdentity(identity) if err != nil || normalized != identity { return operationreceipt.ErrBindingConflict } - if identity.OperationKind != HostUpdateOperationInstall || identity.OperationVersion != HostAPTOperationVersion || result.RequestID != identity.AttemptID || result.ActionID != identity.ActionID { + if identity.OperationKind != operationKind || identity.OperationVersion != HostAPTOperationVersion || resultRequestID != identity.AttemptID || resultActionID != identity.ActionID { return operationreceipt.ErrBindingConflict } - if err := ValidateHostUpdateResultPayload(&result); err != nil { + if err := validateResultPayload(); err != nil { return err } - if result.Verification != HostUpdateVerificationVerified && result.Verification != HostUpdateVerificationFailed { + if !evidenceBearing { return nil } - req := HostUpdatePayload{RequestID: identity.AttemptID, ActionID: identity.ActionID, Operation: identity.OperationKind, OperationVersion: identity.OperationVersion, RequestDigest: identity.RequestDigest, ExpectedInventoryHash: result.Before.InventoryHash} - if err := ValidateHostUpdatePayload(&req); err != nil { - return err - } - return ValidateHostUpdateResultForRequest(req, result) + return validateEvidence() } func ValidateHostUpdateResultForRequestAt(req HostUpdatePayload, result HostUpdateResultPayload, receivedAt time.Time) error { @@ -393,24 +416,18 @@ func ValidateHostStorageCleanupResultForRequest(req HostStorageCleanupPayload, r // inconclusive evidence and may report a different observed Before value while // the generic store continues to bind the receipt to the admitted digest. func ValidateHostStorageCleanupReceiptForIdentity(identity operationreceipt.Identity, result HostStorageCleanupResultPayload) error { - normalized, err := operationreceipt.NormalizeIdentity(identity) - if err != nil || normalized != identity { - return operationreceipt.ErrBindingConflict - } - if identity.OperationKind != HostStorageCleanupOperationPackageCache || identity.OperationVersion != HostAPTOperationVersion || result.RequestID != identity.AttemptID || result.ActionID != identity.ActionID { - return operationreceipt.ErrBindingConflict - } - if err := ValidateHostStorageCleanupResultPayload(&result); err != nil { - return err - } - if result.Verification != HostStorageCleanupVerificationVerified && result.Verification != HostStorageCleanupVerificationFailed { - return nil - } - req := HostStorageCleanupPayload{RequestID: identity.AttemptID, ActionID: identity.ActionID, Operation: identity.OperationKind, OperationVersion: identity.OperationVersion, RequestDigest: identity.RequestDigest, ExpectedFingerprint: result.Before.Fingerprint} - if err := ValidateHostStorageCleanupPayload(&req); err != nil { - return err - } - return ValidateHostStorageCleanupResultForRequest(req, result) + return validateHostAPTReceiptForIdentity( + identity, HostStorageCleanupOperationPackageCache, result.RequestID, result.ActionID, + func() error { return ValidateHostStorageCleanupResultPayload(&result) }, + result.Verification == HostStorageCleanupVerificationVerified || result.Verification == HostStorageCleanupVerificationFailed, + func() error { + req := HostStorageCleanupPayload{RequestID: identity.AttemptID, ActionID: identity.ActionID, Operation: identity.OperationKind, OperationVersion: identity.OperationVersion, RequestDigest: identity.RequestDigest, ExpectedFingerprint: result.Before.Fingerprint} + if err := ValidateHostStorageCleanupPayload(&req); err != nil { + return err + } + return ValidateHostStorageCleanupResultForRequest(req, result) + }, + ) } func ValidateHostStorageCleanupResultForRequestAt(req HostStorageCleanupPayload, result HostStorageCleanupResultPayload, receivedAt time.Time) error { diff --git a/internal/agentexec/server.go b/internal/agentexec/server.go index 8da97f9f9..714c751c8 100644 --- a/internal/agentexec/server.go +++ b/internal/agentexec/server.go @@ -1377,26 +1377,54 @@ func (s *Server) ExecuteCommand(ctx context.Context, agentID string, cmd Execute } } -// ExecuteHostUpdate dispatches the closed typed host-package operation. Unlike -// ExecuteCommand, no command text crosses this boundary; the agent owns the -// package-manager catalog, preflight, mutation, and read-after-write proof. -func (s *Server) ExecuteHostUpdate(ctx context.Context, agentID string, req HostUpdatePayload) (*HostUpdateResultPayload, error) { +// hostOperationPayload exposes the durable operation identity shared by the +// typed host APT operation request payloads. +type hostOperationPayload interface { + hostOperationIdentity() (requestID, actionID, operation string, timeoutSeconds int) +} + +func (p HostUpdatePayload) hostOperationIdentity() (string, string, string, int) { + return p.RequestID, p.ActionID, p.Operation, p.Timeout +} + +func (p HostStorageCleanupPayload) hostOperationIdentity() (string, string, string, int) { + return p.RequestID, p.ActionID, p.Operation, p.Timeout +} + +// hostOperationDispatch names the per-operation pieces of the shared typed +// host-operation dispatch cycle: claim → send → await validated receipt. +type hostOperationDispatch[Req hostOperationPayload, Res any] struct { + msgType MessageType + label string + pending map[string]chan Res + validateResult func(Req, Res, time.Time) error +} + +// prepareHostOperationRequest runs the shared request prologue of the typed +// host-operation dispatchers: normalize the agent id, default the request id, +// then bind and validate the payload. It returns the normalized agent id. +func prepareHostOperationRequest(s *Server, agentID string, requestID *string, bind func() error, validate func() error) (string, error) { if s == nil { - return nil, fmt.Errorf("agent execution server is unavailable") + return "", fmt.Errorf("agent execution server is unavailable") } agentID = strings.TrimSpace(agentID) if agentID == "" { - return nil, fmt.Errorf("agent id is required") + return "", fmt.Errorf("agent id is required") } - if strings.TrimSpace(req.RequestID) == "" { - req.RequestID = uuid.New().String() + if strings.TrimSpace(*requestID) == "" { + *requestID = uuid.New().String() } - if err := BindHostUpdatePayload(&req); err != nil { - return nil, err + if err := bind(); err != nil { + return "", err } - if err := ValidateHostUpdatePayload(&req); err != nil { - return nil, err + if err := validate(); err != nil { + return "", err } + return agentID, nil +} + +func dispatchHostOperation[Req hostOperationPayload, Res any](ctx context.Context, s *Server, agentID string, req Req, op hostOperationDispatch[Req, Res]) (*Res, error) { + requestID, actionID, operation, timeoutSeconds := req.hostOperationIdentity() s.mu.RLock() ac, ok := s.agents[agentID] @@ -1408,135 +1436,86 @@ func (s *Server) ExecuteHostUpdate(ctx context.Context, agentID string, req Host return nil, fmt.Errorf("agent does not support durable operation receipts") } - respCh := make(chan HostUpdateResultPayload, 1) - reqKey := pendingRequestKey(agentID, req.RequestID) - hostOperationKey, err := s.claimPendingHostOperation(agentID, req.RequestID, req.ActionID, req.Operation) + respCh := make(chan Res, 1) + reqKey := pendingRequestKey(agentID, requestID) + hostOperationKey, err := s.claimPendingHostOperation(agentID, requestID, actionID, operation) if err != nil { return nil, err } defer s.releasePendingHostOperation(hostOperationKey) s.mu.Lock() - if _, exists := s.pendingHostUpdates[reqKey]; exists { + if _, exists := op.pending[reqKey]; exists { s.mu.Unlock() - return nil, fmt.Errorf("host update request %q is already pending", req.RequestID) + return nil, fmt.Errorf("%s request %q is already pending", op.label, requestID) } - s.pendingHostUpdates[reqKey] = respCh + op.pending[reqKey] = respCh s.mu.Unlock() defer func() { s.mu.Lock() - delete(s.pendingHostUpdates, reqKey) + delete(op.pending, reqKey) s.mu.Unlock() }() - msg, err := NewMessage(MsgTypeHostUpdate, req.RequestID, req) + msg, err := NewMessage(op.msgType, requestID, req) if err != nil { - return nil, fmt.Errorf("failed to encode host update request: %w", err) + return nil, fmt.Errorf("failed to encode %s request: %w", op.label, err) } ac.writeMu.Lock() err = s.sendMessage(ac.conn, msg) ac.writeMu.Unlock() if err != nil { - return nil, fmt.Errorf("failed to send host update request: %w", err) + return nil, fmt.Errorf("failed to send %s request: %w", op.label, err) } - timer := time.NewTimer(time.Duration(req.Timeout) * time.Second) + timer := time.NewTimer(time.Duration(timeoutSeconds) * time.Second) defer timer.Stop() select { case result := <-respCh: - if err := ValidateHostUpdateResultForRequestAt(req, result, s.currentTime()); err != nil { - return nil, fmt.Errorf("host update result validation failed: %w", err) + if err := op.validateResult(req, result, s.currentTime()); err != nil { + return nil, fmt.Errorf("%s result validation failed: %w", op.label, err) } return &result, nil case <-timer.C: - return nil, fmt.Errorf("host update timed out after %s", time.Duration(req.Timeout)*time.Second) + return nil, fmt.Errorf("%s timed out after %s", op.label, time.Duration(timeoutSeconds)*time.Second) case <-ctx.Done(): return nil, ctx.Err() case <-ac.done: - return nil, fmt.Errorf("agent %s disconnected before host update receipt", agentID) + return nil, fmt.Errorf("agent %s disconnected before %s receipt", agentID, op.label) case <-s.shutdown: return nil, errServerShuttingDown } } +// ExecuteHostUpdate dispatches the closed typed host-package operation. Unlike +// ExecuteCommand, no command text crosses this boundary; the agent owns the +// package-manager catalog, preflight, mutation, and read-after-write proof. +func (s *Server) ExecuteHostUpdate(ctx context.Context, agentID string, req HostUpdatePayload) (*HostUpdateResultPayload, error) { + agentID, err := prepareHostOperationRequest(s, agentID, &req.RequestID, + func() error { return BindHostUpdatePayload(&req) }, + func() error { return ValidateHostUpdatePayload(&req) }) + if err != nil { + return nil, err + } + return dispatchHostOperation(ctx, s, agentID, req, hostOperationDispatch[HostUpdatePayload, HostUpdateResultPayload]{ + msgType: MsgTypeHostUpdate, label: "host update", + pending: s.pendingHostUpdates, validateResult: ValidateHostUpdateResultForRequestAt, + }) +} + // ExecuteHostStorageCleanup dispatches the closed package-cache cleanup // operation. No command text, path, package selector, or removal policy crosses // the server/agent boundary. func (s *Server) ExecuteHostStorageCleanup(ctx context.Context, agentID string, req HostStorageCleanupPayload) (*HostStorageCleanupResultPayload, error) { - if s == nil { - return nil, fmt.Errorf("agent execution server is unavailable") - } - agentID = strings.TrimSpace(agentID) - if agentID == "" { - return nil, fmt.Errorf("agent id is required") - } - if strings.TrimSpace(req.RequestID) == "" { - req.RequestID = uuid.New().String() - } - if err := BindHostStorageCleanupPayload(&req); err != nil { - return nil, err - } - if err := ValidateHostStorageCleanupPayload(&req); err != nil { - return nil, err - } - - s.mu.RLock() - ac, ok := s.agents[agentID] - s.mu.RUnlock() - if !ok { - return nil, fmt.Errorf("agent %s not connected", agentID) - } - if ac.agent.OperationReceiptVersion != operationreceipt.ProtocolVersion { - return nil, fmt.Errorf("agent does not support durable operation receipts") - } - - respCh := make(chan HostStorageCleanupResultPayload, 1) - reqKey := pendingRequestKey(agentID, req.RequestID) - hostOperationKey, err := s.claimPendingHostOperation(agentID, req.RequestID, req.ActionID, req.Operation) + agentID, err := prepareHostOperationRequest(s, agentID, &req.RequestID, + func() error { return BindHostStorageCleanupPayload(&req) }, + func() error { return ValidateHostStorageCleanupPayload(&req) }) if err != nil { return nil, err } - defer s.releasePendingHostOperation(hostOperationKey) - s.mu.Lock() - if _, exists := s.pendingHostStorageCleanups[reqKey]; exists { - s.mu.Unlock() - return nil, fmt.Errorf("host storage cleanup request %q is already pending", req.RequestID) - } - s.pendingHostStorageCleanups[reqKey] = respCh - s.mu.Unlock() - defer func() { - s.mu.Lock() - delete(s.pendingHostStorageCleanups, reqKey) - s.mu.Unlock() - }() - - msg, err := NewMessage(MsgTypeHostStorageCleanup, req.RequestID, req) - if err != nil { - return nil, fmt.Errorf("failed to encode host storage cleanup request: %w", err) - } - ac.writeMu.Lock() - err = s.sendMessage(ac.conn, msg) - ac.writeMu.Unlock() - if err != nil { - return nil, fmt.Errorf("failed to send host storage cleanup request: %w", err) - } - - timer := time.NewTimer(time.Duration(req.Timeout) * time.Second) - defer timer.Stop() - select { - case result := <-respCh: - if err := ValidateHostStorageCleanupResultForRequestAt(req, result, s.currentTime()); err != nil { - return nil, fmt.Errorf("host storage cleanup result validation failed: %w", err) - } - return &result, nil - case <-timer.C: - return nil, fmt.Errorf("host storage cleanup timed out after %s", time.Duration(req.Timeout)*time.Second) - case <-ctx.Done(): - return nil, ctx.Err() - case <-ac.done: - return nil, fmt.Errorf("agent %s disconnected before host storage cleanup receipt", agentID) - case <-s.shutdown: - return nil, errServerShuttingDown - } + return dispatchHostOperation(ctx, s, agentID, req, hostOperationDispatch[HostStorageCleanupPayload, HostStorageCleanupResultPayload]{ + msgType: MsgTypeHostStorageCleanup, label: "host storage cleanup", + pending: s.pendingHostStorageCleanups, validateResult: ValidateHostStorageCleanupResultForRequestAt, + }) } // ExecuteDockerContainerLifecycle dispatches one closed typed container