websub-hub/cmd/hubserver/handler.go

package main

import (
	"crypto/hmac"
	"crypto/sha1"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"time"

	"github.com/pkg/errors"
	"p83.nl/go/websub-hub/cmd/hubserver/storage"
)

// maxBody bounds how much of a fetched topic or verification response is read.
// A feed is kilobytes; anything larger is not a feed worth holding in memory.
const maxBody = 1 << 20

type subscriptionHandler struct {
	store   storage.Service
	baseURL string
	client  *http.Client
}

// handlePublish tells every subscriber of each named topic that its feed
// changed.
//
// The spec (§7.1) says a publisher sends hub.url, one or more of them. An
// older publisher names the feed hub.topic; that name belongs to
// subscribe/unsubscribe, so it is only read as a fallback. Reading hub.topic
// alone silently does nothing for a compliant publisher, which is why this
// used to answer 200 and deliver nothing.
func (handler *subscriptionHandler) handlePublish(w http.ResponseWriter, r *http.Request) error {
	topics := append([]string(nil), r.Form["hub.url"]...)
	if len(topics) == 0 {
		if t := r.Form.Get("hub.topic"); t != "" {
			topics = []string{t}
		}
	}
	if len(topics) == 0 {
		http.Error(w, "hub.url is required", http.StatusBadRequest)
		return nil
	}
	for _, topic := range topics {
		if _, ok := absoluteURL(topic); !ok {
			http.Error(w, fmt.Sprintf("hub.url is not an absolute http(s) url: %q", topic), http.StatusBadRequest)
			return nil
		}
	}

	if err := handler.store.RemoveExpiredSubscribers(); err != nil {
		log.Printf("publish: could not remove expired subscribers: %v", err)
		http.Error(w, "could not remove expired subscribers", http.StatusInternalServerError)
		return nil
	}

	for _, topic := range topics {
		log.Printf("publish: topic = %s\n", topic)
		if err := handler.publishTopic(topic); err != nil {
			log.Printf("publish: %v", err)
			http.Error(w, err.Error(), http.StatusBadGateway)
			return nil
		}
	}

	w.WriteHeader(http.StatusOK)
	return nil
}

// publishTopic fetches one topic and posts its body to every subscriber.
//
// The body is whatever the topic serves: a feed for an XML topic, and the
// HTML listing for a page topic. A page carrying microformats is a topic a
// Microsub server can consume, so the hub must not assume RSS.
func (handler *subscriptionHandler) publishTopic(topic string) error {
	res, err := handler.client.Get(topic)
	if err != nil {
		return errors.Wrapf(err, "could not request topic %s", topic)
	}
	defer res.Body.Close()

	feedContentType := res.Header.Get("Content-Type")
	feedContent, err := ioutil.ReadAll(io.LimitReader(res.Body, maxBody))
	if err != nil {
		return errors.Wrapf(err, "could not read topic %s", topic)
	}
	if res.StatusCode < 200 || res.StatusCode > 299 {
		// A topic that answers 404 or 500 has no feed to deliver; posting the
		// error page to subscribers puts garbage in their readers.
		return errors.Errorf("topic %s answered %d", topic, res.StatusCode)
	}

	subs, err := handler.store.Subscribers(topic)
	if err != nil {
		return errors.Wrapf(err, "could not get subscribers for %s", topic)
	}

	for _, sub := range subs {
		if err := handler.deliver(sub, topic, feedContentType, feedContent); err != nil {
			log.Printf("publish: %v", err)
		}
	}
	return nil
}

// deliver posts the body to one callback, signed when the subscriber
// registered a secret. It tries twice: a callback that is briefly down is the
// common case, and there is no stored queue to retry from later.
func (handler *subscriptionHandler) deliver(sub storage.Subscriber, topic, contentType string, body []byte) error {
	var lastErr error
	for attempt := 1; attempt <= 2; attempt++ {
		postReq, err := http.NewRequest("POST", sub.Callback, strings.NewReader(string(body)))
		if err != nil {
			return errors.Wrapf(err, "could not create request to %s", sub.Callback)
		}
		postReq.Header.Add("Content-Type", contentType)
		postReq.Header.Add("Link", fmt.Sprintf("<%s>; rel=hub, <%s>; rel=self", handler.baseURL, topic))
		if sub.Secret != "" {
			mac := hmac.New(sha1.New, []byte(sub.Secret))
			mac.Write(body)
			postReq.Header.Add("X-Hub-Signature", fmt.Sprintf("sha1=%x", mac.Sum(nil)))
		}

		postRes, err := handler.client.Do(postReq)
		if err != nil {
			lastErr = errors.Wrapf(err, "could not publish to %s", sub.Callback)
			continue
		}
		_, _ = io.Copy(io.Discard, io.LimitReader(postRes.Body, 4096))
		postRes.Body.Close()
		if postRes.StatusCode >= 200 && postRes.StatusCode <= 299 {
			log.Printf("publish: delivered to %s (%d)", sub.Callback, postRes.StatusCode)
			return nil
		}
		lastErr = errors.Errorf("callback %s answered %d", sub.Callback, postRes.StatusCode)
	}
	return lastErr
}

