Fix raw endpoint

README.md

@@ -64,13 +64,13 @@ sudo apt install ntfy
 **Debian/Ubuntu** (*manual install*)**:**
 ```bash
 sudo apt install tmux
-wget https://github.com/binwiederhier/ntfy/releases/download/v1.0.0/ntfy_1.0.0_amd64.deb
-dpkg -i ntfy_1.0.0_amd64.deb
+wget https://github.com/binwiederhier/ntfy/releases/download/v1.1.2/ntfy_1.1.2_amd64.deb
+dpkg -i ntfy_1.1.2_amd64.deb
 ```
 
 **Fedora/RHEL/CentOS:**
 ```bash
-rpm -ivh https://github.com/binwiederhier/ntfy/releases/download/v1.0.0/ntfy_1.0.0_amd64.rpm
+rpm -ivh https://github.com/binwiederhier/ntfy/releases/download/v1.1.2/ntfy_1.1.2_amd64.rpm
 ```
 
 **Docker:**
@@ -85,8 +85,8 @@ go get -u heckel.io/ntfy
 
 **Manual install** (*any x86_64-based Linux*)**:**
 ```bash
-wget https://github.com/binwiederhier/ntfy/releases/download/v1.0.0/ntfy_1.0.0_linux_x86_64.tar.gz
-sudo tar -C /usr/bin -zxf ntfy_1.0.0_linux_x86_64.tar.gz ntfy
+wget https://github.com/binwiederhier/ntfy/releases/download/v1.1.2/ntfy_1.1.2_linux_x86_64.tar.gz
+sudo tar -C /usr/bin -zxf ntfy_1.1.2_linux_x86_64.tar.gz ntfy
 ./ntfy
 ```
 

config/config.go

@@ -14,33 +14,41 @@ const (
 	DefaultManagerInterval       = time.Minute
 )
 
-// Defines the max number of requests, here:
-// 50 requests bucket, replenished at a rate of 1 per second
+// Defines all the limits
+// - request limit: max number of PUT/GET/.. requests (here: 50 requests bucket, replenished at a rate of 1 per second)
+// - global topic limit: max number of topics overall
+// - subscription limit: max number of subscriptions (active HTTP connections) per per-visitor/IP
 var (
-	defaultLimit      = rate.Every(time.Second)
-	defaultLimitBurst = 50
+	defaultGlobalTopicLimit         = 5000
+	defaultVisitorRequestLimit      = rate.Every(time.Second)
+	defaultVisitorRequestLimitBurst = 50
+	defaultVisitorSubscriptionLimit = 30
 )
 
 // Config is the main config struct for the application. Use New to instantiate a default config struct.
 type Config struct {
-	ListenHTTP            string
-	FirebaseKeyFile       string
-	MessageBufferDuration time.Duration
-	KeepaliveInterval     time.Duration
-	ManagerInterval       time.Duration
-	Limit                 rate.Limit
-	LimitBurst            int
+	ListenHTTP               string
+	FirebaseKeyFile          string
+	MessageBufferDuration    time.Duration
+	KeepaliveInterval        time.Duration
+	ManagerInterval          time.Duration
+	GlobalTopicLimit         int
+	VisitorRequestLimit      rate.Limit
+	VisitorRequestLimitBurst int
+	VisitorSubscriptionLimit int
 }
 
 // New instantiates a default new config
 func New(listenHTTP string) *Config {
 	return &Config{
-		ListenHTTP:            listenHTTP,
-		FirebaseKeyFile:       "",
-		MessageBufferDuration: DefaultMessageBufferDuration,
-		KeepaliveInterval:     DefaultKeepaliveInterval,
-		ManagerInterval:       DefaultManagerInterval,
-		Limit:                 defaultLimit,
-		LimitBurst:            defaultLimitBurst,
+		ListenHTTP:               listenHTTP,
+		FirebaseKeyFile:          "",
+		MessageBufferDuration:    DefaultMessageBufferDuration,
+		KeepaliveInterval:        DefaultKeepaliveInterval,
+		ManagerInterval:          DefaultManagerInterval,
+		GlobalTopicLimit:         defaultGlobalTopicLimit,
+		VisitorRequestLimit:      defaultVisitorRequestLimit,
+		VisitorRequestLimitBurst: defaultVisitorRequestLimitBurst,
+		VisitorSubscriptionLimit: defaultVisitorSubscriptionLimit,
 	}
 }

