ntfy/server/server.go

305 lines
8.1 KiB
Go
Raw Normal View History

2021-10-23 03:26:01 +02:00
package server
import (
"bytes"
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-23 19:21:33 +02:00
"fmt"
2021-10-24 04:49:50 +02:00
"golang.org/x/time/rate"
"heckel.io/ntfy/config"
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"
"strings"
"sync"
"time"
)
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
mu sync.Mutex
2021-10-23 03:26:01 +02:00
}
2021-10-24 04:49:50 +02:00
// visitor represents an API user, and its associated rate.Limiter used for rate limiting
type visitor struct {
limiter *rate.Limiter
seen time.Time
}
// 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
}
const (
2021-10-24 04:49:50 +02:00
messageLimit = 1024
visitorExpungeAfter = 30 * time.Minute
2021-10-23 03:26:01 +02:00
)
var (
topicRegex = regexp.MustCompile(`^/[^/]+$`)
jsonRegex = regexp.MustCompile(`^/[^/]+/json$`)
sseRegex = regexp.MustCompile(`^/[^/]+/sse$`)
rawRegex = regexp.MustCompile(`^/[^/]+/raw$`)
staticRegex = regexp.MustCompile(`^/static/.+`)
2021-10-23 03:26:01 +02:00
//go:embed "index.html"
indexSource string
2021-10-24 20:22:53 +02:00
//go:embed static
webStaticFs embed.FS
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
)
func New(conf *config.Config) *Server {
2021-10-23 03:26:01 +02:00
return &Server{
2021-10-24 04:49:50 +02:00
config: conf,
topics: make(map[string]*topic),
visitors: make(map[string]*visitor),
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
}
2021-10-24 04:49:50 +02:00
func (s *Server) updateStatsAndExpire() {
s.mu.Lock()
defer s.mu.Unlock()
// Expire visitors from rate visitors map
for ip, v := range s.visitors {
if time.Since(v.seen) > visitorExpungeAfter {
delete(s.visitors, ip)
2021-10-23 21:22:17 +02:00
}
}
2021-10-24 04:49:50 +02:00
// Print stats
var subscribers, messages int
for _, t := range s.topics {
subs, msgs := t.Stats()
subscribers += subs
messages += msgs
}
log.Printf("Stats: %d topic(s), %d subscriber(s), %d message(s) sent, %d visitor(s)",
len(s.topics), subscribers, messages, len(s.visitors))
2021-10-23 21:22:17 +02:00
}
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-10-24 04:49:50 +02:00
v := s.visitor(r.RemoteAddr)
if !v.limiter.Allow() {
return errHTTPTooManyRequests
}
2021-10-23 03:26:01 +02:00
if r.Method == http.MethodGet && r.URL.Path == "/" {
return s.handleHome(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)
2021-10-23 03:26:01 +02:00
} else if r.Method == http.MethodGet && jsonRegex.MatchString(r.URL.Path) {
2021-10-23 19:21:33 +02:00
return s.handleSubscribeJSON(w, r)
} else if r.Method == http.MethodGet && sseRegex.MatchString(r.URL.Path) {
return s.handleSubscribeSSE(w, r)
} else if r.Method == http.MethodGet && rawRegex.MatchString(r.URL.Path) {
return s.handleSubscribeRaw(w, r)
2021-10-23 03:26:01 +02:00
} else if (r.Method == http.MethodPut || r.Method == http.MethodPost) && topicRegex.MatchString(r.URL.Path) {
return s.handlePublishHTTP(w, r)
2021-10-24 19:34:15 +02:00
} else if r.Method == http.MethodOptions {
return s.handleOptions(w, r)
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 {
_, err := io.WriteString(w, indexSource)
return err
}
func (s *Server) handlePublishHTTP(w http.ResponseWriter, r *http.Request) error {
t, err := s.topic(r.URL.Path[1:])
if err != nil {
return err
}
reader := io.LimitReader(r.Body, messageLimit)
b, err := io.ReadAll(reader)
if err != nil {
return err
}
if err := t.Publish(newDefaultMessage(string(b))); err != nil {
2021-10-24 19:34:15 +02:00
return err
}
w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
return nil
2021-10-23 03:26:01 +02:00
}
2021-10-23 19:21:33 +02:00
func (s *Server) handleSubscribeJSON(w http.ResponseWriter, r *http.Request) 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
}
return s.handleSubscribe(w, r, "json", "application/stream+json", encoder)
2021-10-23 03:26:01 +02:00
}
2021-10-23 19:21:33 +02:00
func (s *Server) handleSubscribeSSE(w http.ResponseWriter, r *http.Request) 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
}
if msg.Event != "" {
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
}
return s.handleSubscribe(w, r, "sse", "text/event-stream", encoder)
2021-10-23 19:21:33 +02:00
}
func (s *Server) handleSubscribeRaw(w http.ResponseWriter, r *http.Request) error {
encoder := func(msg *message) (string, error) {
if msg.Event == "" { // only handle default events
return strings.ReplaceAll(msg.Message, "\n", " ") + "\n", nil
}
return "\n", nil // "keepalive" and "open" events just send an empty line
}
return s.handleSubscribe(w, r, "raw", "text/plain", encoder)
}
func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request, format string, contentType string, encoder messageEncoder) error {
t := s.createTopic(strings.TrimSuffix(r.URL.Path[1:], "/"+format)) // Hack
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
}
subscriberID := t.Subscribe(sub)
defer s.unsubscribe(t, subscriberID)
w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
w.Header().Set("Content-Type", contentType)
if err := sub(newOpenMessage()); err != nil { // Send out open message
return err
}
for {
select {
case <-t.ctx.Done():
return nil
case <-r.Context().Done():
return nil
case <-time.After(s.config.KeepaliveInterval):
if err := sub(newKeepaliveMessage()); err != nil { // Send keepalive message
return err
}
}
}
}
2021-10-24 19:34:15 +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")
return nil
}
2021-10-24 20:22:53 +02:00
func (s *Server) handleStatic(w http.ResponseWriter, r *http.Request) error {
http.FileServer(http.FS(webStaticFs)).ServeHTTP(w, r)
return nil
}
2021-10-23 03:26:01 +02:00
func (s *Server) createTopic(id string) *topic {
s.mu.Lock()
defer s.mu.Unlock()
if _, ok := s.topics[id]; !ok {
s.topics[id] = newTopic(id)
}
return s.topics[id]
}
func (s *Server) topic(topicID string) (*topic, error) {
s.mu.Lock()
defer s.mu.Unlock()
c, ok := s.topics[topicID]
if !ok {
2021-10-24 04:49:50 +02:00
return nil, errHTTPNotFound
2021-10-23 03:26:01 +02:00
}
return c, nil
}
func (s *Server) unsubscribe(t *topic, subscriberID int) {
s.mu.Lock()
defer s.mu.Unlock()
if subscribers := t.Unsubscribe(subscriberID); subscribers == 0 {
delete(s.topics, t.id)
}
}
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(remoteAddr string) *visitor {
s.mu.Lock()
defer s.mu.Unlock()
ip, _, err := net.SplitHostPort(remoteAddr)
if err != nil {
ip = remoteAddr // This should not happen in real life; only in tests.
}
v, exists := s.visitors[ip]
if !exists {
v = &visitor{
rate.NewLimiter(s.config.Limit, s.config.LimitBurst),
time.Now(),
}
s.visitors[ip] = v
return v
}
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)))
}