package notificationexecutor import ( "bytes" "crypto/sha256" "encoding/hex" "encoding/json" "errors" "fmt" "io" "net/http" "net/url" "strings" "git.cin.su/m/Assistant/contracts/capability" commitcontract "git.cin.su/m/Assistant/contracts/commit" capabilitydomain "git.cin.su/m/Assistant/modules/capability/domain" "git.cin.su/m/Assistant/modules/commit/ports" ) const maxResponseBytes = 16 << 10 var ErrInvalidNotification = errors.New("invalid prospective notification") type Payload struct { ActionIntentID commitcontract.ActionIntentID `json:"action_intent_id"` TemplateCode string `json:"template_code"` ProspectiveItemID string `json:"prospective_item_id"` ExpectedRevision int64 `json:"expected_revision"` Title string `json:"title"` Message string `json:"message"` } type DeliveryRequest struct { Payload RequestDigest string `json:"request_digest"` } type Receipt struct { ReceiptID string `json:"receipt_id"` ActionIntentID commitcontract.ActionIntentID `json:"action_intent_id"` RequestDigest string `json:"request_digest"` AcceptedAt int64 `json:"accepted_at_unix_millis"` DeliveryMode string `json:"delivery_mode,omitempty"` ExternalDeliveryAttempted bool `json:"external_delivery_attempted"` } func Digest(payload Payload) (string, error) { if payload.ActionIntentID == "" || payload.TemplateCode != commitcontract.ProspectiveDeliveredTemplateV1 || payload.ProspectiveItemID == "" || payload.ExpectedRevision < 1 || strings.TrimSpace(payload.Title) == "" || strings.TrimSpace(payload.Message) == "" || len(payload.Title) > 160 || len(payload.Message) > 2048 { return "", ErrInvalidNotification } encoded, err := json.Marshal(payload) if err != nil { return "", err } sum := sha256.Sum256(encoded) return hex.EncodeToString(sum[:]), nil } func PayloadFromAction(action commitcontract.AuthorizedAction) (Payload, error) { subject := action.Proposal.NotificationSubject if action.Proposal.CapabilityID != commitcontract.CapabilityProspectiveNotification || action.Proposal.Operation != commitcontract.OperationDeliverProspectiveNotification || subject.BindingID == "" || subject.BindingRevision < 1 { return Payload{}, ErrInvalidNotification } payload := Payload{ActionIntentID: action.Proposal.IntentID, TemplateCode: subject.TemplateCode, ProspectiveItemID: string(subject.ProspectiveItemID), ExpectedRevision: int64(subject.ExpectedRevision), Title: subject.RenderedTitle, Message: subject.RenderedMessage} digest, err := Digest(payload) if err != nil || digest != subject.RequestDigest { return Payload{}, ErrInvalidNotification } return payload, nil } type Executor struct { binding capability.Binding candidate *capability.CandidateBinding secret string client *http.Client executeAllowed bool } // NewShadow accepts only a Capability-owned SHADOW binding. Its caller does // not supply an endpoint; the target remains inside the resolved binding. func NewShadow(binding capability.CandidateBinding, secret string, client *http.Client) (*Executor, error) { if capabilitydomain.ValidateCandidateBinding(binding) != nil || binding.Context.Mode != capability.CandidateBindingModeShadow || binding.Status != capability.CandidateBindingActive { return nil, ErrInvalidNotification } stable := capability.Binding{ID: binding.BaseActiveBindingID, Revision: binding.BaseActiveBindingRevision, OwnerPrincipalID: binding.Context.PrincipalID} value, err := newExecutor(stable, secret, client, true) if err != nil { return nil, err } value.candidate = &binding return value, nil } func New(binding capability.Binding, secret string, client *http.Client) (*Executor, error) { if err := capabilitydomain.UsableBinding(binding); err != nil { return nil, ErrInvalidNotification } return newExecutor(binding, secret, client, true) } func NewReconciler(binding capability.Binding, secret string, client *http.Client) (*Executor, error) { if err := capabilitydomain.ValidateBinding(binding); err != nil { return nil, ErrInvalidNotification } return newExecutor(binding, secret, client, false) } func newExecutor(binding capability.Binding, secret string, client *http.Client, executeAllowed bool) (*Executor, error) { if strings.TrimSpace(secret) == "" || client == nil { return nil, ErrInvalidNotification } copyClient := *client copyClient.CheckRedirect = func(_ *http.Request, _ []*http.Request) error { return http.ErrUseLastResponse } return &Executor{binding: binding, secret: secret, client: ©Client, executeAllowed: executeAllowed}, nil } func (e *Executor) Execute(action commitcontract.AuthorizedAction) ports.ExecutionResult { if !e.executeAllowed { return ports.ExecutionResult{Known: true} } payload, err := e.requestFor(action) if err != nil { return ports.ExecutionResult{Known: true} } body, err := json.Marshal(DeliveryRequest{Payload: payload, RequestDigest: action.Proposal.NotificationSubject.RequestDigest}) if err != nil { return ports.ExecutionResult{Known: true} } endpoint, err := e.endpoint(e.deliveryPath(), string(action.Proposal.IntentID)) if err != nil { return ports.ExecutionResult{Known: true} } request, err := http.NewRequest(http.MethodPut, endpoint, bytes.NewReader(body)) if err != nil { return ports.ExecutionResult{Known: true} } request.Header.Set("Authorization", "Bearer "+e.secret) request.Header.Set("Content-Type", "application/json") request.Header.Set("Accept", "application/json") if e.candidate != nil { request.Header.Set("X-Candidate-Release", e.candidate.Context.ReleaseSetID) request.Header.Set("X-Candidate-Task", e.candidate.Context.TaskID) } response, err := e.client.Do(request) if err != nil { return ports.ExecutionResult{} } defer response.Body.Close() if response.StatusCode != http.StatusOK && response.StatusCode != http.StatusCreated { if response.StatusCode >= 400 && response.StatusCode < 500 && response.StatusCode != http.StatusRequestTimeout && response.StatusCode != http.StatusTooManyRequests { return ports.ExecutionResult{Known: true} } return ports.ExecutionResult{} } receipt, err := decodeReceipt(response.Body) if err != nil || !matchingReceipt(receipt, action) || (e.candidate != nil && (receipt.DeliveryMode != "SHADOW" || receipt.ExternalDeliveryAttempted)) { return ports.ExecutionResult{} } result := ports.ExecutionResult{Known: true, Succeeded: true} if e.candidate != nil { result.Receipt = &commitcontract.ExecutionReceipt{ID: receipt.ReceiptID, ExecutorImplementationRef: "notification-shadow-receiver-v1", ExternalOperationRef: "SHADOW_CAPTURED", ExternalRequestID: string(receipt.ActionIntentID), RequestDigest: receipt.RequestDigest, DispatchAtUnixMillis: receipt.AcceptedAt, ExecutorResult: commitcontract.ExecutorAccepted, ReconciliationKey: receipt.ReceiptID, DeliveryMode: "SHADOW", ExternalDeliveryAttempted: false, ReleaseSetID: e.candidate.Context.ReleaseSetID, TaskID: e.candidate.Context.TaskID, TaskRevision: e.candidate.Context.TaskRevision, CapabilityBindingID: string(e.candidate.ID), CapabilityBindingRevision: e.candidate.Revision, TargetReference: e.candidate.Context.TargetRef} } return result } func (e *Executor) Reconcile(action commitcontract.AuthorizedAction) ports.ExecutionResult { if _, err := e.requestFor(action); err != nil { return ports.ExecutionResult{Known: true} } endpoint, err := e.endpoint(e.binding.ReceiptPath, string(action.Proposal.IntentID)) if err != nil { return ports.ExecutionResult{Known: true} } request, err := http.NewRequest(http.MethodGet, endpoint, nil) if err != nil { return ports.ExecutionResult{Known: true} } request.Header.Set("Authorization", "Bearer "+e.secret) request.Header.Set("Accept", "application/json") response, err := e.client.Do(request) if err != nil { return ports.ExecutionResult{} } defer response.Body.Close() if response.StatusCode != http.StatusOK { return ports.ExecutionResult{} } receipt, err := decodeReceipt(response.Body) if err != nil { return ports.ExecutionResult{} } if !matchingReceipt(receipt, action) { return ports.ExecutionResult{Known: true} } return ports.ExecutionResult{Known: true, Succeeded: true} } func (e *Executor) requestFor(action commitcontract.AuthorizedAction) (Payload, error) { if action.Proposal.NotificationSubject.BindingID != e.binding.ID || action.Proposal.NotificationSubject.BindingRevision != e.binding.Revision || action.PrincipalID != e.binding.OwnerPrincipalID { return Payload{}, ErrInvalidNotification } return PayloadFromAction(action) } func (e *Executor) endpoint(path, intent string) (string, error) { origin := e.binding.HTTPSOrigin if e.candidate != nil { origin = e.candidate.TargetOrigin } base, err := url.Parse(origin) if err != nil { return "", err } base.Path = path + "/" + url.PathEscape(intent) base.RawPath, base.RawQuery, base.Fragment = "", "", "" return base.String(), nil } func (e *Executor) deliveryPath() string { if e.candidate != nil { return e.candidate.DeliveryPath } return e.binding.DeliveryPath } func decodeReceipt(reader io.Reader) (Receipt, error) { var receipt Receipt decoder := json.NewDecoder(io.LimitReader(reader, maxResponseBytes+1)) if err := decoder.Decode(&receipt); err != nil { return Receipt{}, err } if receipt.ReceiptID == "" || receipt.ActionIntentID == "" || len(receipt.RequestDigest) != 64 || receipt.AcceptedAt <= 0 { return Receipt{}, fmt.Errorf("invalid receipt") } return receipt, nil } func matchingReceipt(receipt Receipt, action commitcontract.AuthorizedAction) bool { return receipt.ActionIntentID == action.Proposal.IntentID && receipt.RequestDigest == action.Proposal.NotificationSubject.RequestDigest }