server/index.html

@@ -25,12 +25,12 @@
     <meta property="og:site_name" content="ntfy.sh" />
     <meta property="og:title" content="ntfy.sh | simple HTTP-based pub-sub" />
     <meta property="og:description" content="ntfy is a simple HTTP-based pub-sub notification service. It allows you to send desktop notifications via scripts from any computer, entirely without signup or cost. Made with ❤ by Philipp C. Heckel, Apache License 2.0, source at https://heckel.io/ntfy." />
-    <meta property="og:image" content="/static/img/favicon.png" />
+    <meta property="og:image" content="/static/img/ntfy.png" />
     <meta property="og:url" content="https://ntfy.sh" />
 </head>
 <body>
 <div id="main">
-    <h1>ntfy.sh - simple HTTP-based pub-sub</h1>
+    <h1><img src="static/img/ntfy.png" alt="ntfy"/><br/>ntfy.sh - simple HTTP-based pub-sub</h1>
     <p>
         <b>ntfy</b> (pronounce: <i>notify</i>) is a simple HTTP-based <a href="https://en.wikipedia.org/wiki/Publish%E2%80%93subscribe_pattern">pub-sub</a> notification service.
         It allows you to send <b>desktop notifications via scripts from any computer</b>, entirely <b>without signup or cost</b>.
@@ -81,6 +81,12 @@
     <ul id="topicsList"></ul>
     <audio id="notifySound" src="static/sound/mixkit-message-pop-alert-2354.mp3"></audio>
 
+    <h3>Subscribe via phone</h3>
+    <p>
+        Once it's approved, you can use the <b>Ntfy Android App</b> to receive notifications directly on your phone. Just like
+        the server, this app is also <a href="https://github.com/binwiederhier/ntfy-android">open source</a>.
+    </p>
+
     <h3>Subscribe via your app, or via the CLI</h3>
     <p class="smallMarginBottom">
         Using <a href="https://developer.mozilla.org/en-US/docs/Web/API/EventSource">EventSource</a> in JS, you can consume
@@ -142,6 +148,7 @@
         $ curl -s "ntfy.sh/mytopic/json?poll=1&since=10m"<br/>
         # Returns messages from up to 10 minutes ago and ends the connection
     </code>
+
     <h2>FAQ</h2>
     <p>
         <b>Isn't this like ...?</b><br/>
@@ -165,6 +172,28 @@
         That said, the logs do not contain any topic names or other details about you. Check the code if you don't believe me.
     </p>
 
+    <p>
+        <b>Why is Firebase used?</b><br/>
+        In addition to caching messages locally and delivering them to long-polling subscribers, all messages are also
+        published to Firebase Cloud Messaging (FCM) (if <tt>FirebaseKeyFile</tt> is set, which it is on ntfy.sh). This
+        is to facilitate instant notifications on Android. I tried really, really hard to avoid using FCM, but newer
+        versions of Android made it impossible to implement <a href="https://developer.android.com/guide/background">background services</a>>.
+        I'm sorry.
+    </p>
+
+    <h2>Privacy policy</h2>
+    <p>
+        Neither the server nor the app record any personal information, or share any of the messages and topics with
+        any outside service. All data is exclusively used to make the service function properly. The notable exception
+        is the Firebase Cloud Messaging (FCM) service, which is required to provide instant Android notifications (see
+        FAQ for details).
+    </p>
+
+    <p>
+        The web server does not log or otherwise store request paths, remote IP addresses or even topics or messages,
+        aside from a short on-disk cache (up to a day) to support the <tt>since=</tt> feature and service restarts.
+    </p>
+
     <center id="ironicCenterTagDontFreakOut"><i>Made with ❤️ by <a href="https://heckel.io">Philipp C. Heckel</a></i></center>
 </div>
 <script src="static/js/app.js"></script>