func (handler *subscriptionHandler) handleUnsubscription(w http.ResponseWriter, r *http.Request) error {
	callback := r.Form.Get("hub.callback")
	topic := r.Form.Get("hub.topic")
	mode := r.Form.Get("hub.mode")

	if _, ok := absoluteURL(callback); !ok {
		http.Error(w, "hub.callback is not an absolute http(s) url", http.StatusBadRequest)
		return nil
	}
	if _, ok := absoluteURL(topic); !ok {
		http.Error(w, "hub.topic is not an absolute http(s) url", http.StatusBadRequest)
		return nil
	}

	subscribers, err := handler.store.Subscribers(topic)
	if err != nil {
		log.Printf("unsubscribe: could not get subscribers for %s: %v", topic, err)
		http.Error(w, "could not read subscribers", http.StatusInternalServerError)
		return nil
	}

	found := false
	for _, subscriber := range subscribers {
		if subscriber.Callback != callback {
			continue
		}
		found = true

		ourChallenge := randStringBytes(24)
		if !validateURL(callback, mode, topic, ourChallenge) {
			continue
		}
		if err := handler.store.Unsubscribe(topic, callback); err != nil {
			log.Printf("unsubscribe: could not remove %s for %s: %v", callback, topic, err)
			http.Error(w, "could not unsubscribe", http.StatusInternalServerError)
			return nil
		}
	}

	// One response, whatever happened in the loop: writing 200 and then an
	// error body puts both in the same response and neither is true.
	if !found {
		http.Error(w, "Hub does not handle subscription for topic", http.StatusBadRequest)
		return nil
	}
	w.WriteHeader(http.StatusOK)
	_, _ = fmt.Fprint(w, "Unsubscribed\n")
	return nil
}

func (handler *subscriptionHandler) handleSubscription(w http.ResponseWriter, r *http.Request) error {
	log.Printf("subscription request received: %s %#v\n", r.URL.String(), r.Form)

	callback := r.Form.Get("hub.callback")
	topic := r.Form.Get("hub.topic")
	secret := r.Form.Get("hub.secret")
	leaseSecondsStr := r.Form.Get("hub.lease_seconds")

	leaseSeconds := int64(3600)
	if leaseSecondsStr != "" {
		parsed, err := strconv.ParseInt(leaseSecondsStr, 10, 64)
		if err != nil {
			http.Error(w, "hub.lease_seconds is used, but not a valid integer", http.StatusBadRequest)
			return nil
		}
		leaseSeconds = parsed
	} else {
		leaseSecondsStr = "3600"
	}

	callbackURL, ok := absoluteURL(callback)
	if !ok {
		http.Error(w, "cannot parse callback url", http.StatusBadRequest)
		return nil
	}
	topicURL, ok := absoluteURL(topic)
	if !ok {
		http.Error(w, "cannot parse topic url", http.StatusBadRequest)
		return nil
	}

	log.Println("subscribe: sending 202 header request accepted")
	w.WriteHeader(http.StatusAccepted)
	_, _ = fmt.Fprint(w, "Accepted\r\n")

	go func() {
		ourChallenge := randStringBytes(24)
		if !validateURL(callbackURL.String(), "subscribe", topicURL.String(), ourChallenge) {
			log.Println("subscribe: validation failed")
			return
		}
		log.Println("subscribe: validation valid")
		err := handler.store.Subscribe(topicURL.String(), storage.Subscriber{
			Callback:     callbackURL.String(),
			LeaseSeconds: leaseSeconds,
			Secret:       secret,
			Created:      time.Now(),
		})
		if err != nil {
			log.Printf("saving subscribe failed: %v", err)
		}
	}()
	return nil
}

// validateURL asks the callback to echo the challenge. The body must be the
// challenge and nothing else: a substring match accepts any page that merely
// mentions it, which is not the subscriber proving it received this request.
func validateURL(callback, mode, topic, challenge string) bool {
	u, err := url.Parse(callback)
	if err != nil {
		log.Printf("validateURL: %v", err)
		return false
	}
	q := u.Query()
	q.Set("hub.mode", mode)
	q.Set("hub.topic", topic)
	q.Set("hub.challenge", challenge)
	u.RawQuery = q.Encode()

	log.Println("validateURL: ", u.String())
	client := &http.Client{Timeout: 20 * time.Second}
	res, err := client.Get(u.String())
	if err != nil {
		log.Println("validateURL:", err)
		return false
	}
	defer res.Body.Close()

	body, err := ioutil.ReadAll(io.LimitReader(res.Body, maxBody))
	if err != nil {
		log.Println("validateURL:", err)
		return false
	}
	return strings.TrimSpace(string(body)) == challenge
}

func (handler *subscriptionHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if r.Header.Get("Content-Type") != "application/x-www-form-urlencoded" {
		http.Error(w, "Bad Request", http.StatusBadRequest)
		return
	}
	if err := r.ParseForm(); err != nil {
		http.Error(w, "Bad Request", http.StatusBadRequest)
		return
	}

	switch r.Form.Get("hub.mode") {
	case "subscribe":
		log.Println("hub.mode=subscribe")
		if err := handler.handleSubscription(w, r); err != nil {
			log.Printf("saving subscribe failed: %v", err)
		}
	case "unsubscribe":
		log.Println("hub.mode=unsubscribe")
		if err := handler.handleUnsubscription(w, r); err != nil {
			log.Printf("saving unsubscribe failed: %v", err)
		}
	case "publish":
		log.Println("hub.mode=publish received")
		if err := handler.handlePublish(w, r); err != nil {
			log.Printf("handling publish failed: %v", err)
		}
	default:
		http.Error(w, "Unknown hub.mode", http.StatusBadRequest)
	}
}

func absoluteURL(s string) (*url.URL, bool) {
	u, err := url.Parse(s)
	if err != nil || u.Host == "" {
		return nil, false
	}
	if u.Scheme != "http" && u.Scheme != "https" {
		return nil, false
	}
	return u, true
}