Add a more complete webhook worker. (#424)

Adds a webhook worker which is scaffolded to support multiple event types.
Supports Checkin And Acknowledge events from the start.

cmd/micromdm/serve.go

@@ -102,7 +102,7 @@ func serve(args []string) error {
 		flRepoPath          = flagset.String("filerepo", "", "path to http file repo")
 		flDepSim            = flagset.String("depsim", "", "use depsim URL")
 		flExamples          = flagset.Bool("examples", false, "prints some example usage")
-		flCommandWebhookURL = flagset.String("command-webhook-url", "", "URL to send command responses as raw plists.")
+		flCommandWebhookURL = flagset.String("command-webhook-url", "", "URL to send command responses.")
 		flHomePage          = flagset.Bool("homepage", true, "hosts a simple built-in webpage at the / address")
 	)
 	flagset.Usage = usageFor(flagset, "micromdm serve [flags]")
@@ -163,7 +163,7 @@ func serve(args []string) error {
 	sm.setupSCEP(logger)
 	sm.setupPushService(logger)
 	sm.setupCommandService()
-	sm.setupWebhooks()
+	sm.setupWebhooks(logger)
 	sm.setupCommandQueue(logger)
 	sm.setupDepClient()
 	syncer := sm.setupDEPSync(logger)
@@ -209,8 +209,6 @@ func serve(args []string) error {
 		stdlog.Fatal(err)
 	}
 
-	sm.startWebhooks()
-
 	ctx := context.Background()
 	httpLogger := log.With(logger, "transport", "http")
 
@@ -395,7 +393,6 @@ type server struct {
 	commandService command.Service
 	configService  config.Service
 
-	responseWebhook    *webhook.CommandWebhook
 	webhooksHTTPClient *http.Client
 
 	err error
@@ -415,7 +412,7 @@ func (c *server) setupCommandService() {
 	c.commandService, c.err = command.New(c.pubclient)
 }
 
-func (c *server) setupWebhooks() {
+func (c *server) setupWebhooks(logger log.Logger) {
 	if c.err != nil {
 		return
 	}
@@ -424,23 +421,9 @@ func (c *server) setupWebhooks() {
 		return
 	}
 
-	h, err := webhook.NewCommandWebhook(c.webhooksHTTPClient, mdm.ConnectTopic, c.CommandWebhookURL)
-	if err != nil {
-		c.err = err
-		return
-	}
-
-	c.responseWebhook = h
-}
-
-func (c *server) startWebhooks() {
-	if c.err != nil {
-		return
-	}
-
-	if c.responseWebhook != nil {
-		c.responseWebhook.StartListener(c.pubclient)
-	}
+	ctx := context.Background()
+	ww := webhook.New(c.CommandWebhookURL, c.pubclient, webhook.WithLogger(logger), webhook.WithHTTPClient(c.webhooksHTTPClient))
+	go ww.Run(ctx)
 }
 
 func (c *server) setupRemoveService() {

workflow/webhook/acknowledge.go

@@ -0,0 +1,37 @@
+package webhook
+
+import (
+	"github.com/pkg/errors"
+
+	"github.com/micromdm/micromdm/mdm"
+)
+
+type AcknowledgeEvent struct {
+	UDID        string            `json:"udid"`
+	Status      string            `json:"status"`
+	CommandUUID string            `json:"command_uuid"`
+	Params      map[string]string `json:"url_params"`
+	RawPayload  []byte            `json:"raw_payload"`
+}
+
+func acknowledgeEvent(topic string, data []byte) (*Event, error) {
+	var ev mdm.AcknowledgeEvent
+	if err := mdm.UnmarshalAcknowledgeEvent(data, &ev); err != nil {
+		return nil, errors.Wrap(err, "unmarshal acknowledge event for webhook")
+	}
+	webhookEvent := Event{
+		Topic:     topic,
+		EventID:   ev.ID,
+		CreatedAt: ev.Time,
+
+		AcknowledgeEvent: &AcknowledgeEvent{
+			UDID:        ev.Response.UDID,
+			Status:      ev.Response.Status,
+			CommandUUID: ev.Response.CommandUUID,
+			Params:      ev.Params,
+			RawPayload:  ev.Raw,
+		},
+	}
+
+	return &webhookEvent, nil
+}

workflow/webhook/checkin.go

@@ -0,0 +1,34 @@
+package webhook
+
+import (
+	"github.com/pkg/errors"
+
+	"github.com/micromdm/micromdm/mdm"
+)
+
+type CheckinEvent struct {
+	UDID       string            `json:"udid"`
+	Params     map[string]string `json:"url_params"`
+	RawPayload []byte            `json:"raw_payload"`
+}
+
+func checkinEvent(topic string, data []byte) (*Event, error) {
+	var ev mdm.CheckinEvent
+	if err := mdm.UnmarshalCheckinEvent(data, &ev); err != nil {
+		return nil, errors.Wrap(err, "unmarshal checkin event for webhook")
+	}
+
+	webhookEvent := Event{
+		Topic:     topic,
+		EventID:   ev.ID,
+		CreatedAt: ev.Time,
+
+		CheckinEvent: &CheckinEvent{
+			UDID:       ev.Command.UDID,
+			Params:     ev.Params,
+			RawPayload: ev.Raw,
+		},
+	}
+
+	return &webhookEvent, nil
+}

workflow/webhook/command.go

@@ -1,60 +0,0 @@
-package webhook
-
-import (
-	"bytes"
-	"context"
-	"fmt"
-	"net/http"
-
-	"github.com/micromdm/micromdm/mdm"
-	"github.com/micromdm/micromdm/platform/pubsub"
-	"github.com/pkg/errors"
-)
-
-const contentType = "application/x-apple-aspen-mdm"
-
-type CommandWebhook struct {
-	Topic       string
-	CallbackURL string
-	HTTPClient  *http.Client
-}
-
-func NewCommandWebhook(httpClient *http.Client, topic, callbackURL string) (*CommandWebhook, error) {
-	if topic == "" {
-		return nil, errors.New("webhook: topic should not be empty")
-	}
-
-	if callbackURL == "" {
-		return nil, errors.New("webhook: callbackURL should not be empty")
-	}
-
-	return &CommandWebhook{HTTPClient: httpClient, Topic: topic, CallbackURL: callbackURL}, nil
-}
-
-func (cw CommandWebhook) StartListener(sub pubsub.Subscriber) error {
-	connectEvents, err := sub.Subscribe(context.TODO(), "commandWebhook", cw.Topic)
-	if err != nil {
-		return errors.Wrapf(err,
-			"subscribing commandWebhook to %s topic", cw.Topic)
-	}
-
-	go func() {
-		for {
-			select {
-			case event := <-connectEvents:
-				var ev mdm.AcknowledgeEvent
-				if err := mdm.UnmarshalAcknowledgeEvent(event.Message, &ev); err != nil {
-					fmt.Println(err)
-					continue
-				}
-
-				_, err := cw.HTTPClient.Post(cw.CallbackURL, contentType, bytes.NewBuffer(ev.Raw))
-				if err != nil {
-					fmt.Printf("error sending command response: %s\n", err)
-				}
-			}
-		}
-	}()
-
-	return nil
-}

workflow/webhook/http_post.go

@@ -0,0 +1,43 @@
+package webhook
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"net/http"
+
+	"github.com/pkg/errors"
+)
+
+type httpClient interface {
+	Do(*http.Request) (*http.Response, error)
+}
+
+func postWebhookEvent(
+	ctx context.Context,
+	client httpClient,
+	url string,
+	event interface{},
+) error {
+	raw, err := json.MarshalIndent(event, "", "  ")
+	if err != nil {
+		return errors.Wrap(err, "marshal webhook event")
+	}
+
+	req, err := http.NewRequest(http.MethodPost, url, bytes.NewBuffer(raw))
+	if err != nil {
+		return errors.Wrap(err, "create webhook http request")
+	}
+	req.Header.Set("Content-Type", "application/json; charset=utf-8")
+
+	resp, err := client.Do(req.WithContext(ctx))
+	if err != nil {
+		return errors.Wrap(err, "post webhook event to URL")
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode >= 400 {
+		return errors.Errorf("received unexpected HTTP status %d %s", resp.StatusCode, resp.Status)
+	}
+	return nil
+}

workflow/webhook/webhook.go

@@ -0,0 +1,119 @@
+package webhook
+
+import (
+	"context"
+	"net/http"
+	"time"
+
+	"github.com/go-kit/kit/log"
+	"github.com/go-kit/kit/log/level"
+	"github.com/pkg/errors"
+
+	"github.com/micromdm/micromdm/mdm"
+	"github.com/micromdm/micromdm/platform/pubsub"
+)
+
+type Event struct {
+	Topic     string    `json:"topic"`
+	EventID   string    `json:"event_id"`
+	CreatedAt time.Time `json:"created_at"`
+
+	AcknowledgeEvent *AcknowledgeEvent `json:"acknowledge_event,omitempty"`
+	CheckinEvent     *CheckinEvent     `json:"checkin_event,omitempty"`
+}
+
+type Worker struct {
+	logger log.Logger
+	url    string
+	client *http.Client
+	sub    pubsub.Subscriber
+}
+
+type Option func(*Worker)
+
+func WithLogger(logger log.Logger) Option {
+	return func(w *Worker) {
+		w.logger = logger
+	}
+}
+
+func WithHTTPClient(client *http.Client) Option {
+	return func(w *Worker) {
+		w.client = client
+	}
+}
+
+func New(url string, sub pubsub.Subscriber, opts ...Option) *Worker {
+	worker := &Worker{
+		url:    url,
+		sub:    sub,
+		logger: log.NewNopLogger(),
+		client: http.DefaultClient,
+	}
+
+	for _, optFn := range opts {
+		optFn(worker)
+	}
+
+	return worker
+}
+
+func (w *Worker) Run(ctx context.Context) error {
+	const subscription = "webhook_worker"
+
+	ackEvents, err := w.sub.Subscribe(ctx, subscription, mdm.ConnectTopic)
+	if err != nil {
+		return errors.Wrapf(err, "subscribe %s to %s", subscription, mdm.ConnectTopic)
+	}
+
+	authenticateEvents, err := w.sub.Subscribe(ctx, subscription, mdm.AuthenticateTopic)
+	if err != nil {
+		return errors.Wrapf(err, "subscribe %s to %s", subscription, mdm.AuthenticateTopic)
+	}
+
+	tokenUpdateEvents, err := w.sub.Subscribe(ctx, subscription, mdm.TokenUpdateTopic)
+	if err != nil {
+		return errors.Wrapf(err, "subscribe %s to %s", subscription, mdm.TokenUpdateTopic)
+	}
+
+	checkoutEvents, err := w.sub.Subscribe(ctx, subscription, mdm.CheckoutTopic)
+	if err != nil {
+		return errors.Wrapf(err, "subscribe %s to %s", subscription, mdm.CheckoutTopic)
+	}
+
+	for {
+		var (
+			event *Event
+			err   error
+		)
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case ev := <-ackEvents:
+			event, err = acknowledgeEvent(ev.Topic, ev.Message)
+		case ev := <-authenticateEvents:
+			event, err = checkinEvent(ev.Topic, ev.Message)
+		case ev := <-tokenUpdateEvents:
+			event, err = checkinEvent(ev.Topic, ev.Message)
+		case ev := <-checkoutEvents:
+			event, err = checkinEvent(ev.Topic, ev.Message)
+		}
+
+		if err != nil {
+			level.Info(w.logger).Log(
+				"msg", "create webhook event",
+				"err", err,
+			)
+			continue
+		}
+
+		if err := postWebhookEvent(ctx, w.client, w.url, event); err != nil {
+			level.Info(w.logger).Log(
+				"msg", "post webhook event",
+				"err", err,
+			)
+			continue
+		}
+
+	}
+}