server/server.go

@@ -9,7 +9,6 @@ import (
 	firebase "firebase.google.com/go"
 	"firebase.google.com/go/messaging"
 	"fmt"
-	"golang.org/x/time/rate"
 	"google.golang.org/api/option"
 	"heckel.io/ntfy/config"
 	"io"
@@ -23,9 +22,9 @@ import (
 	"time"
 )
 
-// TODO add "max connections open" limit
 // TODO add "max messages in a topic" limit
-// TODO add "max topics" limit
+// TODO implement persistence
+// TODO implement "since=<ID>"
 
 // Server is the main server
 type Server struct {
@@ -37,12 +36,6 @@ type Server struct {
 	mu       sync.Mutex
 }
 
-// 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
@@ -54,8 +47,7 @@ func (e errHTTP) Error() string {
 }
 
 const (
-	messageLimit        = 1024
-	visitorExpungeAfter = 30 * time.Minute
+	messageLimit = 1024
 )
 
 var (
@@ -147,21 +139,21 @@ func (s *Server) handle(w http.ResponseWriter, r *http.Request) {
 
 func (s *Server) handleInternal(w http.ResponseWriter, r *http.Request) error {
 	v := s.visitor(r.RemoteAddr)
-	if !v.limiter.Allow() {
-		return errHTTPTooManyRequests
+	if err := v.RequestAllowed(); err != nil {
+		return err
 	}
 	if r.Method == http.MethodGet && r.URL.Path == "/" {
 		return s.handleHome(w, r)
 	} else if r.Method == http.MethodGet && staticRegex.MatchString(r.URL.Path) {
 		return s.handleStatic(w, r)
 	} else if (r.Method == http.MethodPut || r.Method == http.MethodPost) && topicRegex.MatchString(r.URL.Path) {
-		return s.handlePublish(w, r)
+		return s.handlePublish(w, r, v)
 	} else if r.Method == http.MethodGet && jsonRegex.MatchString(r.URL.Path) {
-		return s.handleSubscribeJSON(w, r)
+		return s.handleSubscribeJSON(w, r, v)
 	} else if r.Method == http.MethodGet && sseRegex.MatchString(r.URL.Path) {
-		return s.handleSubscribeSSE(w, r)
+		return s.handleSubscribeSSE(w, r, v)
 	} else if r.Method == http.MethodGet && rawRegex.MatchString(r.URL.Path) {
-		return s.handleSubscribeRaw(w, r)
+		return s.handleSubscribeRaw(w, r, v)
 	} else if r.Method == http.MethodOptions {
 		return s.handleOptions(w, r)
 	}
@@ -178,8 +170,11 @@ func (s *Server) handleStatic(w http.ResponseWriter, r *http.Request) error {
 	return nil
 }
 
-func (s *Server) handlePublish(w http.ResponseWriter, r *http.Request) error {
-	t := s.createTopic(r.URL.Path[1:])
+func (s *Server) handlePublish(w http.ResponseWriter, r *http.Request, v *visitor) 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 {
@@ -195,7 +190,7 @@ func (s *Server) handlePublish(w http.ResponseWriter, r *http.Request) error {
 	return nil
 }
 
-func (s *Server) handleSubscribeJSON(w http.ResponseWriter, r *http.Request) error {
+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 {
@@ -203,10 +198,10 @@ func (s *Server) handleSubscribeJSON(w http.ResponseWriter, r *http.Request) err
 		}
 		return buf.String(), nil
 	}
-	return s.handleSubscribe(w, r, "json", "application/stream+json", encoder)
+	return s.handleSubscribe(w, r, v, "json", "application/stream+json", encoder)
 }
 
-func (s *Server) handleSubscribeSSE(w http.ResponseWriter, r *http.Request) error {
+func (s *Server) handleSubscribeSSE(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 {
@@ -217,21 +212,28 @@ func (s *Server) handleSubscribeSSE(w http.ResponseWriter, r *http.Request) erro
 		}
 		return fmt.Sprintf("data: %s\n", buf.String()), nil
 	}
-	return s.handleSubscribe(w, r, "sse", "text/event-stream", encoder)
+	return s.handleSubscribe(w, r, v, "sse", "text/event-stream", encoder)
 }
 
-func (s *Server) handleSubscribeRaw(w http.ResponseWriter, r *http.Request) error {
+func (s *Server) handleSubscribeRaw(w http.ResponseWriter, r *http.Request, v *visitor) error {
 	encoder := func(msg *message) (string, error) {
-		if msg.Event == "" { // only handle default events
+		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
 	}
-	return s.handleSubscribe(w, r, "raw", "text/plain", encoder)
+	return s.handleSubscribe(w, r, v, "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
+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 {
+		return errHTTPTooManyRequests
+	}
+	defer v.RemoveSubscription()
+	t, err := s.topic(strings.TrimSuffix(r.URL.Path[1:], "/"+format)) // Hack
+	if err != nil {
+		return err
+	}
 	since, err := parseSince(r)
 	if err != nil {
 		return err
@@ -270,6 +272,7 @@ func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request, format
 		case <-r.Context().Done():
 			return nil
 		case <-time.After(s.config.KeepaliveInterval):
+			v.Keepalive()
 			if err := sub(newKeepaliveMessage(t.id)); err != nil { // Send keepalive message
 				return err
 			}
@@ -308,16 +311,19 @@ func (s *Server) handleOptions(w http.ResponseWriter, r *http.Request) error {
 	return nil
 }
 
-func (s *Server) createTopic(id string) *topic {
+func (s *Server) topic(id string) (*topic, error) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if _, ok := s.topics[id]; !ok {
+		if len(s.topics) >= s.config.GlobalTopicLimit {
+			return nil, errHTTPTooManyRequests
+		}
 		s.topics[id] = newTopic(id)
 		if s.firebase != nil {
 			s.topics[id].Subscribe(s.firebase)
 		}
 	}
-	return s.topics[id]
+	return s.topics[id], nil
 }
 
 func (s *Server) updateStatsAndExpire() {
@@ -326,16 +332,16 @@ func (s *Server) updateStatsAndExpire() {
 
 	// Expire visitors from rate visitors map
 	for ip, v := range s.visitors {
-		if time.Since(v.seen) > visitorExpungeAfter {
+		if v.Stale() {
 			delete(s.visitors, ip)
 		}
 	}
 
-	// Prune old messages, remove topics without subscribers
+	// Prune old messages, remove subscriptions without subscribers
 	for _, t := range s.topics {
 		t.Prune(s.config.MessageBufferDuration)
 		subs, msgs := t.Stats()
-		if msgs == 0 && (subs == 0 || (s.firebase != nil && subs == 1)) {
+		if msgs == 0 && (subs == 0 || (s.firebase != nil && subs == 1)) { // Firebase is a subscriber!
 			delete(s.topics, t.id)
 		}
 	}
@@ -362,12 +368,8 @@ func (s *Server) visitor(remoteAddr string) *visitor {
 	}
 	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
+		s.visitors[ip] = newVisitor(s.config)
+		return s.visitors[ip]
 	}
 	v.seen = time.Now()
 	return v

server/static/css/app.css

@@ -6,12 +6,13 @@ html, body {
     font-size: 1.1em;
 }
 
-a {
-    color: #39005a;
+a, a:visited {
+    color: #3a9784;
 }
 
 a:hover {
     text-decoration: none;
+    color: #317f6f;
 }
 
 h1 {
@@ -20,7 +21,6 @@ h1 {
     font-size: 2.5em;
 }
 
-
 h2 {
     margin-top: 20px;
     margin-bottom: 5px;

server/visitor.go

@@ -0,0 +1,65 @@
+package server
+
+import (
+	"golang.org/x/time/rate"
+	"heckel.io/ntfy/config"
+	"heckel.io/ntfy/util"
+	"sync"
+	"time"
+)
+
+const (
+	visitorExpungeAfter = 30 * time.Minute
+)
+
+// visitor represents an API user, and its associated rate.Limiter used for rate limiting
+type visitor struct {
+	config        *config.Config
+	limiter       *rate.Limiter
+	subscriptions *util.Limiter
+	seen          time.Time
+	mu            sync.Mutex
+}
+
+func newVisitor(conf *config.Config) *visitor {
+	return &visitor{
+		config:        conf,
+		limiter:       rate.NewLimiter(conf.VisitorRequestLimit, conf.VisitorRequestLimitBurst),
+		subscriptions: util.NewLimiter(int64(conf.VisitorSubscriptionLimit)),
+		seen:          time.Now(),
+	}
+}
+
+func (v *visitor) RequestAllowed() error {
+	if !v.limiter.Allow() {
+		return errHTTPTooManyRequests
+	}
+	return nil
+}
+
+func (v *visitor) AddSubscription() error {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	if err := v.subscriptions.Add(1); err != nil {
+		return errHTTPTooManyRequests
+	}
+	return nil
+}
+
+func (v *visitor) RemoveSubscription() {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	v.subscriptions.Sub(1)
+}
+
+func (v *visitor) Keepalive() {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	v.seen = time.Now()
+}
+
+func (v *visitor) Stale() bool {
+	v.mu.Lock()
+	defer v.mu.Unlock()
+	return time.Since(v.seen) > visitorExpungeAfter
+}

util/limit.go

@@ -0,0 +1,65 @@
+package util
+
+import (
+	"errors"
+	"sync"
+)
+
+// ErrLimitReached is the error returned by the Limiter and LimitWriter when the predefined limit has been reached
+var ErrLimitReached = errors.New("limit reached")
+
+// Limiter is a helper that allows adding values up to a well-defined limit. Once the limit is reached
+// ErrLimitReached will be returned. Limiter may be used by multiple goroutines.
+type Limiter struct {
+	value int64
+	limit int64
+	mu    sync.Mutex
+}
+
+// NewLimiter creates a new Limiter
+func NewLimiter(limit int64) *Limiter {
+	return &Limiter{
+		limit: limit,
+	}
+}
+
+// Add adds n to the limiters internal value, but only if the limit has not been reached. If the limit would be
+// exceeded after adding n, ErrLimitReached is returned.
+func (l *Limiter) Add(n int64) error {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	if l.limit == 0 {
+		l.value += n
+		return nil
+	} else if l.value+n <= l.limit {
+		l.value += n
+		return nil
+	} else {
+		return ErrLimitReached
+	}
+}
+
+// Sub subtracts a value from the limiters internal value
+func (l *Limiter) Sub(n int64) {
+	l.Add(-n)
+}
+
+// Set sets the value of the limiter to n. This function ignores the limit. It is meant to set the value
+// based on reality.
+func (l *Limiter) Set(n int64) {
+	l.mu.Lock()
+	l.value = n
+	l.mu.Unlock()
+}
+
+// Value returns the internal value of the limiter
+func (l *Limiter) Value() int64 {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return l.value
+}
+
+// Limit returns the defined limit
+func (l *Limiter) Limit() int64 {
+	return l.limit
+}