ntfy/server/server.go

550 lines
15 KiB
Go
Raw Normal View History

2021-10-23 03:26:01 +02:00
package server
import (
"bytes"
2021-10-29 05:50:38 +02:00
"context"
2021-10-24 20:22:53 +02:00
"embed"
2021-10-23 03:26:01 +02:00
_ "embed" // required for go:embed
"encoding/json"
2021-10-29 05:50:38 +02:00
firebase "firebase.google.com/go"
"firebase.google.com/go/messaging"
2021-10-23 19:21:33 +02:00
"fmt"
2021-10-29 05:50:38 +02:00
"google.golang.org/api/option"
"heckel.io/ntfy/config"
2021-11-08 15:24:34 +01:00
"heckel.io/ntfy/util"
"html/template"
2021-10-23 03:26:01 +02:00
"io"
"log"
2021-10-24 04:49:50 +02:00
"net"
2021-10-23 03:26:01 +02:00
"net/http"
"regexp"
2021-10-29 19:58:14 +02:00
"strconv"
2021-10-23 03:26:01 +02:00
"strings"
"sync"
"time"
)
2021-10-29 19:58:14 +02:00
// TODO add "max messages in a topic" limit
2021-11-01 21:39:40 +01:00
// TODO implement "since=<ID>"
2021-10-29 19:58:14 +02:00
2021-10-24 04:49:50 +02:00
// Server is the main server
2021-10-23 03:26:01 +02:00
type Server struct {
2021-10-24 04:49:50 +02:00
config *config.Config
topics map[string]*topic
visitors map[string]*visitor
2021-10-29 19:58:14 +02:00
firebase subscriber
messages int64
2021-11-03 02:09:49 +01:00
cache cache
2021-10-24 04:49:50 +02:00
mu sync.Mutex
2021-10-23 03:26:01 +02:00
}
2021-10-24 04:49:50 +02:00
// errHTTP is a generic HTTP error for any non-200 HTTP error
type errHTTP struct {
Code int
Status string
}
func (e errHTTP) Error() string {
return fmt.Sprintf("http: %s", e.Status)
2021-10-23 03:26:01 +02:00
}
2021-11-08 15:24:34 +01:00
type indexPage struct {
Topic string
CacheDuration string
}
type sinceTime time.Time
func (t sinceTime) IsAll() bool {
return t == sinceAllMessages
}
func (t sinceTime) IsNone() bool {
return t == sinceNoMessages
}
func (t sinceTime) Time() time.Time {
return time.Time(t)
}
var (
sinceAllMessages = sinceTime(time.Unix(0, 0))
sinceNoMessages = sinceTime(time.Unix(1, 0))
)
2021-10-23 03:26:01 +02:00
const (
messageLimit = 512
2021-10-23 03:26:01 +02:00
)
var (
2021-11-08 15:24:34 +01:00
topicRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}$`) // Regex must match JS & Android app!
2021-11-15 13:56:58 +01:00
jsonRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}(,[-_A-Za-z0-9]{1,64})*/json$`)
sseRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}(,[-_A-Za-z0-9]{1,64})*/sse$`)
rawRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}(,[-_A-Za-z0-9]{1,64})*/raw$`)
2021-10-29 19:58:14 +02:00
staticRegex = regexp.MustCompile(`^/static/.+`)
2021-10-23 03:26:01 +02:00
2021-11-08 15:24:34 +01:00
//go:embed "index.gohtml"
indexSource string
indexTemplate = template.Must(template.New("index").Parse(indexSource))
2021-11-18 15:22:33 +01:00
//go:embed "example.html"
2021-11-27 22:12:08 +01:00
exampleSource string
2021-11-18 15:22:33 +01:00
2021-10-24 20:22:53 +02:00
//go:embed static
webStaticFs embed.FS
2021-10-29 19:58:14 +02:00
errHTTPBadRequest = &errHTTP{http.StatusBadRequest, http.StatusText(http.StatusBadRequest)}
2021-10-24 04:49:50 +02:00
errHTTPNotFound = &errHTTP{http.StatusNotFound, http.StatusText(http.StatusNotFound)}
errHTTPTooManyRequests = &errHTTP{http.StatusTooManyRequests, http.StatusText(http.StatusTooManyRequests)}
2021-10-23 03:26:01 +02:00
)
2021-10-29 05:50:38 +02:00
func New(conf *config.Config) (*Server, error) {
2021-10-29 19:58:14 +02:00
var firebaseSubscriber subscriber
2021-10-29 05:50:38 +02:00
if conf.FirebaseKeyFile != "" {
2021-10-29 19:58:14 +02:00
var err error
firebaseSubscriber, err = createFirebaseSubscriber(conf)
2021-10-29 05:50:38 +02:00
if err != nil {
return nil, err
}
}
2021-11-03 02:09:49 +01:00
cache, err := createCache(conf)
2021-11-02 19:08:21 +01:00
if err != nil {
return nil, err
}
2021-11-03 02:09:49 +01:00
topics, err := cache.Topics()
if err != nil {
return nil, err
2021-11-02 19:08:21 +01:00
}
for _, t := range topics {
if firebaseSubscriber != nil {
t.Subscribe(firebaseSubscriber)
}
}
2021-10-23 03:26:01 +02:00
return &Server{
2021-10-24 04:49:50 +02:00
config: conf,
2021-11-03 02:09:49 +01:00
cache: cache,
2021-10-29 19:58:14 +02:00
firebase: firebaseSubscriber,
2021-11-02 19:08:21 +01:00
topics: topics,
2021-10-24 04:49:50 +02:00
visitors: make(map[string]*visitor),
2021-10-29 05:50:38 +02:00
}, nil
2021-10-23 03:26:01 +02:00
}
2021-11-03 02:09:49 +01:00
func createCache(conf *config.Config) (cache, error) {
if conf.CacheFile != "" {
return newSqliteCache(conf.CacheFile)
2021-11-02 19:08:21 +01:00
}
2021-11-03 02:09:49 +01:00
return newMemCache(), nil
2021-11-02 19:08:21 +01:00
}
2021-10-29 19:58:14 +02:00
func createFirebaseSubscriber(conf *config.Config) (subscriber, error) {
fb, err := firebase.NewApp(context.Background(), nil, option.WithCredentialsFile(conf.FirebaseKeyFile))
if err != nil {
return nil, err
}
msg, err := fb.Messaging(context.Background())
if err != nil {
return nil, err
}
return func(m *message) error {
_, err := msg.Send(context.Background(), &messaging.Message{
2021-10-29 21:47:46 +02:00
Topic: m.Topic,
2021-10-29 19:58:14 +02:00
Data: map[string]string{
2021-11-27 22:12:08 +01:00
"id": m.ID,
"time": fmt.Sprintf("%d", m.Time),
"event": m.Event,
"topic": m.Topic,
"priority": fmt.Sprintf("%d", m.Priority),
"tags": strings.Join(m.Tags, ","),
"title": m.Title,
"message": m.Message,
2021-10-29 19:58:14 +02:00
},
})
return err
}, nil
}
2021-10-23 03:26:01 +02:00
func (s *Server) Run() error {
2021-10-24 04:49:50 +02:00
go func() {
ticker := time.NewTicker(s.config.ManagerInterval)
for {
<-ticker.C
s.updateStatsAndExpire()
}
}()
2021-10-23 21:22:17 +02:00
return s.listenAndServe()
}
func (s *Server) listenAndServe() error {
log.Printf("Listening on %s", s.config.ListenHTTP)
2021-10-23 03:26:01 +02:00
http.HandleFunc("/", s.handle)
return http.ListenAndServe(s.config.ListenHTTP, nil)
2021-10-23 03:26:01 +02:00
}
func (s *Server) handle(w http.ResponseWriter, r *http.Request) {
if err := s.handleInternal(w, r); err != nil {
2021-10-24 04:49:50 +02:00
if e, ok := err.(*errHTTP); ok {
s.fail(w, r, e.Code, e)
} else {
s.fail(w, r, http.StatusInternalServerError, err)
}
2021-10-23 03:26:01 +02:00
}
}
func (s *Server) handleInternal(w http.ResponseWriter, r *http.Request) error {
2021-11-08 15:24:34 +01:00
if r.Method == http.MethodGet && (r.URL.Path == "/" || topicRegex.MatchString(r.URL.Path)) {
2021-10-23 03:26:01 +02:00
return s.handleHome(w, r)
2021-11-18 15:22:33 +01:00
} else if r.Method == http.MethodGet && r.URL.Path == "/example.html" {
return s.handleExample(w, r)
} else if r.Method == http.MethodHead && r.URL.Path == "/" {
return s.handleEmpty(w, r)
2021-10-24 20:22:53 +02:00
} else if r.Method == http.MethodGet && staticRegex.MatchString(r.URL.Path) {
return s.handleStatic(w, r)
} else if r.Method == http.MethodOptions {
return s.handleOptions(w, r)
2021-10-29 19:58:14 +02:00
} else if (r.Method == http.MethodPut || r.Method == http.MethodPost) && topicRegex.MatchString(r.URL.Path) {
return s.withRateLimit(w, r, s.handlePublish)
2021-10-23 03:26:01 +02:00
} else if r.Method == http.MethodGet && jsonRegex.MatchString(r.URL.Path) {
return s.withRateLimit(w, r, s.handleSubscribeJSON)
2021-10-23 19:21:33 +02:00
} else if r.Method == http.MethodGet && sseRegex.MatchString(r.URL.Path) {
return s.withRateLimit(w, r, s.handleSubscribeSSE)
} else if r.Method == http.MethodGet && rawRegex.MatchString(r.URL.Path) {
return s.withRateLimit(w, r, s.handleSubscribeRaw)
2021-10-23 03:26:01 +02:00
}
2021-10-24 04:49:50 +02:00
return errHTTPNotFound
2021-10-23 03:26:01 +02:00
}
func (s *Server) handleHome(w http.ResponseWriter, r *http.Request) error {
2021-11-08 15:24:34 +01:00
return indexTemplate.Execute(w, &indexPage{
Topic: r.URL.Path[1:],
CacheDuration: util.DurationToHuman(s.config.CacheDuration),
})
2021-10-23 03:26:01 +02:00
}
func (s *Server) handleEmpty(w http.ResponseWriter, r *http.Request) error {
return nil
}
2021-11-18 15:22:33 +01:00
func (s *Server) handleExample(w http.ResponseWriter, r *http.Request) error {
_, err := io.WriteString(w, exampleSource)
return err
}
2021-10-29 19:58:14 +02:00
func (s *Server) handleStatic(w http.ResponseWriter, r *http.Request) error {
http.FileServer(http.FS(webStaticFs)).ServeHTTP(w, r)
return nil
}
2021-11-01 21:39:40 +01:00
func (s *Server) handlePublish(w http.ResponseWriter, r *http.Request, v *visitor) error {
2021-11-15 13:56:58 +01:00
t, err := s.topicFromID(r.URL.Path[1:])
2021-11-01 21:39:40 +01:00
if err != nil {
return err
}
2021-10-23 03:26:01 +02:00
reader := io.LimitReader(r.Body, messageLimit)
b, err := io.ReadAll(reader)
if err != nil {
return err
}
2021-11-02 19:08:21 +01:00
m := newDefaultMessage(t.id, string(b))
2021-11-18 20:21:00 +01:00
if m.Message == "" {
return errHTTPBadRequest
}
2021-11-27 22:12:08 +01:00
title, priority, tags := parseHeaders(r.Header)
m.Title = title
m.Priority = priority
m.Tags = tags
2021-11-02 19:08:21 +01:00
if err := t.Publish(m); err != nil {
2021-10-29 05:50:38 +02:00
return err
}
2021-11-03 02:09:49 +01:00
if err := s.cache.AddMessage(m); err != nil {
return err
2021-11-02 19:08:21 +01:00
}
2021-10-24 19:34:15 +02:00
w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
2021-11-03 16:33:34 +01:00
if err := json.NewEncoder(w).Encode(m); err != nil {
return err
}
2021-10-29 19:58:14 +02:00
s.mu.Lock()
s.messages++
s.mu.Unlock()
2021-10-24 19:34:15 +02:00
return nil
2021-10-23 03:26:01 +02:00
}
2021-11-27 22:12:08 +01:00
func parseHeaders(header http.Header) (title string, priority int, tags []string) {
title = readHeader(header, "x-title", "title", "ti", "t")
priorityStr := readHeader(header, "x-priority", "priority", "prio", "p")
if priorityStr != "" {
switch strings.ToLower(priorityStr) {
case "1", "min":
priority = 1
case "2", "low":
priority = 2
case "4", "high":
priority = 4
case "5", "max", "urgent":
priority = 5
default:
priority = 3
}
}
tagsStr := readHeader(header, "x-tags", "tags", "ta")
if tagsStr != "" {
tags = strings.Split(tagsStr, ",")
}
return title, priority, tags
}
func readHeader(header http.Header, names ...string) string {
for _, name := range names {
value := header.Get(name)
if value != "" {
return value
}
}
return ""
}
2021-11-01 20:21:38 +01:00
func (s *Server) handleSubscribeJSON(w http.ResponseWriter, r *http.Request, v *visitor) error {
encoder := func(msg *message) (string, error) {
var buf bytes.Buffer
if err := json.NewEncoder(&buf).Encode(&msg); err != nil {
return "", err
2021-10-23 03:26:01 +02:00
}
return buf.String(), nil
2021-10-23 03:26:01 +02:00
}
2021-11-07 19:08:03 +01:00
return s.handleSubscribe(w, r, v, "json", "application/x-ndjson", encoder)
2021-10-23 03:26:01 +02:00
}
2021-11-01 20:21:38 +01:00
func (s *Server) handleSubscribeSSE(w http.ResponseWriter, r *http.Request, v *visitor) error {
encoder := func(msg *message) (string, error) {
2021-10-23 19:21:33 +02:00
var buf bytes.Buffer
if err := json.NewEncoder(&buf).Encode(&msg); err != nil {
return "", err
2021-10-23 19:21:33 +02:00
}
2021-10-29 14:29:27 +02:00
if msg.Event != messageEvent {
return fmt.Sprintf("event: %s\ndata: %s\n", msg.Event, buf.String()), nil // Browser's .onmessage() does not fire on this!
2021-10-23 19:21:33 +02:00
}
return fmt.Sprintf("data: %s\n", buf.String()), nil
2021-10-23 21:22:17 +02:00
}
2021-11-01 20:21:38 +01:00
return s.handleSubscribe(w, r, v, "sse", "text/event-stream", encoder)
2021-10-23 19:21:33 +02:00
}
2021-11-01 20:21:38 +01:00
func (s *Server) handleSubscribeRaw(w http.ResponseWriter, r *http.Request, v *visitor) error {
encoder := func(msg *message) (string, error) {
2021-11-02 19:10:56 +01:00
if msg.Event == messageEvent { // only handle default events
return strings.ReplaceAll(msg.Message, "\n", " ") + "\n", nil
}
return "\n", nil // "keepalive" and "open" events just send an empty line
}
2021-11-01 20:21:38 +01:00
return s.handleSubscribe(w, r, v, "raw", "text/plain", encoder)
}
2021-11-01 20:21:38 +01:00
func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request, v *visitor, format string, contentType string, encoder messageEncoder) error {
if err := v.AddSubscription(); err != nil {
2021-11-01 21:39:40 +01:00
return errHTTPTooManyRequests
2021-11-01 20:21:38 +01:00
}
defer v.RemoveSubscription()
2021-11-15 13:56:58 +01:00
topicsStr := strings.TrimSuffix(r.URL.Path[1:], "/"+format) // Hack
topicIDs := strings.Split(topicsStr, ",")
topics, err := s.topicsFromIDs(topicIDs...)
2021-11-01 21:39:40 +01:00
if err != nil {
return err
}
2021-10-29 19:58:14 +02:00
since, err := parseSince(r)
if err != nil {
return err
}
poll := r.URL.Query().Has("poll")
sub := func(msg *message) error {
m, err := encoder(msg)
if err != nil {
return err
}
if _, err := w.Write([]byte(m)); err != nil {
return err
}
if fl, ok := w.(http.Flusher); ok {
fl.Flush()
}
return nil
}
2021-11-07 19:08:03 +01:00
w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
w.Header().Set("Content-Type", contentType+"; charset=utf-8") // Android/Volley client needs charset!
2021-10-29 19:58:14 +02:00
if poll {
2021-11-15 13:56:58 +01:00
return s.sendOldMessages(topics, since, sub)
2021-10-29 19:58:14 +02:00
}
2021-11-15 13:56:58 +01:00
subscriberIDs := make([]int, 0)
for _, t := range topics {
subscriberIDs = append(subscriberIDs, t.Subscribe(sub))
}
defer func() {
for i, subscriberID := range subscriberIDs {
topics[i].Unsubscribe(subscriberID) // Order!
}
}()
if err := sub(newOpenMessage(topicsStr)); err != nil { // Send out open message
2021-10-29 19:58:14 +02:00
return err
}
2021-11-15 13:56:58 +01:00
if err := s.sendOldMessages(topics, since, sub); err != nil {
return err
}
for {
select {
case <-r.Context().Done():
return nil
case <-time.After(s.config.KeepaliveInterval):
2021-11-01 20:21:38 +01:00
v.Keepalive()
2021-11-15 13:56:58 +01:00
if err := sub(newKeepaliveMessage(topicsStr)); err != nil { // Send keepalive message
return err
}
}
}
}
2021-11-15 13:56:58 +01:00
func (s *Server) sendOldMessages(topics []*topic, since sinceTime, sub subscriber) error {
if since.IsNone() {
2021-10-29 19:58:14 +02:00
return nil
}
2021-11-15 13:56:58 +01:00
for _, t := range topics {
messages, err := s.cache.Messages(t.id, since)
if err != nil {
2021-10-29 19:58:14 +02:00
return err
}
2021-11-15 13:56:58 +01:00
for _, m := range messages {
if err := sub(m); err != nil {
return err
}
}
2021-10-29 19:58:14 +02:00
}
2021-10-24 19:34:15 +02:00
return nil
}
// parseSince returns a timestamp identifying the time span from which cached messages should be received.
//
// Values in the "since=..." parameter can be either a unix timestamp or a duration (e.g. 12h), or
// "all" for all messages.
func parseSince(r *http.Request) (sinceTime, error) {
2021-10-29 19:58:14 +02:00
if !r.URL.Query().Has("since") {
if r.URL.Query().Has("poll") {
return sinceAllMessages, nil
}
return sinceNoMessages, nil
}
if r.URL.Query().Get("since") == "all" {
return sinceAllMessages, nil
2021-10-29 19:58:14 +02:00
}
if s, err := strconv.ParseInt(r.URL.Query().Get("since"), 10, 64); err == nil {
return sinceTime(time.Unix(s, 0)), nil
2021-10-29 19:58:14 +02:00
}
if d, err := time.ParseDuration(r.URL.Query().Get("since")); err == nil {
return sinceTime(time.Now().Add(-1 * d)), nil
2021-10-29 19:58:14 +02:00
}
return sinceNoMessages, errHTTPBadRequest
2021-10-29 19:58:14 +02:00
}
func (s *Server) handleOptions(w http.ResponseWriter, r *http.Request) error {
w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
w.Header().Set("Access-Control-Allow-Methods", "GET, PUT, POST")
2021-10-24 20:22:53 +02:00
return nil
}
2021-11-15 13:56:58 +01:00
func (s *Server) topicFromID(id string) (*topic, error) {
topics, err := s.topicsFromIDs(id)
if err != nil {
return nil, err
}
return topics[0], nil
}
2021-11-27 22:12:08 +01:00
func (s *Server) topicsFromIDs(ids ...string) ([]*topic, error) {
2021-10-23 03:26:01 +02:00
s.mu.Lock()
defer s.mu.Unlock()
2021-11-15 13:56:58 +01:00
topics := make([]*topic, 0)
2021-11-27 22:12:08 +01:00
for _, id := range ids {
2021-11-15 13:56:58 +01:00
if _, ok := s.topics[id]; !ok {
if len(s.topics) >= s.config.GlobalTopicLimit {
return nil, errHTTPTooManyRequests
}
s.topics[id] = newTopic(id, time.Now())
if s.firebase != nil {
s.topics[id].Subscribe(s.firebase)
}
2021-10-29 19:58:14 +02:00
}
2021-11-15 13:56:58 +01:00
topics = append(topics, s.topics[id])
2021-10-23 03:26:01 +02:00
}
2021-11-15 13:56:58 +01:00
return topics, nil
2021-10-23 03:26:01 +02:00
}
2021-10-29 19:58:14 +02:00
func (s *Server) updateStatsAndExpire() {
2021-10-23 03:26:01 +02:00
s.mu.Lock()
defer s.mu.Unlock()
2021-10-29 19:58:14 +02:00
// Expire visitors from rate visitors map
for ip, v := range s.visitors {
2021-11-01 20:21:38 +01:00
if v.Stale() {
2021-10-29 19:58:14 +02:00
delete(s.visitors, ip)
}
2021-10-23 03:26:01 +02:00
}
2021-11-02 19:08:21 +01:00
// Prune cache
2021-11-03 16:33:34 +01:00
if err := s.cache.Prune(s.config.CacheDuration); err != nil {
2021-11-03 02:09:49 +01:00
log.Printf("error pruning cache: %s", err.Error())
2021-11-02 19:08:21 +01:00
}
2021-11-01 20:21:38 +01:00
// Prune old messages, remove subscriptions without subscribers
2021-11-03 02:09:49 +01:00
var subscribers, messages int
2021-10-29 19:58:14 +02:00
for _, t := range s.topics {
2021-11-03 02:09:49 +01:00
subs := t.Subscribers()
msgs, err := s.cache.MessageCount(t.id)
if err != nil {
log.Printf("cannot get stats for topic %s: %s", t.id, err.Error())
continue
}
2021-11-01 21:39:40 +01:00
if msgs == 0 && (subs == 0 || (s.firebase != nil && subs == 1)) { // Firebase is a subscriber!
2021-10-29 19:58:14 +02:00
delete(s.topics, t.id)
2021-11-03 02:09:49 +01:00
continue
2021-10-29 19:58:14 +02:00
}
subscribers += subs
messages += msgs
}
2021-11-03 02:09:49 +01:00
// Print stats
2021-10-29 19:58:14 +02:00
log.Printf("Stats: %d message(s) published, %d topic(s) active, %d subscriber(s), %d message(s) buffered, %d visitor(s)",
s.messages, len(s.topics), subscribers, messages, len(s.visitors))
}
2021-10-24 04:49:50 +02:00
func (s *Server) withRateLimit(w http.ResponseWriter, r *http.Request, handler func(w http.ResponseWriter, r *http.Request, v *visitor) error) error {
v := s.visitor(r)
if err := v.RequestAllowed(); err != nil {
return err
}
return handler(w, r, v)
}
2021-10-24 04:49:50 +02:00
// visitor creates or retrieves a rate.Limiter for the given visitor.
// This function was taken from https://www.alexedwards.net/blog/how-to-rate-limit-http-requests (MIT).
func (s *Server) visitor(r *http.Request) *visitor {
2021-10-24 04:49:50 +02:00
s.mu.Lock()
defer s.mu.Unlock()
remoteAddr := r.RemoteAddr
2021-10-24 04:49:50 +02:00
ip, _, err := net.SplitHostPort(remoteAddr)
if err != nil {
ip = remoteAddr // This should not happen in real life; only in tests.
}
if s.config.BehindProxy && r.Header.Get("X-Forwarded-For") != "" {
ip = r.Header.Get("X-Forwarded-For")
}
2021-10-24 04:49:50 +02:00
v, exists := s.visitors[ip]
if !exists {
2021-11-01 20:21:38 +01:00
s.visitors[ip] = newVisitor(s.config)
return s.visitors[ip]
2021-10-24 04:49:50 +02:00
}
v.seen = time.Now()
return v
}
func (s *Server) fail(w http.ResponseWriter, r *http.Request, code int, err error) {
log.Printf("[%s] %s - %d - %s", r.RemoteAddr, r.Method, code, err.Error())
w.WriteHeader(code)
io.WriteString(w, fmt.Sprintf("%s\n", http.StatusText(code)))
}