Assistant/adapters/notificationexecutor/executor.go

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"`
}

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
	secret         string
	client         *http.Client
	executeAllowed bool
}

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: &copyClient, 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.binding.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")
	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) {
		return ports.ExecutionResult{}
	}
	return ports.ExecutionResult{Known: true, Succeeded: true}
}

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) {
	base, err := url.Parse(e.binding.HTTPSOrigin)
	if err != nil {
		return "", err
	}
	base.Path = path + "/" + url.PathEscape(intent)
	base.RawPath, base.RawQuery, base.Fragment = "", "", ""
	return base.String(), nil
}

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
}