diff --git a/.drone.yml b/.drone.yml
index 36dc6b9..add11ae 100644
--- a/.drone.yml
+++ b/.drone.yml
@@ -2,19 +2,14 @@ kind: pipeline
type: docker
name: default
-workspace:
- base: /go
- path: src/p83.nl/go/websub-hub
-
steps:
- name: testing
- image: golang:1.12-alpine
+ # Module build: the go.mod pins the dependencies, so there is no go get and
+ # no GOPATH layout to maintain. 1.12 could not compile io.Discard.
+ image: golang:1.22-alpine
commands:
- - export GOPATH=/go
- export CGO_ENABLED=0
- - apk --no-cache add git
- - go get -d -t ./...
- - go build p83.nl/go/websub-hub/cmd/hubserver
+ - go build ./...
- go test ./...
- name: publish
diff --git a/cmd/hubserver/handler.go b/cmd/hubserver/handler.go
index 9fa2eea..4e8bbe8 100644
--- a/cmd/hubserver/handler.go
+++ b/cmd/hubserver/handler.go
@@ -4,11 +4,11 @@ import (
"crypto/hmac"
"crypto/sha1"
"fmt"
+ "io"
"io/ioutil"
"log"
"net/http"
"net/url"
- "os"
"strconv"
"strings"
"time"
@@ -17,75 +17,129 @@ import (
"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 {
- topic := r.Form.Get("hub.topic")
- log.Printf("publish: topic = %s\n", topic)
-
- client := &http.Client{}
- req, err := http.NewRequest("GET", topic, nil)
- if err != nil {
- return errors.Wrap(err, "could not create request")
+ topics := append([]string(nil), r.Form["hub.url"]...)
+ if len(topics) == 0 {
+ if t := r.Form.Get("hub.topic"); t != "" {
+ topics = []string{t}
+ }
}
- req.Header.Add("Accept", "*/*")
- res, err := client.Do(req)
+ 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(res.Body)
+ feedContent, err := ioutil.ReadAll(io.LimitReader(res.Body, maxBody))
if err != nil {
- return errors.Wrap(err, "could not read body")
+ return errors.Wrapf(err, "could not read topic %s", topic)
}
-
- err = handler.store.RemoveExpiredSubscribers()
- if err != nil {
- return errors.Wrap(err, "could not remove expired subscribers")
+ 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.Wrap(err, "could not get subscribers")
+ return errors.Wrapf(err, "could not get subscribers for %s", topic)
}
for _, sub := range subs {
- log.Printf("publish: preparing post to %s\n", sub.Callback)
- postReq, err := http.NewRequest("POST", sub.Callback, strings.NewReader(string(feedContent)))
- if err != nil {
- log.Printf("could not creating request to %s: %s", sub.Callback, err)
- continue
+ if err := handler.deliver(sub, topic, feedContentType, feedContent); err != nil {
+ log.Printf("publish: %v", err)
}
- postReq.Header.Add("Content-Type", feedContentType)
- postReq.Header.Add("Link",
- fmt.Sprintf(
- "<%s>; rel=hub, <%s>; rel=self",
- handler.baseURL,
- topic,
- ))
+ }
+ 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(feedContent)
- signature := mac.Sum(nil)
- postReq.Header.Add("X-Hub-Signature", fmt.Sprintf("sha1=%x", signature))
+ mac.Write(body)
+ postReq.Header.Add("X-Hub-Signature", fmt.Sprintf("sha1=%x", mac.Sum(nil)))
}
- log.Printf("publish: send post to %s\n", sub.Callback)
- postRes, err := client.Do(postReq)
+
+ postRes, err := handler.client.Do(postReq)
if err != nil {
- log.Printf("could not publish to %s: %s", sub.Callback, err)
+ lastErr = errors.Wrapf(err, "could not publish to %s", sub.Callback)
continue
}
- log.Printf("publish: post sent to %s\n", sub.Callback)
- log.Println("Response:")
- _ = postRes.Write(os.Stdout)
+ _, _ = 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 nil
+ return lastErr
}
func (handler *subscriptionHandler) handleUnsubscription(w http.ResponseWriter, r *http.Request) error {
@@ -93,9 +147,20 @@ func (handler *subscriptionHandler) handleUnsubscription(w http.ResponseWriter,
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 {
- return errors.Wrapf(err, "could not get subscribers for %s", topic)
+ 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
@@ -103,37 +168,28 @@ func (handler *subscriptionHandler) handleUnsubscription(w http.ResponseWriter,
if subscriber.Callback != callback {
continue
}
-
found = true
- ourChallenge := randStringBytes(12)
-
- validationURL, err := url.Parse(callback)
- if err != nil {
- log.Println(err)
- return err
+ ourChallenge := randStringBytes(24)
+ if !validateURL(callback, mode, topic, ourChallenge) {
+ continue
}
- q := validationURL.Query()
- q.Add("hub.mode", mode)
- q.Add("hub.topic", topic)
- q.Add("hub.challenge", ourChallenge)
- validationURL.RawQuery = q.Encode()
- if validateURL(validationURL.String(), ourChallenge) {
- err = handler.store.Unsubscribe(topic, callback)
- if err != nil {
- return err
- }
+ 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
}
}
- w.WriteHeader(200)
- if found {
- _, err = fmt.Fprintf(w, "Unsubscribed\n")
- } else {
- http.Error(w, "Hub does not handle subscription for topic", 400)
+ // 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
}
-
- return err
+ w.WriteHeader(http.StatusOK)
+ _, _ = fmt.Fprint(w, "Unsubscribed\n")
+ return nil
}
func (handler *subscriptionHandler) handleSubscription(w http.ResponseWriter, r *http.Request) error {
@@ -143,138 +199,124 @@ func (handler *subscriptionHandler) handleSubscription(w http.ResponseWriter, r
topic := r.Form.Get("hub.topic")
secret := r.Form.Get("hub.secret")
leaseSecondsStr := r.Form.Get("hub.lease_seconds")
- leaseSeconds, err := strconv.ParseInt(leaseSecondsStr, 10, 64)
- if leaseSecondsStr != "" && err != nil {
- http.Error(w, "hub.lease_seconds is used, but not a valid integer", 400)
- log.Printf("hub.lease_seconds is used, but not a valid integer (%s)\n", leaseSecondsStr)
- return err
- }
- log.Printf("subscribe: received for topic=%s to callback=%s (lease=%ds)\n", topic, callback, leaseSeconds)
-
- if _, e := r.Form["hub.lease_seconds"]; !e {
- leaseSeconds = 3600
+ 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"
- log.Printf("subscribe: lease_seconds was empty use default %ds\n", leaseSeconds)
}
- callbackURL, err := url.Parse(callback)
- if callback == "" || err != nil {
- http.Error(w, "cannot parse callback url", 400)
- log.Printf("cannot parse callback url: %s", callback)
- return err
+ callbackURL, ok := absoluteURL(callback)
+ if !ok {
+ http.Error(w, "cannot parse callback url", http.StatusBadRequest)
+ return nil
}
- if !callbackURL.IsAbs() {
- http.Error(w, "callback url is not absolute", 400)
- return fmt.Errorf("callback url is not absolute: %s", callback)
- }
-
- topicURL, err := url.Parse(topic)
- if topic == "" || err != nil {
- http.Error(w, "cannot parse topic url", 400)
- return fmt.Errorf("cannot parse topic url: %s\n", topic)
- }
- if !topicURL.IsAbs() {
- http.Error(w, "topic url is not absolute", 400)
- return fmt.Errorf("topic url is not absolute: %s", topic)
+ 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(202)
+ w.WriteHeader(http.StatusAccepted)
_, _ = fmt.Fprint(w, "Accepted\r\n")
go func() {
- ourChallenge := randStringBytes(12)
-
- validationURL := *callbackURL
- q := validationURL.Query()
- q.Add("hub.mode", "subscribe")
- q.Add("hub.topic", topicURL.String())
- q.Add("hub.challenge", ourChallenge)
- q.Add("hub.lease_seconds", leaseSecondsStr)
- if secret != "" {
- q.Add("hub.verify_token", secret)
- }
- validationURL.RawQuery = q.Encode()
-
- log.Printf("subscribe: async validation with url %s\n", validationURL.String())
-
- if validateURL(validationURL.String(), ourChallenge) {
- log.Println("subscribe: validation valid")
- err = handler.store.Subscribe(topicURL.String(), storage.Subscriber{callbackURL.String(), leaseSeconds, secret, time.Now()})
- if err != nil {
- log.Printf("saving subscribe failed: %v", err)
- }
- } else {
+ 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
}
-func validateURL(url, challenge string) bool {
- log.Println("validateURL: ", url)
- client := http.Client{}
-
- req, err := http.NewRequest(http.MethodGet, url, 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.Println(err)
+ log.Printf("validateURL: %v", err)
return false
}
- res, err := client.Do(req)
+ 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(err)
+ log.Println("validateURL:", err)
return false
}
defer res.Body.Close()
- body, err := ioutil.ReadAll(res.Body)
+ body, err := ioutil.ReadAll(io.LimitReader(res.Body, maxBody))
if err != nil {
- log.Println(err)
+ log.Println("validateURL:", err)
return false
}
-
- return strings.Contains(string(body), challenge)
+ 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", 400)
+ http.Error(w, "Bad Request", http.StatusBadRequest)
+ return
+ }
+ if err := r.ParseForm(); err != nil {
+ http.Error(w, "Bad Request", http.StatusBadRequest)
return
}
- err := r.ParseForm()
- if err != nil {
- http.Error(w, "Bad Request", 400)
- return
- }
-
- mode := r.Form.Get("hub.mode")
-
- if mode == "subscribe" {
+ switch r.Form.Get("hub.mode") {
+ case "subscribe":
log.Println("hub.mode=subscribe")
- err = handler.handleSubscription(w, r)
- if err != nil {
+ if err := handler.handleSubscription(w, r); err != nil {
log.Printf("saving subscribe failed: %v", err)
}
- return
- } else if mode == "unsubscribe" {
+ case "unsubscribe":
log.Println("hub.mode=unsubscribe")
- err = handler.handleUnsubscription(w, r)
- if err != nil {
+ if err := handler.handleUnsubscription(w, r); err != nil {
log.Printf("saving unsubscribe failed: %v", err)
}
- return
- } else if mode == "publish" {
+ case "publish":
log.Println("hub.mode=publish received")
- err = handler.handlePublish(w, r)
- if err != nil {
+ if err := handler.handlePublish(w, r); err != nil {
log.Printf("handling publish failed: %v", err)
}
- return
- } else {
- http.Error(w, "Unknown hub.mode", 400)
- return
+ 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
+}
diff --git a/cmd/hubserver/handler_test.go b/cmd/hubserver/handler_test.go
new file mode 100644
index 0000000..c23cf99
--- /dev/null
+++ b/cmd/hubserver/handler_test.go
@@ -0,0 +1,215 @@
+package main
+
+import (
+ "fmt"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "net/url"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "p83.nl/go/websub-hub/cmd/hubserver/storage"
+)
+
+type fakeStore struct {
+ mu sync.Mutex
+ subs map[string][]storage.Subscriber
+}
+
+func newFakeStore() *fakeStore { return &fakeStore{subs: map[string][]storage.Subscriber{}} }
+
+func (f *fakeStore) Subscribe(topic string, sub storage.Subscriber) error {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ f.subs[topic] = append(f.subs[topic], sub)
+ return nil
+}
+
+func (f *fakeStore) Unsubscribe(topic, callback string) error {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ var keep []storage.Subscriber
+ for _, s := range f.subs[topic] {
+ if s.Callback != callback {
+ keep = append(keep, s)
+ }
+ }
+ f.subs[topic] = keep
+ return nil
+}
+
+func (f *fakeStore) Subscribers(topic string) ([]storage.Subscriber, error) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return append([]storage.Subscriber(nil), f.subs[topic]...), nil
+}
+
+func (f *fakeStore) RemoveExpiredSubscribers() error { return nil }
+func (f *fakeStore) Close() error { return nil }
+
+// recordings holds the delivery POSTs a fake callback received.
+type recordings struct {
+ mu sync.Mutex
+ posts []map[string]string
+}
+
+func (r *recordings) add(h http.Header, body string) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ r.posts = append(r.posts, map[string]string{"link": h.Get("Link"), "body": body})
+}
+
+func (r *recordings) got() []map[string]string {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ return append([]map[string]string(nil), r.posts...)
+}
+
+func newHandler(t *testing.T) (*subscriptionHandler, *fakeStore) {
+ t.Helper()
+ store := newFakeStore()
+ return &subscriptionHandler{
+ store: store,
+ baseURL: "https://hub.test/",
+ client: &http.Client{Timeout: 5 * time.Second},
+ }, store
+}
+
+func post(t *testing.T, h *subscriptionHandler, form url.Values) *httptest.ResponseRecorder {
+ t.Helper()
+ req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(form.Encode()))
+ req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+ rec := httptest.NewRecorder()
+ h.ServeHTTP(rec, req)
+ return rec
+}
+
+// A publish names the feed in hub.url (spec §7.1) and the hub delivers the
+// topic's body to every subscriber.
+func TestPublishUsesHubURL(t *testing.T) {
+ h, store := newHandler(t)
+
+ topic := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("Content-Type", "application/rss+xml")
+ fmt.Fprint(w, "fresh")
+ }))
+ defer topic.Close()
+
+ var rec recordings
+ callback := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ body, _ := io.ReadAll(r.Body)
+ rec.add(r.Header, string(body))
+ w.WriteHeader(http.StatusNoContent)
+ }))
+ defer callback.Close()
+
+ if err := store.Subscribe(topic.URL, storage.Subscriber{Callback: callback.URL, LeaseSeconds: 3600, Secret: "s"}); err != nil {
+ t.Fatal(err)
+ }
+
+ res := post(t, h, url.Values{"hub.mode": {"publish"}, "hub.url": {topic.URL}})
+ if res.Code != http.StatusOK {
+ t.Fatalf("publish = %d %s", res.Code, res.Body.String())
+ }
+ posts := rec.got()
+ if len(posts) != 1 {
+ t.Fatalf("callback saw %d posts, want 1", len(posts))
+ }
+ if posts[0]["body"] != "fresh" {
+ t.Fatalf("body = %q", posts[0]["body"])
+ }
+ if !strings.Contains(posts[0]["link"], `rel=hub`) || !strings.Contains(posts[0]["link"], `rel=self`) {
+ t.Fatalf("Link header = %q", posts[0]["link"])
+ }
+}
+
+// An older publisher names the topic hub.topic; that must still publish.
+func TestPublishFallsBackToHubTopic(t *testing.T) {
+ h, store := newHandler(t)
+
+ topic := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ fmt.Fprint(w, "feed")
+ }))
+ defer topic.Close()
+
+ var rec recordings
+ callback := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ rec.add(r.Header, "")
+ w.WriteHeader(http.StatusNoContent)
+ }))
+ defer callback.Close()
+
+ _ = store.Subscribe(topic.URL, storage.Subscriber{Callback: callback.URL, LeaseSeconds: 3600})
+
+ res := post(t, h, url.Values{"hub.mode": {"publish"}, "hub.topic": {topic.URL}})
+ if res.Code != http.StatusOK {
+ t.Fatalf("publish = %d", res.Code)
+ }
+ if len(rec.got()) != 1 {
+ t.Fatalf("callback saw %d posts, want 1", len(rec.got()))
+ }
+}
+
+// A publish with neither name is refused instead of answering 200 and doing
+// nothing, which is how this endpoint used to lose every notification.
+func TestPublishWithoutTopicIsRefused(t *testing.T) {
+ h, _ := newHandler(t)
+
+ res := post(t, h, url.Values{"hub.mode": {"publish"}})
+ if res.Code != http.StatusBadRequest {
+ t.Fatalf("publish = %d, want 400", res.Code)
+ }
+ if !strings.Contains(res.Body.String(), "hub.url") {
+ t.Fatalf("body = %q", res.Body.String())
+ }
+}
+
+// A topic that answers an error has no feed to deliver.
+func TestPublishSkipsFailedTopic(t *testing.T) {
+ h, store := newHandler(t)
+
+ topic := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ http.NotFound(w, r)
+ }))
+ defer topic.Close()
+
+ var rec recordings
+ callback := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ rec.add(r.Header, "")
+ w.WriteHeader(http.StatusNoContent)
+ }))
+ defer callback.Close()
+
+ _ = store.Subscribe(topic.URL, storage.Subscriber{Callback: callback.URL, LeaseSeconds: 3600})
+
+ res := post(t, h, url.Values{"hub.mode": {"publish"}, "hub.url": {topic.URL}})
+ if res.Code != http.StatusBadGateway {
+ t.Fatalf("publish = %d, want 502", res.Code)
+ }
+ if len(rec.got()) != 0 {
+ t.Fatalf("callback received an error page: %v", rec.got())
+ }
+}
+
+// Verification requires the subscriber to echo the challenge, not to mention
+// it somewhere on a page.
+func TestValidateURLRequiresExactChallenge(t *testing.T) {
+ challenge := "abc123"
+ var reply string
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ fmt.Fprint(w, reply)
+ }))
+ defer srv.Close()
+
+ reply = " " + challenge + "\n"
+ if !validateURL(srv.URL, "subscribe", "https://example.test/feed.xml", challenge) {
+ t.Fatal("exact echo rejected")
+ }
+ reply = "thanks for subscribing, the challenge was " + challenge + " but this is a page"
+ if validateURL(srv.URL, "subscribe", "https://example.test/feed.xml", challenge) {
+ t.Fatal("substring match accepted")
+ }
+}
diff --git a/cmd/hubserver/main.go b/cmd/hubserver/main.go
index 0ed82eb..cfb9269 100644
--- a/cmd/hubserver/main.go
+++ b/cmd/hubserver/main.go
@@ -1,11 +1,12 @@
package main
import (
+ "crypto/rand"
"flag"
"fmt"
"log"
- "math/rand"
"net/http"
+ "os"
"time"
"p83.nl/go/websub-hub/cmd/hubserver/storage"
@@ -15,14 +16,21 @@ func init() {
log.SetFlags(log.Ldate | log.Ltime | log.Lshortfile)
}
-const letterBytes = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
+const challengeBytes = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
+// randStringBytes is the hub's challenge. It comes from crypto/rand: a
+// challenge a subscriber has to echo is a secret, and math/rand makes it
+// predictable for anyone who observes a few of them.
func randStringBytes(n int) string {
- b := make([]byte, n)
- for i := range b {
- b[i] = letterBytes[rand.Intn(len(letterBytes))]
+ buf := make([]byte, n)
+ if _, err := rand.Read(buf); err != nil {
+ log.Fatalf("could not read random bytes: %v", err)
}
- return string(b)
+ out := make([]byte, n)
+ for i, b := range buf {
+ out[i] = challengeBytes[int(b)%len(challengeBytes)]
+ }
+ return string(out)
}
type Stat struct {
@@ -34,18 +42,36 @@ func main() {
hostPort := flag.String("http", ":80", "host and port to listen on")
baseURL := flag.String("baseurl", "", "baseurl that the server should response with")
database := flag.String("database", "localhost:9999", "database host:port")
+ dbUser := flag.String("database-user", envOr("HUB_DB_USER", "postgres"), "database user")
+ dbPassword := flag.String("database-password", envOr("HUB_DB_PASSWORD", "simple"), "database password")
flag.Parse()
- log.Printf("Using arguments http=%s baseurl=%s database=%s", *hostPort, *baseURL, *database)
+ log.Printf("Using arguments http=%s baseurl=%s database=%s user=%s", *hostPort, *baseURL, *database, *dbUser)
+ if os.Getenv("HUB_DB_PASSWORD") == "" {
+ log.Printf("warning: HUB_DB_PASSWORD is unset, using the built-in default; set it in the deployment")
+ }
- dsn := fmt.Sprintf("postgres://%v:%v@%s/?sslmode=disable", "postgres", "simple", *database)
+ dsn := fmt.Sprintf("postgres://%s:%s@%s/?sslmode=disable", *dbUser, *dbPassword, *database)
store, err := storage.New(dsn)
if err != nil {
log.Fatal(err)
}
defer store.Close()
- handler := &subscriptionHandler{store, *baseURL}
+ handler := &subscriptionHandler{
+ store: store,
+ baseURL: *baseURL,
+ // One client with a timeout: a topic or callback that hangs must not
+ // hold the hub forever.
+ client: &http.Client{Timeout: 30 * time.Second},
+ }
http.Handle("/", handler)
- log.Fatal(http.ListenAndServe(*hostPort, nil))
+ log.Fatal(http.ListenAndServe(*hostPort, handler))
+}
+
+func envOr(key, fallback string) string {
+ if v := os.Getenv(key); v != "" {
+ return v
+ }
+ return fallback
}
diff --git a/cmd/hubserver/storage/postgres.go b/cmd/hubserver/storage/postgres.go
index a12f847..77424c7 100644
--- a/cmd/hubserver/storage/postgres.go
+++ b/cmd/hubserver/storage/postgres.go
@@ -40,7 +40,9 @@ DO UPDATE SET lease_seconds = excluded.lease_seconds,
sub.LeaseSeconds,
sub.Secret,
)
- if e, ok := err.(pq.Error); ok {
+ // pq.Error implements error on the pointer, so the value assertion never
+ // compiled into anything useful.
+ if e, ok := err.(*pq.Error); ok {
spew.Dump(e)
}
return err
diff --git a/go.mod b/go.mod
new file mode 100644
index 0000000..8ded358
--- /dev/null
+++ b/go.mod
@@ -0,0 +1,9 @@
+module p83.nl/go/websub-hub
+
+go 1.21
+
+require (
+ github.com/davecgh/go-spew v1.1.1
+ github.com/lib/pq v1.12.3
+ github.com/pkg/errors v0.9.1
+)
diff --git a/go.sum b/go.sum
new file mode 100644
index 0000000..6417a09
--- /dev/null
+++ b/go.sum
@@ -0,0 +1,6 @@
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/lib/pq v1.12.3 h1:tTWxr2YLKwIvK90ZXEw8GP7UFHtcbTtty8zsI+YjrfQ=
+github.com/lib/pq v1.12.3/go.mod h1:/p+8NSbOcwzAEI7wiMXFlgydTwcgTr3OSKMsD2BitpA=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=