Merge remote-tracking branch 'origin/main' into shared-daily-limits

# Conflicts:
#	docs/releases.md
#	server/server.go
This commit is contained in:
binwiederhier committed 2026-10-31 03:14:26 +00:00
commit 0f3a57a2a2
43 files changed
+2175 -176

No files matched your search

+8 -2
View File
@@ -2,9 +2,15 @@
package cmd
import "syscall"
import (
"errors"
"syscall"
)
func processExists(pid int) bool {
err := syscall.Kill(pid, syscall.Signal(0))
return err == nil
if err == nil {
return true
}
return errors.Is(err, syscall.EPERM)
}
+24
View File
@@ -0,0 +1,24 @@
//go:build darwin || linux || dragonfly || freebsd || netbsd || openbsd
package cmd
import (
"os"
"os/exec"
"testing"
"github.com/stretchr/testify/require"
)
func TestProcessExists_OtherUser(t *testing.T) {
if os.Geteuid() == 0 {
t.Skip("root can signal every process")
}
require.True(t, processExists(1)) // PID 1 is owned by root, so kill(1, 0) returns EPERM
}
func TestProcessExists_Exited(t *testing.T) {
cmd := exec.Command("true")
require.NoError(t, cmd.Run())
require.False(t, processExists(cmd.Process.Pid))
}
+22 -15
View File
@@ -39,6 +39,7 @@ var flagsServe = append(
altsrc.NewStringFlag(&cli.StringFlag{Name: "listen-https", Aliases: []string{"listen_https", "L"}, EnvVars: []string{"NTFY_LISTEN_HTTPS"}, Usage: "ip:port used as HTTPS listen address"}),
altsrc.NewStringFlag(&cli.StringFlag{Name: "listen-unix", Aliases: []string{"listen_unix", "U"}, EnvVars: []string{"NTFY_LISTEN_UNIX"}, Usage: "listen on unix socket path"}),
altsrc.NewIntFlag(&cli.IntFlag{Name: "listen-unix-mode", Aliases: []string{"listen_unix_mode"}, EnvVars: []string{"NTFY_LISTEN_UNIX_MODE"}, DefaultText: "system default", Usage: "file permissions of unix socket, e.g. 0700"}),
altsrc.NewStringFlag(&cli.StringFlag{Name: "listen-admin-unix", Aliases: []string{"listen_admin_unix"}, EnvVars: []string{"NTFY_LISTEN_ADMIN_UNIX"}, Value: server.DefaultListenAdminUnix(), DefaultText: "/run/ntfy/admin.sock (empty on Windows)", Usage: "unix socket path for credential-less local admin access (ntfy user/access/token/tier); \"-\" disables it"}),
altsrc.NewStringFlag(&cli.StringFlag{Name: "key-file", Aliases: []string{"key_file", "K"}, EnvVars: []string{"NTFY_KEY_FILE"}, Usage: "private key file, if listen-https is set"}),
altsrc.NewStringFlag(&cli.StringFlag{Name: "cert-file", Aliases: []string{"cert_file", "E"}, EnvVars: []string{"NTFY_CERT_FILE"}, Usage: "certificate file, if listen-https is set"}),
altsrc.NewStringFlag(&cli.StringFlag{Name: "firebase-key-file", Aliases: []string{"firebase_key_file", "F"}, EnvVars: []string{"NTFY_FIREBASE_KEY_FILE"}, Usage: "Firebase credentials file; if set additionally publish to FCM topic"}),
@@ -161,6 +162,7 @@ func execServe(c *cli.Context) error {
listenHTTPS := c.String("listen-https")
listenUnix := c.String("listen-unix")
listenUnixMode := c.Int("listen-unix-mode")
listenAdminUnix := c.String("listen-admin-unix")
keyFile := c.String("key-file")
certFile := c.String("cert-file")
firebaseKeyFile := c.String("firebase-key-file")
@@ -348,7 +350,17 @@ func execServe(c *cli.Context) error {
return fmt.Errorf("config option visitor-attachment-daily-bandwidth-limit must be lower than %d", math.MaxInt)
}
// Check values
// Check values; checks that don't return stay out of the else-if chain, which would end there
if baseURL != "" {
u, err := url.Parse(baseURL)
if err != nil {
return fmt.Errorf("if set, base-url must be a valid URL, e.g. https://ntfy.mydomain.com: %v", err)
} else if u.Scheme != "http" && u.Scheme != "https" {
return errors.New("if set, base-url must be a valid URL starting with http:// or https://, e.g. https://ntfy.mydomain.com")
} else if u.Path != "" {
return fmt.Errorf("if set, base-url must not have a path (%s), as hosting ntfy on a sub-path is not supported, e.g. https://ntfy.mydomain.com", u.Path)
}
}
if databaseURL != "" && !strings.HasPrefix(databaseURL, "postgres://") && !strings.HasPrefix(databaseURL, "postgresql://") {
return errors.New("if database-url is set, it must start with postgres:// or postgresql://")
} else if databaseURL != "" && (authFile != "" || cacheFile != "" || webPushFile != "") {
@@ -391,15 +403,6 @@ func execServe(c *cli.Context) error {
return errors.New("if smtp-server-listen is set, smtp-server-domain must also be set")
} else if attachmentCacheDir != "" && baseURL == "" {
return errors.New("if attachment-cache-dir is set, base-url must also be set")
} else if baseURL != "" {
u, err := url.Parse(baseURL)
if err != nil {
return fmt.Errorf("if set, base-url must be a valid URL, e.g. https://ntfy.mydomain.com: %v", err)
} else if u.Scheme != "http" && u.Scheme != "https" {
return errors.New("if set, base-url must be a valid URL starting with http:// or https://, e.g. https://ntfy.mydomain.com")
} else if u.Path != "" {
return fmt.Errorf("if set, base-url must not have a path (%s), as hosting ntfy on a sub-path is not supported, e.g. https://ntfy.mydomain.com", u.Path)
}
} else if upstreamBaseURL != "" && !strings.HasPrefix(upstreamBaseURL, "http://") && !strings.HasPrefix(upstreamBaseURL, "https://") {
return errors.New("if set, upstream-base-url must start with http:// or https://")
} else if upstreamBaseURL != "" && strings.HasSuffix(upstreamBaseURL, "/") {
@@ -420,11 +423,8 @@ func execServe(c *cli.Context) error {
return errors.New("if stripe-secret-key is set, stripe-webhook-key and base-url must also be set")
} else if twilioAccount != "" && (twilioAuthToken == "" || twilioPhoneNumber == "" || twilioVerifyService == "" || baseURL == "" || (authFile == "" && databaseURL == "")) {
return errors.New("if twilio-account is set, twilio-auth-token, twilio-phone-number, twilio-verify-service, base-url, and auth-file (or database-url) must also be set")
} else if messageSizeLimit > server.DefaultMessageSizeLimit {
log.Warn("message-size-limit is greater than 4K, this is not recommended and largely untested, and may lead to issues with some clients")
if messageSizeLimit > 5*1024*1024 {
return errors.New("message-size-limit cannot be higher than 5M")
}
} else if messageSizeLimit > 5*1024*1024 {
return errors.New("message-size-limit cannot be higher than 5M")
} else if !server.WebPushAvailable && (webPushPrivateKey != "" || webPushPublicKey != "" || webPushFile != "") {
return errors.New("cannot enable WebPush, support is not available in this build (nowebpush)")
} else if webPushExpiryWarningDuration > 0 && webPushExpiryWarningDuration > webPushExpiryDuration {
@@ -446,6 +446,9 @@ func execServe(c *cli.Context) error {
} else if runtime.GOOS == "windows" && listenUnix != "" {
return errors.New("listen-unix is not supported on Windows")
}
if messageSizeLimit > server.DefaultMessageSizeLimit {
log.Warn("message-size-limit is greater than 4K, this is not recommended and largely untested, and may lead to issues with some clients")
}
if cacheDuration > 0 && attachmentExpiryDuration > cacheDuration {
log.Warn("attachment-expiry-duration (%s) is longer than cache-duration (%s); attachments never outlive their message, so they will be deleted after %s", attachmentExpiryDurationStr, cacheDurationStr, cacheDurationStr)
}
@@ -483,6 +486,9 @@ func execServe(c *cli.Context) error {
if err != nil {
return err
}
if listenAdminUnix == "-" {
listenAdminUnix = "" // Unsets the default, like listen-http
}
// Resolve hosts
visitorRequestLimitExemptPrefixes := make([]netip.Prefix, 0)
@@ -530,6 +536,7 @@ func execServe(c *cli.Context) error {
conf.ListenHTTPS = listenHTTPS
conf.ListenUnix = listenUnix
conf.ListenUnixMode = fs.FileMode(listenUnixMode)
conf.ListenAdminUnix = listenAdminUnix
conf.KeyFile = keyFile
conf.CertFile = certFile
conf.FirebaseKeyFile = firebaseKeyFile
+30
View File
@@ -491,6 +491,23 @@ func TestCLI_Serve_Unix_Curl(t *testing.T) {
require.Equal(t, "this is a message", m.Message)
}
func TestCLI_Serve_ListenAdminUnixDisabled(t *testing.T) {
dir := t.TempDir()
t.Chdir(dir) // A relative "-" socket would land here
sockFile := filepath.Join(dir, "ntfy.sock")
configFile := newEmptyFile(t)
go func() {
app, _, _, _ := newTestApp()
err := app.Run([]string{"ntfy", "serve", "--config=" + configFile, "--listen-http=-", "--listen-unix=" + sockFile, "--auth-file=" + filepath.Join(dir, "user.db"), "--listen-admin-unix=-"})
require.Nil(t, err)
}()
for i := 0; i < 40 && !util.FileExists(sockFile); i++ {
time.Sleep(50 * time.Millisecond)
}
require.True(t, util.FileExists(sockFile))
require.False(t, util.FileExists(filepath.Join(dir, "-")))
}
func TestCLI_Serve_WebSocket(t *testing.T) {
port := 10000 + rand.Intn(20000)
go func() {
@@ -691,6 +708,19 @@ func TestCLI_Serve_AttachmentExpiryLongerThanCacheDurationWarning(t *testing.T)
require.NotContains(t, buf.String(), "attachment-expiry-duration")
}
func TestCLI_Serve_ValidationAfterBaseURLAndMessageSizeLimit(t *testing.T) {
configFile := newEmptyFile(t) // Avoid issues with existing server.yml file on system
// A valid base-url used to end the checks, so the ones after it never ran; an invalid
// auth-default-access fails right after the checks, so the server never starts
app, _, _, _ := newTestApp()
err := app.Run([]string{"ntfy", "serve", "--config=" + configFile, "--auth-default-access=invalid", "--base-url=https://ntfy.example.com", "--upstream-base-url=https://ntfy.example.com"})
require.ErrorContains(t, err, "base-url and upstream-base-url cannot be identical")
// The same for a message-size-limit above 4K
app, _, _, _ = newTestApp()
err = app.Run([]string{"ntfy", "serve", "--config=" + configFile, "--auth-default-access=invalid", "--message-size-limit=8k", "--visitor-prefix-bits-ipv4=0"})
require.ErrorContains(t, err, "visitor-prefix-bits-ipv4 must be between 1 and 32")
}
func newEmptyFile(t *testing.T) string {
filename := filepath.Join(t.TempDir(), "empty")
require.Nil(t, os.WriteFile(filename, []byte{}, 0600))
+42 -33
View File
@@ -10,10 +10,10 @@ import (
)
const (
tag = "db"
replicaHealthCheckInitialDelay = 5 * time.Second
replicaHealthCheckInterval = 30 * time.Second
replicaHealthCheckTimeout = 10 * time.Second
tag = "db"
healthCheckInitialDelay = 5 * time.Second
healthCheckInterval = 30 * time.Second
healthCheckTimeout = 10 * time.Second
)
// DB wraps a primary *sql.DB and optional read replicas. All standard query/exec methods
@@ -28,7 +28,8 @@ type DB struct {
// New creates a new DB that wraps the given primary and optional replica connections.
// If replicas is nil or empty, ReadOnly() simply returns the primary.
// Replicas start unhealthy and are checked immediately by a background goroutine.
// Replicas start unhealthy, the primary healthy (it was just opened); a background goroutine
// checks all of them, so the metrics show a node that is cut off from its database.
func New(primary *Host, replicas []*Host) *DB {
ctx, cancel := context.WithCancel(context.Background())
d := &DB{
@@ -36,9 +37,8 @@ func New(primary *Host, replicas []*Host) *DB {
replicas: replicas,
cancel: cancel,
}
if len(d.replicas) > 0 {
go d.healthCheckLoop(ctx)
}
primary.healthy.Store(true)
go d.healthCheckLoop(ctx)
return d
}
@@ -92,16 +92,18 @@ func (d *DB) Primary() *sql.DB {
func (d *DB) Stats() []*PoolStats {
stats := []*PoolStats{
{
Role: PoolRolePrimary,
Addr: d.primary.Addr,
Stats: d.primary.DB.Stats(),
Role: PoolRolePrimary,
Addr: d.primary.Addr,
Healthy: d.primary.healthy.Load(),
Stats: d.primary.DB.Stats(),
},
}
for _, r := range d.replicas {
stats = append(stats, &PoolStats{
Role: PoolRoleReplica,
Addr: r.Addr,
Stats: r.DB.Stats(),
Role: PoolRoleReplica,
Addr: r.Addr,
Healthy: r.healthy.Load(),
Stats: r.DB.Stats(),
})
}
return stats
@@ -142,39 +144,46 @@ func (d *DB) Close() error {
return d.primary.DB.Close()
}
// healthCheckLoop checks replicas immediately, then periodically on a ticker.
// healthCheckLoop checks the primary and the replicas shortly after start, then periodically.
func (d *DB) healthCheckLoop(ctx context.Context) {
select {
case <-ctx.Done():
return
case <-time.After(replicaHealthCheckInitialDelay):
d.checkReplicas(ctx)
case <-time.After(healthCheckInitialDelay):
d.checkHealth(ctx)
}
for {
select {
case <-ctx.Done():
return
case <-time.After(replicaHealthCheckInterval):
d.checkReplicas(ctx)
case <-time.After(healthCheckInterval):
d.checkHealth(ctx)
}
}
}
// checkReplicas pings each replica with a timeout and updates its health status.
func (d *DB) checkReplicas(ctx context.Context) {
// checkHealth pings the primary and each replica with a timeout and updates their health status.
func (d *DB) checkHealth(ctx context.Context) {
checkHost(ctx, d.primary, "primary")
for _, r := range d.replicas {
wasHealthy := r.healthy.Load()
pingCtx, cancel := context.WithTimeout(ctx, replicaHealthCheckTimeout)
err := r.DB.PingContext(pingCtx)
cancel()
if err != nil {
r.healthy.Store(false)
log.Tag(tag).Error("Database replica %s is unhealthy: %s", r.Addr, err)
} else {
r.healthy.Store(true)
if !wasHealthy {
log.Tag(tag).Info("Database replica %s is healthy", r.Addr)
}
checkHost(ctx, r, "replica")
}
}
// checkHost pings the host and updates its health status, logging when it changes (and every
// failure, so a lasting outage stays visible in the logs)
func checkHost(ctx context.Context, h *Host, kind string) {
wasHealthy := h.healthy.Load()
pingCtx, cancel := context.WithTimeout(ctx, healthCheckTimeout)
err := h.DB.PingContext(pingCtx)
cancel()
if err != nil {
h.healthy.Store(false)
log.Tag(tag).Error("Database %s %s is unhealthy: %s", kind, h.Addr, err)
} else {
h.healthy.Store(true)
if !wasHealthy {
log.Tag(tag).Info("Database %s %s is healthy", kind, h.Addr)
}
}
}
+19
View File
@@ -5,6 +5,7 @@ import (
"database/sql"
"path/filepath"
"testing"
"time"
_ "github.com/mattn/go-sqlite3"
"github.com/stretchr/testify/require"
@@ -74,3 +75,21 @@ func TestDB_Stats(t *testing.T) {
require.Equal(t, 3, stats[1].Stats.MaxOpenConnections)
require.Equal(t, 0, stats[1].Stats.InUse)
}
func TestDB_HealthCheck(t *testing.T) {
// The primary is checked like the replicas, so a node cut off from it shows as unhealthy
// (ntfy_db_up), and healthy again once it is back
path := filepath.Join(t.TempDir(), "primary.db")
primary, err := sql.Open("sqlite3", path)
require.Nil(t, err)
d := db.New(&db.Host{
Addr: "primary:5432",
DB: primary,
}, nil)
defer d.Close()
require.True(t, d.Stats()[0].Healthy) // Just opened: healthy until a check says otherwise
require.Nil(t, primary.Close())
require.Eventually(t, func() bool {
return !d.Stats()[0].Healthy
}, 15*time.Second, 50*time.Millisecond)
}
+4 -3
View File
@@ -32,7 +32,8 @@ const (
// PoolStats is one connection pool's statistics: the primary's or a replica's
type PoolStats struct {
Role string // PoolRolePrimary or PoolRoleReplica
Addr string // "host:port"
Stats sql.DBStats
Role string // PoolRolePrimary or PoolRoleReplica
Addr string // "host:port"
Healthy bool // The last health check (a ping) succeeded
Stats sql.DBStats
}
+29
View File
@@ -500,6 +500,10 @@ To size the pools, watch the `ntfy_db_pool_*` [metrics](#monitoring), one series
and `ntfy_db_pool_wait_seconds_total` count queries that had to wait for a free connection. The open connection count
says little, since the pool rotates through all of them. Waits that keep going up mean the pool is too small.
`ntfy_db_up` is 1 if the server reached the pool's database at its last health check (a ping every 30 seconds), and 0
if not, one series per pool like the above. Alert on `ntfy_db_up{role="primary"} == 0`: a server that cannot reach its
primary cannot publish or authenticate, and enforces the daily limits on its own until it is back.
## Message cache
If desired, ntfy can temporarily keep notifications in an in-memory or an on-disk cache. Caching messages for a short period
@@ -914,6 +918,29 @@ In this example, the `auth-users` section defines two users, `phil` and `backup-
defines access tokens for these users. `phil` has a token `tk_3gd7d2yftt4b8ixyfe9mnmro88o76`, while `backup-service`
has a token `tk_f099we8uzj7xi5qshzajwp6jffvkz` with the label "Backup script".
### Admin socket
The `ntfy user`, `ntfy access`, `ntfy token` and `ntfy tier` commands talk to the running ntfy server through a local
admin socket, by default `/run/ntfy/admin.sock` (`/var/run/ntfy/admin.sock` on macOS). That way, changes take effect
immediately: a removed user is disconnected right away, a new tier applies to the user's next request, and there is no
delay until the server reloads its access control list.
The socket needs no password or token: **being able to open the socket is the authorization**. It is created with
mode `0660` in a directory with mode `0750`, both owned by the user ntfy runs as, so only that user, members of its group
and `root` can use it. With the Debian/Ubuntu/Fedora packages, the shipped systemd unit creates `/run/ntfy` (via
`RuntimeDirectory=ntfy`), so `sudo ntfy user add ...` and members of the `ntfy` group just work.
The socket is only opened if [access control](#access-control) is configured (`auth-file` or `database-url`). If the
server cannot create the socket, it logs a warning at startup and runs without it:
```
WARN Admin socket /run/ntfy/admin.sock not available, 'ntfy user/access/token/tier' will write to the database directly, which is deprecated, see https://ntfy.sh/docs/deprecations/ (error=mkdir /run/ntfy: permission denied, tag=startup)
```
To change the path, set `listen-admin-unix` in the `server.yml` file; the commands read it from there, too. To disable the
socket, set `listen-admin-unix: "-"`. The admin socket is not available on Windows. If you change the path, pick a
directory that only the ntfy user can write to; otherwise the server logs a warning at startup, since other users may be
able to use or replace the socket.
### Example: Private instance
The easiest way to configure a private instance is to set `auth-default-access` to `deny-all` in the `server.yml`,
and to configure users in the `auth-users` section (see [users via the config](#users-via-the-config)),
@@ -2529,6 +2556,7 @@ variable before running the `ntfy` command (e.g. `export NTFY_LISTEN_HTTP=:80`).
| `listen-https` | `NTFY_LISTEN_HTTPS` | `[host]:port` | - | Listen address for the HTTPS web server. If set, you also need to set `key-file` and `cert-file`. |
| `listen-unix` | `NTFY_LISTEN_UNIX` | *filename* | - | Path to a Unix socket to listen on |
| `listen-unix-mode` | `NTFY_LISTEN_UNIX_MODE` | *file mode* | *system default* | File mode of the Unix socket, e.g. 0700 or 0777 |
| `listen-admin-unix` | `NTFY_LISTEN_ADMIN_UNIX` | *filename* | `/run/ntfy/admin.sock` | Path of the local [admin socket](#admin-socket) the `ntfy user/access/token/tier` commands use; `"-"` disables it. Only opened if auth is configured; not available on Windows |
| `key-file` | `NTFY_KEY_FILE` | *filename* | - | HTTPS/TLS private key file, only used if `listen-https` is set. |
| `cert-file` | `NTFY_CERT_FILE` | *filename* | - | HTTPS/TLS certificate file, only used if `listen-https` is set. |
| `firebase-key-file` | `NTFY_FIREBASE_KEY_FILE` | *filename* | - | If set, also publish messages to a Firebase Cloud Messaging (FCM) topic for your app. This is optional and only required to save battery when using the Android app. See [Firebase (FCM)](#firebase-fcm). |
@@ -2651,6 +2679,7 @@ OPTIONS:
--listen-https value, --listen_https value, -L value ip:port used as HTTPS listen address [$NTFY_LISTEN_HTTPS]
--listen-unix value, --listen_unix value, -U value listen on unix socket path [$NTFY_LISTEN_UNIX]
--listen-unix-mode value, --listen_unix_mode value file permissions of unix socket, e.g. 0700 (default: system default) [$NTFY_LISTEN_UNIX_MODE]
--listen-admin-unix value, --listen_admin_unix value unix socket path for credential-less local admin access (ntfy user/access/token/tier); "-" disables it (default: /run/ntfy/admin.sock (empty on Windows)) [$NTFY_LISTEN_ADMIN_UNIX]
--key-file value, --key_file value, -K value private key file, if listen-https is set [$NTFY_KEY_FILE]
--cert-file value, --cert_file value, -E value certificate file, if listen-https is set [$NTFY_CERT_FILE]
--firebase-key-file value, --firebase_key_file value, -F value Firebase credentials file; if set additionally publish to FCM topic [$NTFY_FIREBASE_KEY_FILE]
+11
View File
@@ -2293,6 +2293,9 @@ and the [ntfy Android app](https://github.com/binwiederhier/ntfy-android/release
**Features:**
* Daily limits (messages, e-mails, phone calls) are kept in the user database (`auth-file` for SQLite, or `database-url` for PostgreSQL) and survive restarts; with PostgreSQL, all servers share them, so a visitor gets one daily allowance in total, not one per server (see [message limits](config.md#message-limits))
* Local admin socket (`listen-admin-unix`, default `/run/ntfy/admin.sock`) that serves the admin API without credentials to whoever can open it (the ntfy user and group, and root); the shipped systemd unit now sets `RuntimeDirectory=ntfy` so the server can create it. Custom unit files without it log a warning at startup and keep working as before (see [admin socket](config.md#admin-socket))
* Admin API (internal, undocumented): manage tokens of other users, tiers and password reset links
* Export `ntfy_db_up` (1 if the last health check of a PostgreSQL pool's database succeeded, per pool) and health-check the primary like the replicas, to alert on a server that is cut off from its database (see [PostgreSQL](config.md#postgresql))
**Behavior changes:**
@@ -2300,6 +2303,14 @@ and the [ntfy Android app](https://github.com/binwiederhier/ntfy-android/release
* Visitors without a tier have a daily e-mail limit again (24h / `visitor-email-limit-replenish`), now that it is shared between servers and survives restarts
* The daily reset job is gone; the `stats_*` columns of the user table are zeroed by the upgrade and no longer read or written
**Bug fixes + maintenance:**
* Admin API: return 4xx errors instead of a 500 for invalid input, and remove a deleted user's web push subscriptions
* Web app: Deleted notifications no longer come back when a poll replays messages published in the same second as the last one ([#2098](https://github.com/binwiederhier/ntfy/issues/2098), thanks to [@mattford63](https://github.com/mattford63) and [@arminus](https://github.com/arminus) for reporting)
* Web app: A message replayed after a reconnect keeps its read state and no longer notifies again, and an older version of a sequence no longer replaces a newer one ([#2090](https://github.com/binwiederhier/ntfy/issues/2090), thanks to [@tobykaspar79](https://github.com/tobykaspar79) for reporting)
* Check the whole server config again when `base-url` or a `message-size-limit` above 4K is set; most checks after them were skipped, so an invalid config (e.g. `upstream-base-url` equal to `base-url`) now fails to start ([#1969](https://github.com/binwiederhier/ntfy/pull/1969), thanks to [@mikejgray](https://github.com/mikejgray) for the contribution)
* CLI: `ntfy pub --wait-pid` now waits for processes owned by other users instead of failing with "not running" ([#1710](https://github.com/binwiederhier/ntfy/pull/1710), thanks to [@bekoeppel](https://github.com/bekoeppel) for the contribution)
### ntfy iOS app v1.8.0 (UNRELEASED)
**Features:**
+1 -1
View File
@@ -111,7 +111,7 @@ Or, if you already started the long-running process and want to wait for it usin
=== "Using a `pidof`"
```
$ ntfy pub --wait-pid $(pidof rsync) mytopic | jq .
$ ntfy pub --wait-pid $(pidof -s rsync) mytopic | jq .
{
"id": "orM6hJKNYkWb",
"time": 1655825827,
+1 -1
View File
@@ -59,7 +59,7 @@ require (
github.com/envoyproxy/go-control-plane/envoy v1.39.0 // indirect
github.com/envoyproxy/protoc-gen-validate v1.3.3 // indirect
github.com/felixge/httpsnoop v1.1.0 // indirect
github.com/go-jose/go-jose/v4 v4.1.5 // indirect
github.com/go-jose/go-jose/v4 v4.1.6 // indirect
github.com/go-logr/logr v1.4.4 // indirect
github.com/go-logr/stdr v1.2.2 // indirect
github.com/golang-jwt/jwt/v4 v4.5.2 // indirect
+2 -2
View File
@@ -67,8 +67,8 @@ github.com/felixge/httpsnoop v1.1.0 h1:3YtUj32ZZkqZtt3sZZsClsymw/QDuVfpNhoA31zeO
github.com/felixge/httpsnoop v1.1.0/go.mod h1:Zqxgdd+1Rkcz8euOqdr7lqgCRJztwr5hp9vDSi5UZCE=
github.com/gabriel-vasile/mimetype v1.4.16 h1:5izNFuvA37st1dDRAcA7TtmngBmRgiIPDmTtiEvjx1E=
github.com/gabriel-vasile/mimetype v1.4.16/go.mod h1:3oFFwxX8kTklZwzIl2aQEDZT0lKY8xH92ST88bAwba4=
github.com/go-jose/go-jose/v4 v4.1.5 h1:RjgjO2LOtWOJKUC5wpwY9LR3B3vwVAz6JS2YHfYU6eA=
github.com/go-jose/go-jose/v4 v4.1.5/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08=
github.com/go-jose/go-jose/v4 v4.1.6 h1:sNRkftxFJxwsILiQukqp5z83QBSmntVwmgRY7AqBDO4=
github.com/go-jose/go-jose/v4 v4.1.6/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08=
github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
github.com/go-logr/logr v1.4.4 h1:tG4xh9yMsRCAiodLVTxyrkzSZ9+o0L1Kg/+cPVcbP/8=
github.com/go-logr/logr v1.4.4/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+24 -17
View File
@@ -167,10 +167,11 @@ func init() {
type DBPool struct {
Role string // db.PoolRolePrimary or db.PoolRoleReplica
Addr string // "host:port"
Up bool // The last health check of the pool's database succeeded
Stats sql.DBStats
}
// SetDBPoolsFunc sets the function that the ntfy_db_pool_* metrics read their values from on
// SetDBPoolsFunc sets the function that the ntfy_db_* metrics read their values from on
// every scrape; nil (the default) reports no pools
func SetDBPoolsFunc(fn func() []*DBPool) {
dbPools.fn.Store(&fn)
@@ -184,39 +185,45 @@ type dbPoolCollector struct {
metrics []*dbPoolMetric
}
// dbPoolMetric is one ntfy_db_pool_* metric: its description, and how to read it from a pool
// dbPoolMetric is one ntfy_db_* metric: its description, and how to read it from a pool
type dbPoolMetric struct {
desc *prometheus.Desc
valueType prometheus.ValueType
value func(stats sql.DBStats) float64
value func(pool *DBPool) float64
}
func newDBPoolCollector() *dbPoolCollector {
return &dbPoolCollector{
metrics: []*dbPoolMetric{
newDBPoolMetric("ntfy_db_pool_connections_max", "Maximum open database connections (pool_max_conns), per pool", prometheus.GaugeValue, func(stats sql.DBStats) float64 {
return float64(stats.MaxOpenConnections)
newDBPoolMetric("ntfy_db_up", "1 if the last health check of the pool's database (a ping every 30s) succeeded, 0 if not, per pool", prometheus.GaugeValue, func(pool *DBPool) float64 {
if pool.Up {
return 1
}
return 0
}),
newDBPoolMetric("ntfy_db_pool_connections_open", "Open database connections, in use or idle, per pool", prometheus.GaugeValue, func(stats sql.DBStats) float64 {
return float64(stats.OpenConnections)
newDBPoolMetric("ntfy_db_pool_connections_max", "Maximum open database connections (pool_max_conns), per pool", prometheus.GaugeValue, func(pool *DBPool) float64 {
return float64(pool.Stats.MaxOpenConnections)
}),
newDBPoolMetric("ntfy_db_pool_connections_in_use", "Database connections currently in use, per pool", prometheus.GaugeValue, func(stats sql.DBStats) float64 {
return float64(stats.InUse)
newDBPoolMetric("ntfy_db_pool_connections_open", "Open database connections, in use or idle, per pool", prometheus.GaugeValue, func(pool *DBPool) float64 {
return float64(pool.Stats.OpenConnections)
}),
newDBPoolMetric("ntfy_db_pool_connections_idle", "Idle database connections, per pool", prometheus.GaugeValue, func(stats sql.DBStats) float64 {
return float64(stats.Idle)
newDBPoolMetric("ntfy_db_pool_connections_in_use", "Database connections currently in use, per pool", prometheus.GaugeValue, func(pool *DBPool) float64 {
return float64(pool.Stats.InUse)
}),
newDBPoolMetric("ntfy_db_pool_waits_total", "Times a query had to wait for a free database connection, per pool", prometheus.CounterValue, func(stats sql.DBStats) float64 {
return float64(stats.WaitCount)
newDBPoolMetric("ntfy_db_pool_connections_idle", "Idle database connections, per pool", prometheus.GaugeValue, func(pool *DBPool) float64 {
return float64(pool.Stats.Idle)
}),
newDBPoolMetric("ntfy_db_pool_wait_seconds_total", "Total time spent waiting for a free database connection, per pool", prometheus.CounterValue, func(stats sql.DBStats) float64 {
return stats.WaitDuration.Seconds()
newDBPoolMetric("ntfy_db_pool_waits_total", "Times a query had to wait for a free database connection, per pool", prometheus.CounterValue, func(pool *DBPool) float64 {
return float64(pool.Stats.WaitCount)
}),
newDBPoolMetric("ntfy_db_pool_wait_seconds_total", "Total time spent waiting for a free database connection, per pool", prometheus.CounterValue, func(pool *DBPool) float64 {
return pool.Stats.WaitDuration.Seconds()
}),
},
}
}
func newDBPoolMetric(name, help string, valueType prometheus.ValueType, value func(stats sql.DBStats) float64) *dbPoolMetric {
func newDBPoolMetric(name, help string, valueType prometheus.ValueType, value func(pool *DBPool) float64) *dbPoolMetric {
return &dbPoolMetric{
desc: prometheus.NewDesc(name, help, []string{"role", "host"}, nil),
valueType: valueType,
@@ -239,7 +246,7 @@ func (c *dbPoolCollector) Collect(ch chan<- prometheus.Metric) {
}
for _, pool := range (*fn)() {
for _, m := range c.metrics {
ch <- prometheus.MustNewConstMetric(m.desc, m.valueType, m.value(pool.Stats), pool.Role, pool.Addr)
ch <- prometheus.MustNewConstMetric(m.desc, m.valueType, m.value(pool), pool.Role, pool.Addr)
}
}
}
+7 -1
View File
@@ -36,6 +36,7 @@ var expectedMetricNames = []string{
"ntfy_db_pool_connections_open",
"ntfy_db_pool_wait_seconds_total",
"ntfy_db_pool_waits_total",
"ntfy_db_up",
"ntfy_emails_received_failure",
"ntfy_emails_received_success",
"ntfy_emails_sent_failure",
@@ -101,6 +102,7 @@ func TestDBPoolMetrics(t *testing.T) {
{
Role: db.PoolRolePrimary,
Addr: "db:5432",
Up: true,
Stats: sql.DBStats{
MaxOpenConnections: 30,
OpenConnections: 12,
@@ -140,9 +142,13 @@ ntfy_db_pool_wait_seconds_total{host="replica:5432",role="%[2]s"} 0
# TYPE ntfy_db_pool_waits_total counter
ntfy_db_pool_waits_total{host="db:5432",role="%[1]s"} 3
ntfy_db_pool_waits_total{host="replica:5432",role="%[2]s"} 0
# HELP ntfy_db_up 1 if the last health check of the pool's database (a ping every 30s) succeeded, 0 if not, per pool
# TYPE ntfy_db_up gauge
ntfy_db_up{host="db:5432",role="%[1]s"} 1
ntfy_db_up{host="replica:5432",role="%[2]s"} 0
`, db.PoolRolePrimary, db.PoolRoleReplica)
require.Nil(t, testutil.GatherAndCompare(prometheus.DefaultGatherer, strings.NewReader(expected),
"ntfy_db_pool_connections_in_use", "ntfy_db_pool_connections_max", "ntfy_db_pool_waits_total", "ntfy_db_pool_wait_seconds_total"))
"ntfy_db_pool_connections_in_use", "ntfy_db_pool_connections_max", "ntfy_db_pool_waits_total", "ntfy_db_pool_wait_seconds_total", "ntfy_db_up"))
open, err := testutil.GatherAndCount(prometheus.DefaultGatherer, "ntfy_db_pool_connections_open", "ntfy_db_pool_connections_idle")
require.Nil(t, err)
require.Equal(t, 4, open)
+16
View File
@@ -7,6 +7,7 @@ import (
"io/fs"
"net/netip"
"reflect"
"runtime"
"text/template"
"time"
@@ -17,6 +18,7 @@ import (
// Defines default config settings (excluding limits, see below)
const (
DefaultListenAdminUnixMode = fs.FileMode(0660) // Owner (the ntfy user) and the ntfy group; root bypasses anyway
DefaultListenHTTP = ":80"
DefaultCacheDuration = 12 * time.Hour
DefaultCacheBatchTimeout = time.Duration(0)
@@ -129,6 +131,7 @@ type Config struct {
ListenHTTPS string
ListenUnix string
ListenUnixMode fs.FileMode
ListenAdminUnix string // Path of the local admin socket (CLI access without credentials); empty disables it
KeyFile string
CertFile string
DatabaseURL string // PostgreSQL connection string (e.g. "postgres://user:pass@host:5432/ntfy")
@@ -244,6 +247,18 @@ type Config struct {
BuildCommit string // Injected by App
}
// DefaultListenAdminUnix returns the default admin socket path. On Windows there is none: unix socket
// file permissions do not protect it there, so the CLI uses the database or --server instead.
func DefaultListenAdminUnix() string {
switch runtime.GOOS {
case "windows":
return ""
case "darwin":
return "/var/run/ntfy/admin.sock"
}
return "/run/ntfy/admin.sock"
}
// NewConfig instantiates a default new server config
func NewConfig() *Config {
return &Config{
@@ -253,6 +268,7 @@ func NewConfig() *Config {
ListenHTTPS: "",
ListenUnix: "",
ListenUnixMode: 0,
ListenAdminUnix: DefaultListenAdminUnix(),
KeyFile: "",
CertFile: "",
DatabaseURL: "",
+11
View File
@@ -154,9 +154,18 @@ var (
errHTTPBadRequestTimezoneInvalid = &errHTTP{40059, http.StatusBadRequest, "invalid request: invalid timezone", "https://ntfy.sh/docs/publish/#scheduled-delivery", nil}
errHTTPBadRequestPhoneVerificationCodeInvalid = &errHTTP{40060, http.StatusBadRequest, "invalid request: phone number verification code invalid", "", nil}
errHTTPBadRequestTooManyTopics = &errHTTP{40061, http.StatusBadRequest, "invalid request: too many topics", "https://ntfy.sh/docs/subscribe/api/#subscribe-to-multiple-topics", nil}
errHTTPBadRequestTierCodeInvalid = &errHTTP{40062, http.StatusBadRequest, "invalid request: tier code invalid, or tier name missing", "", nil}
errHTTPBadRequestTierReservationsLimit = &errHTTP{40063, http.StatusBadRequest, "invalid request: user has more topic reservations than the tier allows, remove reservations first", "", nil}
errHTTPBadRequestAccessAdminUser = &errHTTP{40064, http.StatusBadRequest, "invalid request: admins can access all topics, access control entries are not allowed", "", nil}
errHTTPBadRequestTokenNotFound = &errHTTP{40065, http.StatusBadRequest, "invalid request: token does not exist", "", nil}
errHTTPBadRequestRoleInvalid = &errHTTP{40066, http.StatusBadRequest, "invalid request: role must be user or admin", "", nil}
errHTTPBadRequestUserNoPrimaryEmail = &errHTTP{40067, http.StatusBadRequest, "invalid request: user has no primary email, cannot send reset link", "", nil}
errHTTPBadRequestBaseURLNotConfigured = &errHTTP{40068, http.StatusBadRequest, "invalid request: base-url must be configured on the server to generate links", "https://ntfy.sh/docs/config/", nil}
errHTTPBadRequestPasswordHashInvalid = &errHTTP{40069, http.StatusBadRequest, "invalid request: password hash must be a bcrypt hash with cost 10 or higher, use 'ntfy user hash' to generate", "", nil}
errHTTPNotFound = &errHTTP{40401, http.StatusNotFound, "page not found", "", nil}
errHTTPUnauthorized = &errHTTP{40101, http.StatusUnauthorized, "unauthorized", "https://ntfy.sh/docs/publish/#authentication", nil}
errHTTPForbidden = &errHTTP{40301, http.StatusForbidden, "forbidden", "https://ntfy.sh/docs/publish/#authentication", nil}
errHTTPForbiddenAdminSocketOnly = &errHTTP{40302, http.StatusForbidden, "forbidden: this operation is only allowed over the local admin socket", "https://ntfy.sh/docs/config/#admin-socket", nil}
errHTTPConflictUserExists = &errHTTP{40901, http.StatusConflict, "conflict: user already exists", "", nil}
errHTTPConflictTopicReserved = &errHTTP{40902, http.StatusConflict, "conflict: access control entry for topic or topic pattern already exists", "", nil}
errHTTPConflictSubscriptionExists = &errHTTP{40903, http.StatusConflict, "conflict: topic subscription already exists", "", nil}
@@ -165,6 +174,8 @@ var (
errHTTPConflictProvisionedTokenChange = &errHTTP{40906, http.StatusConflict, "conflict: cannot change or delete provisioned token", "", nil}
errHTTPConflictEmailExists = &errHTTP{40907, http.StatusConflict, "conflict: email address already exists", "", nil}
errHTTPConflictEmailPrimaryElsewhere = &errHTTP{40908, http.StatusConflict, "conflict: email address is the primary email on another account", "", nil}
errHTTPConflictTierExists = &errHTTP{40909, http.StatusConflict, "conflict: tier already exists", "", nil}
errHTTPConflictTierInUse = &errHTTP{40910, http.StatusConflict, "conflict: tier is still assigned to users", "", nil}
errHTTPGonePhoneVerificationExpired = &errHTTP{41001, http.StatusGone, "phone number verification expired or does not exist", "", nil}
errHTTPEntityTooLargeAttachment = &errHTTP{41301, http.StatusRequestEntityTooLarge, "attachment too large, or bandwidth limit reached", "https://ntfy.sh/docs/publish/#limitations", nil}
errHTTPEntityTooLargeMatrixRequest = &errHTTP{41302, http.StatusRequestEntityTooLarge, "Matrix request is larger than the max allowed length", "", nil}
+1
View File
@@ -30,6 +30,7 @@ const (
tagWebsocket = "websocket"
tagMatrix = "matrix"
tagWebPush = "webpush"
tagAdmin = "admin"
)
var (
+2
View File
@@ -5,6 +5,8 @@ After=network.target
[Service]
User=ntfy
Group=ntfy
RuntimeDirectory=ntfy
RuntimeDirectoryMode=0750
ExecStart=/usr/bin/ntfy serve --no-log-dates
ExecReload=/bin/kill --signal HUP $MAINPID
Restart=on-failure
+55 -2
View File
@@ -60,6 +60,7 @@ type Server struct {
httpProfileServer *http.Server
httpClusterServer *http.Server // Dedicated private listener for node-to-node fan-out (experimental-cluster-listen)
unixListener net.Listener
adminUnixListener net.Listener // Serves the admin API on the local admin socket, see handleAdminSocket
smtpServer *smtp.Server
smtpServerBackend *smtpBackend
mailer mail.Sender
@@ -126,6 +127,8 @@ var (
apiTiersPath = "/v1/tiers"
apiUsersPath = "/v1/users"
apiUsersAccessPath = "/v1/users/access"
apiUsersTokensPath = "/v1/users/tokens"
apiUsersPasswordResetPath = "/v1/users/password/reset"
apiAccountPath = "/v1/account"
apiAccountLoginPath = "/v1/account/login"
apiAccountTokenPath = "/v1/account/token"
@@ -141,6 +144,7 @@ var (
apiAccountEmailResendPath = "/v1/account/email/resend"
apiAccountPasswordResetRequestPath = "/v1/account/password/reset/request"
apiAccountPasswordResetPath = "/v1/account/password/reset"
apiAccountBillingTiersPath = "/v1/account/billing/tiers"
apiAccountBillingPortalPath = "/v1/account/billing/portal"
apiAccountBillingWebhookPath = "/v1/account/billing/webhook"
apiAccountBillingSubscriptionPath = "/v1/account/billing/subscription"
@@ -389,13 +393,14 @@ func New(conf *Config) (*Server, error) {
return s, nil
}
// dbPoolStats reports the database connection pools for the ntfy_db_pool_* metrics
// dbPoolStats reports the database connection pools for the ntfy_db_* metrics
func (s *Server) dbPoolStats() []*metrics.DBPool {
pools := make([]*metrics.DBPool, 0)
for _, p := range s.db.Stats() {
pools = append(pools, &metrics.DBPool{
Role: p.Role,
Addr: p.Addr,
Up: p.Healthy,
Stats: p.Stats,
})
}
@@ -452,6 +457,9 @@ func (s *Server) Run() error {
if s.config.ListenUnix != "" {
listenStr += fmt.Sprintf(" %s[unix]", s.config.ListenUnix)
}
if s.config.ListenAdminUnix != "" && s.userManager != nil {
listenStr += fmt.Sprintf(" %s[admin]", s.config.ListenAdminUnix)
}
if s.config.SMTPServerListen != "" {
listenStr += fmt.Sprintf(" %s[smtp]", s.config.SMTPServerListen)
}
@@ -519,6 +527,27 @@ func (s *Server) Run() error {
errChan <- httpServer.Serve(s.unixListener)
}()
}
if s.config.ListenAdminUnix != "" && s.userManager != nil { // Without auth, there is nothing to administer
go func() {
var err error
s.mu.Lock()
if s.stopped.Load() {
s.mu.Unlock()
return // Nobody would close this listener anymore
}
s.adminUnixListener, err = s.listenAdminSocket()
if err != nil {
s.mu.Unlock()
// Not fatal, unlike the other listeners: the CLI falls back to writing to the database (deprecated)
log.Tag(tagStartup).Err(err).Warn("Admin socket %s not available, 'ntfy user/access/token/tier' will write to the database directly, which is deprecated, see https://ntfy.sh/docs/deprecations/", s.config.ListenAdminUnix)
return
}
defer s.adminUnixListener.Close()
s.mu.Unlock()
httpServer := &http.Server{Handler: http.HandlerFunc(s.handleAdminSocket)}
errChan <- httpServer.Serve(s.adminUnixListener)
}()
}
if s.config.MetricsListenHTTP != "" {
s.httpMetricsServer = &http.Server{Addr: s.config.MetricsListenHTTP, Handler: promhttp.Handler()}
go func() {
@@ -594,6 +623,9 @@ func (s *Server) stop() {
if s.unixListener != nil {
s.unixListener.Close()
}
if s.adminUnixListener != nil {
s.adminUnixListener.Close()
}
if s.smtpServer != nil {
s.smtpServer.Close()
}
@@ -606,6 +638,9 @@ func (s *Server) stop() {
if s.tail != nil {
s.tail.Close() // Before the databases close under it
}
if s.cluster != nil {
s.cluster.Close() // Before the databases close, so it can deregister the node and its heartbeat stops
}
s.limiter.Close() // Writes the unflushed daily counts, before the databases close under it
s.closeDatabases()
if s.ban != nil {
@@ -734,6 +769,22 @@ func (s *Server) handleInternal(w http.ResponseWriter, r *http.Request, v *visit
return s.ensureAdmin(s.handleAccessAllow)(w, r, v)
} else if r.Method == http.MethodDelete && r.URL.Path == apiUsersAccessPath {
return s.ensureAdmin(s.handleAccessReset)(w, r, v)
} else if r.Method == http.MethodGet && r.URL.Path == apiUsersTokensPath {
return s.ensureAdmin(s.handleUsersTokensGet)(w, r, v)
} else if r.Method == http.MethodPost && r.URL.Path == apiUsersTokensPath {
return s.ensureAdmin(s.handleUsersTokensAdd)(w, r, v)
} else if r.Method == http.MethodDelete && r.URL.Path == apiUsersTokensPath {
return s.ensureAdmin(s.handleUsersTokensDelete)(w, r, v)
} else if r.Method == http.MethodPost && r.URL.Path == apiUsersPasswordResetPath {
return s.ensureAdmin(s.handleUsersPasswordReset)(w, r, v)
} else if r.Method == http.MethodGet && r.URL.Path == apiTiersPath && v.User().IsAdmin() {
return s.ensureAdmin(s.handleAdminTiersGet)(w, r, v) // Non-admins get the billing tiers, see below
} else if r.Method == http.MethodPost && r.URL.Path == apiTiersPath {
return s.ensureAdmin(s.handleAdminTiersAdd)(w, r, v)
} else if r.Method == http.MethodPut && r.URL.Path == apiTiersPath {
return s.ensureAdmin(s.handleAdminTiersUpdate)(w, r, v)
} else if r.Method == http.MethodDelete && r.URL.Path == apiTiersPath {
return s.ensureAdmin(s.handleAdminTiersDelete)(w, r, v)
} else if r.Method == http.MethodPost && r.URL.Path == apiAccountPath {
return s.ensureUserManager(s.handleAccountCreate)(w, r, v)
} else if r.Method == http.MethodGet && r.URL.Path == apiAccountPath {
@@ -800,8 +851,10 @@ func (s *Server) handleInternal(w http.ResponseWriter, r *http.Request, v *visit
return s.ensureWebPushEnabled(s.limitRequests(s.handleWebPushDelete))(w, r, v)
} else if r.Method == http.MethodGet && r.URL.Path == apiStatsPath {
return s.handleStats(w, r, v)
} else if r.Method == http.MethodGet && r.URL.Path == apiTiersPath {
} else if r.Method == http.MethodGet && r.URL.Path == apiAccountBillingTiersPath {
return s.ensurePaymentsEnabled(s.handleBillingTiersGet)(w, r, v)
} else if r.Method == http.MethodGet && r.URL.Path == apiTiersPath {
return s.ensurePaymentsEnabled(s.handleBillingTiersGet)(w, r, v) // Non-admins: web apps before v2.36.0, remove after 2026-11-22
} else if r.Method == http.MethodGet && r.URL.Path == matrixPushPath {
return s.handleMatrixDiscovery(w)
} else if r.Method == http.MethodGet && r.URL.Path == metricsPath && s.metricsHandler != nil {
+7
View File
@@ -31,6 +31,13 @@
# listen-unix: <socket-path>
# listen-unix-mode: <linux permissions, e.g. 0700>
# Local admin socket for the "ntfy user", "ntfy access", "ntfy token" and "ntfy tier" commands. Whoever can open
# the socket (mode 0660: the ntfy user and group, and root) can manage users without a password. The commands
# fall back to writing to the database directly if no server listens on it. Set to "-" to disable.
# The socket is only opened if auth-file or database-url is set. Not available on Windows.
#
# listen-admin-unix: "/run/ntfy/admin.sock"
# Path to the private key & cert file for the HTTPS web server. Not used if "listen-https" is not set.
#
# key-file: <filename>
+496 -61
View File
@@ -2,14 +2,53 @@ package server
import (
"errors"
"fmt"
"io/fs"
"net"
"net/http"
"net/netip"
"os"
"path/filepath"
"slices"
"strings"
"time"
"heckel.io/ntfy/v2/cluster"
"heckel.io/ntfy/v2/log"
"heckel.io/ntfy/v2/user"
)
const (
apiTierReset = "-" // Tier value that removes a user's tier (PUT /v1/users); same marker the CLI uses
adminSocketDirMode = fs.FileMode(0750)
adminSocketDialTimeout = time.Second // Used to detect a stale socket file (nobody listening)
)
var (
// adminSocketPaths are the paths served over the admin socket. All of them are admin-only
// routes in handleInternal (ensureAdmin), which the socket's synthetic admin user passes.
adminSocketPaths = []string{
apiVersionPath,
apiUsersPath,
apiUsersAccessPath,
apiUsersTokensPath,
apiUsersPasswordResetPath,
apiTiersPath,
}
// adminSocketUser is the synthetic admin every socket request runs as. Its name is one that
// user.AllowedUsername rejects, so it can never collide with a real account.
adminSocketUser = &user.User{
ID: "u_admin_socket",
Name: "(admin socket)",
Role: user.RoleAdmin,
Billing: &user.Billing{}, // Read by the visitor's log context
}
// adminSocketIP is the visitor IP of every socket request; they all come from this machine
adminSocketIP = netip.AddrFrom4([4]byte{127, 0, 0, 1})
)
func (s *Server) handleVersion(w http.ResponseWriter, r *http.Request, v *visitor) error {
return s.writeJSON(w, &apiVersionResponse{
Version: s.config.BuildVersion,
@@ -29,22 +68,26 @@ func (s *Server) handleUsersGet(w http.ResponseWriter, r *http.Request, v *visit
}
usersResponse := make([]*apiUserResponse, len(users))
for i, u := range users {
tier := ""
if u.Tier != nil {
tier = u.Tier.Code
}
userGrants := make([]*apiUserGrantResponse, len(grants[u.ID]))
for i, g := range grants[u.ID] {
userGrants[i] = &apiUserGrantResponse{
Topic: g.TopicPattern,
Permission: g.Permission.String(),
Topic: g.TopicPattern,
Permission: g.Permission.String(),
Provisioned: g.Provisioned,
}
}
usersResponse[i] = &apiUserResponse{
Username: u.Name,
Role: string(u.Role),
Tier: tier,
Grants: userGrants,
Username: u.Name,
Role: string(u.Role),
Provisioned: u.Provisioned,
Grants: userGrants,
}
if u.Tier != nil {
usersResponse[i].Tier = u.Tier.Code
usersResponse[i].TierName = u.Tier.Name
}
if u.Name == user.Everyone {
usersResponse[i].DefaultAccess = s.userManager.DefaultAccess().String()
}
}
return s.writeJSON(w, usersResponse)
@@ -55,7 +98,13 @@ func (s *Server) handleUsersAdd(w http.ResponseWriter, r *http.Request, v *visit
if err != nil {
return err
} else if !user.AllowedUsername(req.Username) || (req.Password == "" && req.Hash == "") {
return errHTTPBadRequest.Wrap("username invalid, or password/password_hash missing")
return errHTTPBadRequest.Wrap("username invalid, or \"password\" or \"hash\" missing")
}
role := user.RoleUser
if req.Role != "" {
if role, err = checkRoleChange(r, req.Role); err != nil {
return err
}
}
u, err := s.userManager.User(req.Username)
if err != nil && !errors.Is(err, user.ErrUserNotFound) {
@@ -63,25 +112,22 @@ func (s *Server) handleUsersAdd(w http.ResponseWriter, r *http.Request, v *visit
} else if u != nil {
return errHTTPConflictUserExists
}
var tier *user.Tier
if req.Tier != "" {
tier, err = s.userManager.Tier(req.Tier)
if errors.Is(err, user.ErrTierNotFound) {
return errHTTPBadRequestTierInvalid
} else if err != nil {
return err
if _, err := s.userManager.Tier(req.Tier); err != nil {
return mapUserErrorToHTTP(err)
}
}
password, hashed := req.Password, false
if req.Hash != "" {
password, hashed = req.Hash, true
}
if err := s.userManager.AddUser(req.Username, password, user.RoleUser, hashed); err != nil {
return err
adminLog(v, r).Info("Adding user %s (role %s)", req.Username, role)
if err := s.userManager.AddUser(req.Username, password, role, hashed); err != nil {
return mapUserErrorToHTTP(err)
}
if tier != nil {
if req.Tier != "" {
if err := s.userManager.ChangeTier(req.Username, req.Tier); err != nil {
return err
return mapUserErrorToHTTP(err)
}
}
return s.writeJSON(w, newSuccessResponse())
@@ -93,46 +139,66 @@ func (s *Server) handleUsersUpdate(w http.ResponseWriter, r *http.Request, v *vi
return err
} else if !user.AllowedUsername(req.Username) {
return errHTTPBadRequest.Wrap("username invalid")
} else if req.Password == "" && req.Hash == "" && req.Tier == "" {
return errHTTPBadRequest.Wrap("need to provide at least one of \"password\", \"password_hash\" or \"tier\"")
} else if req.Password == "" && req.Hash == "" && req.Tier == "" && req.Role == "" {
return errHTTPBadRequest.Wrap("need to provide at least one of \"password\", \"hash\", \"tier\" or \"role\"")
}
var role user.Role
if req.Role != "" {
if role, err = checkRoleChange(r, req.Role); err != nil {
return err
}
}
if req.Tier != "" && req.Tier != apiTierReset {
if _, err := s.userManager.Tier(req.Tier); err != nil {
return mapUserErrorToHTTP(err)
}
}
u, err := s.userManager.User(req.Username)
if err != nil && !errors.Is(err, user.ErrUserNotFound) {
return err
} else if u != nil {
if u.IsAdmin() {
return errHTTPForbidden
if u.IsAdmin() && !isAdminSocketRequest(r) {
return errHTTPForbiddenAdminSocketOnly
}
if req.Hash != "" {
if err := s.userManager.ChangePassword(req.Username, req.Hash, true); errors.Is(err, user.ErrProvisionedUserChange) {
return errHTTPConflictProvisionedUserChange
} else if err != nil {
return err
if req.Hash != "" || req.Password != "" {
adminLog(v, r).Info("Changing password of user %s", req.Username)
if req.Hash != "" {
err = s.userManager.ChangePassword(req.Username, req.Hash, true)
} else {
err = s.userManager.ChangePassword(req.Username, req.Password, false)
}
} else if req.Password != "" {
if err := s.userManager.ChangePassword(req.Username, req.Password, false); errors.Is(err, user.ErrProvisionedUserChange) {
return errHTTPConflictProvisionedUserChange
} else if err != nil {
return err
if err != nil {
return mapUserErrorToHTTP(err)
}
}
} else if req.Role != "" {
return errHTTPBadRequestUserNotFound // Role changes only apply to existing users, see apiUserAddOrUpdateRequest
} else {
// Upsert: the admin UI creates users with PUT
password, hashed := req.Password, false
if req.Hash != "" {
password, hashed = req.Hash, true
}
adminLog(v, r).Info("Adding user %s (role %s)", req.Username, user.RoleUser)
if err := s.userManager.AddUser(req.Username, password, user.RoleUser, hashed); err != nil {
return err
return mapUserErrorToHTTP(err)
}
}
if req.Tier != "" {
if _, err = s.userManager.Tier(req.Tier); errors.Is(err, user.ErrTierNotFound) {
return errHTTPBadRequestTierInvalid
} else if err != nil {
return err
if req.Tier == apiTierReset {
adminLog(v, r).Info("Removing tier of user %s", req.Username)
if err := s.userManager.ResetTier(req.Username); err != nil {
return mapUserErrorToHTTP(err)
}
} else if req.Tier != "" {
adminLog(v, r).Info("Changing tier of user %s to %s", req.Username, req.Tier)
if err := s.userManager.ChangeTier(req.Username, req.Tier); err != nil {
return err
return mapUserErrorToHTTP(err)
}
}
if req.Role != "" {
adminLog(v, r).Info("Changing role of user %s to %s", req.Username, role)
if err := s.userManager.ChangeRole(req.Username, role); err != nil {
return mapUserErrorToHTTP(err)
}
}
return s.writeJSON(w, newSuccessResponse())
@@ -144,17 +210,21 @@ func (s *Server) handleUsersDelete(w http.ResponseWriter, r *http.Request, v *vi
return err
}
u, err := s.userManager.User(req.Username)
if errors.Is(err, user.ErrUserNotFound) {
return errHTTPBadRequestUserNotFound
} else if err != nil {
return err
} else if !u.IsUser() {
if err != nil {
return mapUserErrorToHTTP(err)
} else if !u.IsUser() && !isAdminSocketRequest(r) {
return errHTTPUnauthorized.Wrap("can only remove regular users from API")
} else if u.Name == user.Everyone {
return errHTTPBadRequest.Wrap("cannot remove the everyone user")
}
if err := s.userManager.RemoveUser(req.Username); errors.Is(err, user.ErrProvisionedUserChange) {
return errHTTPConflictProvisionedUserChange
} else if err != nil {
return err
adminLog(v, r).Info("Removing user %s", req.Username)
if err := s.userManager.RemoveUser(req.Username); err != nil {
return mapUserErrorToHTTP(err)
}
if s.webPush != nil {
if err := s.webPush.RemoveSubscriptionsByUserID(u.ID); err != nil {
adminLog(v, r).Err(err).Warn("Failed to remove web push subscriptions of user %s", req.Username)
}
}
s.userDeleted(u) // FIXME super inefficient
return s.writeJSON(w, newSuccessResponse())
@@ -165,18 +235,19 @@ func (s *Server) handleAccessAllow(w http.ResponseWriter, r *http.Request, v *vi
if err != nil {
return err
}
_, err = s.userManager.User(req.Username)
if errors.Is(err, user.ErrUserNotFound) {
return errHTTPBadRequestUserNotFound
} else if err != nil {
return err
u, err := s.userManager.User(req.Username)
if err != nil {
return mapUserErrorToHTTP(err)
} else if u.IsAdmin() {
return errHTTPBadRequestAccessAdminUser
}
permission, err := user.ParsePermission(req.Permission)
if err != nil {
return errHTTPBadRequestPermissionInvalid
}
adminLog(v, r).Info("Granting %s access to topic %s for user %s", permission, req.Topic, req.Username)
if err := s.userManager.AllowAccess(req.Username, req.Topic, permission); err != nil {
return err
return mapUserErrorToHTTP(err)
}
return s.writeJSON(w, newSuccessResponse())
}
@@ -185,20 +256,291 @@ func (s *Server) handleAccessReset(w http.ResponseWriter, r *http.Request, v *vi
req, err := readJSONWithLimit[apiAccessResetRequest](r.Body, jsonBodyBytesLimit, false)
if err != nil {
return err
} else if req.Username == "" && req.Topic != "" {
return errHTTPBadRequest.Wrap("topic requires a username")
} else if req.Username == "" && !isAdminSocketRequest(r) {
return errHTTPForbiddenAdminSocketOnly // Resetting the whole ACL also opens all reserved topics
}
var u *user.User
if req.Username != "" {
if u, err = s.userManager.User(req.Username); err != nil {
return mapUserErrorToHTTP(err)
}
}
adminLog(v, r).Info("Resetting access for user %s, topic %s", req.Username, req.Topic)
if err := s.userManager.ResetAccess(req.Username, req.Topic); err != nil {
return mapUserErrorToHTTP(err)
}
// Without a username, the whole ACL is gone. Subscribers are not killed here; the periodic
// re-authorization of long-lived subscriptions catches up with them.
if u != nil {
topicPattern := req.Topic
if topicPattern == "" {
topicPattern = "*"
}
if err := s.killUserSubscriber(u, topicPattern); err != nil { // This may be a pattern
return err
}
}
return s.writeJSON(w, newSuccessResponse())
}
func (s *Server) handleUsersTokensGet(w http.ResponseWriter, r *http.Request, v *visitor) error {
var users []*user.User
if username := r.URL.Query().Get("username"); username != "" {
u, err := s.userManager.User(username)
if err != nil {
return mapUserErrorToHTTP(err)
} else if u.IsAdmin() && !isAdminSocketRequest(r) {
return errHTTPForbiddenAdminSocketOnly // Admin users' tokens can only be managed over the admin socket
}
users = []*user.User{u}
} else if !isAdminSocketRequest(r) {
return errHTTPForbiddenAdminSocketOnly // All users' tokens in one call, too much for a leaked admin token
} else {
var err error
if users, err = s.userManager.Users(); err != nil {
return err
}
}
response := make([]*apiUserTokenResponse, 0)
for _, u := range users {
tokens, err := s.userManager.Tokens(u.ID)
if err != nil {
return err
}
for _, t := range tokens {
response = append(response, newAPIUserTokenResponse(u.Name, t))
}
}
return s.writeJSON(w, response)
}
func (s *Server) handleUsersTokensAdd(w http.ResponseWriter, r *http.Request, v *visitor) error {
req, err := readJSONWithLimit[apiUsersTokensAddRequest](r.Body, jsonBodyBytesLimit, false)
if err != nil {
return err
} else if req.Username == user.Everyone {
return errHTTPBadRequest.Wrap("cannot create tokens for the everyone user")
}
u, err := s.userManager.User(req.Username)
if err != nil {
return mapUserErrorToHTTP(err)
} else if u.IsAdmin() && !isAdminSocketRequest(r) {
return errHTTPForbiddenAdminSocketOnly // Admin users' tokens can only be managed over the admin socket
}
expires := time.Unix(0, 0)
if req.Expires != 0 {
expires = time.Unix(req.Expires, 0)
if !expires.After(time.Now()) {
return errHTTPBadRequest.Wrap("expires must be in the future")
}
}
adminLog(v, r).Info("Creating token for user %s", req.Username)
token, err := s.userManager.CreateToken(u.ID, req.Label, expires, netip.IPv4Unspecified(), false)
if err != nil {
return mapUserErrorToHTTP(err)
}
return s.writeJSON(w, newAPIUserTokenResponse(u.Name, token))
}
func (s *Server) handleUsersTokensDelete(w http.ResponseWriter, r *http.Request, v *visitor) error {
req, err := readJSONWithLimit[apiUsersTokensDeleteRequest](r.Body, jsonBodyBytesLimit, false)
if err != nil {
return err
}
if err := s.userManager.ResetAccess(req.Username, req.Topic); err != nil {
return err
u, err := s.userManager.User(req.Username)
if err != nil {
return mapUserErrorToHTTP(err)
} else if u.IsAdmin() && !isAdminSocketRequest(r) {
return errHTTPForbiddenAdminSocketOnly // Admin users' tokens can only be managed over the admin socket
}
if err := s.killUserSubscriber(u, req.Topic); err != nil { // This may be a pattern
return err
adminLog(v, r).Info("Removing token for user %s", req.Username)
if err := s.userManager.RemoveToken(u.ID, req.Token); err != nil {
return mapUserErrorToHTTP(err)
}
return s.writeJSON(w, newSuccessResponse())
}
func (s *Server) handleUsersPasswordReset(w http.ResponseWriter, r *http.Request, v *visitor) error {
req, err := readJSONWithLimit[apiUsersPasswordResetRequest](r.Body, jsonBodyBytesLimit, false)
if err != nil {
return err
} else if s.config.BaseURL == "" {
return errHTTPBadRequestBaseURLNotConfigured
}
u, err := s.userManager.User(req.Username)
if err != nil {
return mapUserErrorToHTTP(err)
} else if u.Provisioned {
return errHTTPConflictProvisionedUserChange
} else if u.IsAdmin() && !isAdminSocketRequest(r) {
return errHTTPForbiddenAdminSocketOnly
}
var email string
if req.SendEmail {
if s.mailer == nil {
return errHTTPBadRequest.Wrap("smtp-sender-addr is not configured on the server")
}
if email, err = s.userManager.PrimaryEmail(u.ID); err != nil {
return err
} else if email == "" {
return errHTTPBadRequestUserNoPrimaryEmail
}
}
token, err := s.userManager.AddMagicLink(user.MagicLinkKindPasswordReset, u.ID, "", passwordResetTokenExpiry)
if err != nil {
return err
}
link := s.config.BaseURL + webAppPasswordResetPathPrefix + token
adminLog(v, r).Info("Creating password reset link for user %s", req.Username)
if req.SendEmail {
if err := s.mailer.SendPasswordReset(email, link); err != nil {
adminLog(v, r).Err(err).Warn("Failed to send password reset email to %s", email)
return fmt.Errorf("failed to send reset email to %s: %w", email, err)
}
}
return s.writeJSON(w, &apiUsersPasswordResetResponse{
Link: link,
Email: email,
})
}
func (s *Server) handleAdminTiersGet(w http.ResponseWriter, r *http.Request, v *visitor) error {
tiers, err := s.userManager.Tiers()
if err != nil {
return err
}
counts, err := s.userManager.TierUsersCount()
if err != nil {
return err
}
slices.SortFunc(tiers, func(a, b *user.Tier) int {
return strings.Compare(a.Code, b.Code)
})
response := make([]*apiAdminTier, len(tiers))
for i, t := range tiers {
response[i] = &apiAdminTier{
ID: t.ID,
Code: t.Code,
Name: t.Name,
MessagesLimit: t.MessageLimit,
MessagesExpiryDuration: int64(t.MessageExpiryDuration.Seconds()),
EmailsLimit: t.EmailLimit,
CallsLimit: t.CallLimit,
ReservationsLimit: t.ReservationLimit,
AttachmentFileSizeLimit: t.AttachmentFileSizeLimit,
AttachmentTotalSizeLimit: t.AttachmentTotalSizeLimit,
AttachmentExpiryDuration: int64(t.AttachmentExpiryDuration.Seconds()),
AttachmentBandwidthLimit: t.AttachmentBandwidthLimit,
StripeMonthlyPriceID: t.StripeMonthlyPriceID,
StripeYearlyPriceID: t.StripeYearlyPriceID,
Users: counts[t.Code],
}
}
return s.writeJSON(w, response)
}
func (s *Server) handleAdminTiersAdd(w http.ResponseWriter, r *http.Request, v *visitor) error {
tier, err := readAdminTierRequest(r)
if err != nil {
return err
}
if _, err := s.userManager.Tier(tier.Code); err == nil {
return errHTTPConflictTierExists
} else if !errors.Is(err, user.ErrTierNotFound) {
return err
}
adminLog(v, r).Info("Adding tier %s", tier.Code)
if err := s.userManager.AddTier(tier); err != nil {
return mapUserErrorToHTTP(err)
}
return s.writeJSON(w, newSuccessResponse())
}
func (s *Server) handleAdminTiersUpdate(w http.ResponseWriter, r *http.Request, v *visitor) error {
tier, err := readAdminTierRequest(r)
if err != nil {
return err
}
if _, err := s.userManager.Tier(tier.Code); err != nil {
return mapUserErrorToHTTP(err)
}
adminLog(v, r).Info("Updating tier %s", tier.Code)
if err := s.userManager.UpdateTier(tier); err != nil {
return mapUserErrorToHTTP(err)
}
return s.writeJSON(w, newSuccessResponse())
}
func (s *Server) handleAdminTiersDelete(w http.ResponseWriter, r *http.Request, v *visitor) error {
req, err := readJSONWithLimit[apiAdminTierDeleteRequest](r.Body, jsonBodyBytesLimit, false)
if err != nil {
return err
}
if _, err := s.userManager.Tier(req.Code); err != nil {
return mapUserErrorToHTTP(err)
}
counts, err := s.userManager.TierUsersCount()
if err != nil {
return err
} else if counts[req.Code] > 0 {
return errHTTPConflictTierInUse
}
adminLog(v, r).Info("Removing tier %s", req.Code)
if err := s.userManager.RemoveTier(req.Code); err != nil {
return mapUserErrorToHTTP(err)
}
return s.writeJSON(w, newSuccessResponse())
}
// listenAdminSocket creates the socket's directory, removes a stale socket file (one with no
// server behind it) and listens. A socket that answers belongs to another running server; in
// that case this one does not take it over.
func (s *Server) listenAdminSocket() (net.Listener, error) {
path := s.config.ListenAdminUnix
if err := os.MkdirAll(filepath.Dir(path), adminSocketDirMode); err != nil {
return nil, err
}
if reason := insecureAdminSocketDir(filepath.Dir(path)); reason != "" {
// Not fatal: the path is the admin's choice
log.Tag(tagStartup).Warn("Admin socket directory %s %s; other users may be able to use or replace the admin socket", filepath.Dir(path), reason)
}
if conn, err := net.DialTimeout("unix", path, adminSocketDialTimeout); err == nil {
conn.Close()
return nil, fmt.Errorf("another server is listening on %s", path)
}
if err := os.Remove(path); err != nil && !errors.Is(err, fs.ErrNotExist) {
return nil, err
}
listener, err := net.Listen("unix", path)
if err != nil {
return nil, fmt.Errorf("cannot listen on %s: %w", path, err)
}
if err := os.Chmod(path, DefaultListenAdminUnixMode); err != nil {
listener.Close()
return nil, err
}
return listener, nil
}
// handleAdminSocket serves one admin API request that arrived over the local socket. There is
// no authentication: whoever can open the socket (file mode 0660, owner ntfy) is an admin.
// The request runs as adminSocketUser, bypasses rate limiting and the ban feed, and carries
// contextAdminSocket so handlers can allow what the network API refuses.
func (s *Server) handleAdminSocket(w http.ResponseWriter, r *http.Request) {
r = withContext(r, map[contextKey]any{contextAdminSocket: true})
v := newVisitor(s.visitors, adminSocketIP, adminSocketUser)
adminLog(v, r).Debug("Admin socket request")
if !slices.Contains(adminSocketPaths, r.URL.Path) {
s.handleError(w, r, v, errHTTPNotFound) // Publishing, subscribing, the web app, etc. are not served here
return
}
if err := s.handleInternal(w, r, v); err != nil {
s.handleError(w, r, v, err)
}
}
func (s *Server) killUserSubscriber(u *user.User, topicPattern string) error {
topics, err := s.topicsFromPattern(topicPattern)
if err != nil {
@@ -300,3 +642,96 @@ func (s *Server) topicsForAccessChange(topicPattern string) []*topic {
}
return topics
}
// checkRoleChange validates a requested role change. Roles can only be set over the admin socket.
func checkRoleChange(r *http.Request, role string) (user.Role, error) {
if !isAdminSocketRequest(r) {
return "", errHTTPForbiddenAdminSocketOnly
} else if !user.AllowedRole(user.Role(role)) {
return "", errHTTPBadRequestRoleInvalid
}
return user.Role(role), nil
}
// readAdminTierRequest reads and validates an apiAdminTier request, and converts it to a user.Tier
func readAdminTierRequest(r *http.Request) (*user.Tier, error) {
req, err := readJSONWithLimit[apiAdminTier](r.Body, jsonBodyBytesLimit, false)
if err != nil {
return nil, err
} else if !user.AllowedTier(req.Code) || strings.TrimSpace(req.Name) == "" {
return nil, errHTTPBadRequestTierCodeInvalid
}
return &user.Tier{
Code: req.Code,
Name: strings.TrimSpace(req.Name),
MessageLimit: req.MessagesLimit,
MessageExpiryDuration: time.Duration(req.MessagesExpiryDuration) * time.Second,
EmailLimit: req.EmailsLimit,
CallLimit: req.CallsLimit,
ReservationLimit: req.ReservationsLimit,
AttachmentFileSizeLimit: req.AttachmentFileSizeLimit,
AttachmentTotalSizeLimit: req.AttachmentTotalSizeLimit,
AttachmentExpiryDuration: time.Duration(req.AttachmentExpiryDuration) * time.Second,
AttachmentBandwidthLimit: req.AttachmentBandwidthLimit,
StripeMonthlyPriceID: req.StripeMonthlyPriceID,
StripeYearlyPriceID: req.StripeYearlyPriceID,
}, nil
}
func newAPIUserTokenResponse(username string, t *user.Token) *apiUserTokenResponse {
response := &apiUserTokenResponse{
Username: username,
Token: t.Value,
Label: t.Label,
Expires: t.Expires.Unix(),
Provisioned: t.Provisioned,
}
// Tokens created without a request (admin API, CLI, provisioning) carry their creation time and
// origin 0.0.0.0 until first used; report those as never used
if t.LastOrigin.IsValid() && !t.LastOrigin.IsUnspecified() {
response.LastAccess = t.LastAccess.Unix()
response.LastOrigin = t.LastOrigin.String()
}
return response
}
// mapUserErrorToHTTP maps user manager errors to HTTP errors
func mapUserErrorToHTTP(err error) error {
switch {
case errors.Is(err, user.ErrUserNotFound):
return errHTTPBadRequestUserNotFound
case errors.Is(err, user.ErrUserExists):
return errHTTPConflictUserExists
case errors.Is(err, user.ErrTierNotFound):
return errHTTPBadRequestTierInvalid
case errors.Is(err, user.ErrTokenNotFound):
return errHTTPBadRequestTokenNotFound
case errors.Is(err, user.ErrProvisionedUserChange):
return errHTTPConflictProvisionedUserChange
case errors.Is(err, user.ErrProvisionedTokenChange):
return errHTTPConflictProvisionedTokenChange
case errors.Is(err, user.ErrTooManyReservations):
return errHTTPBadRequestTierReservationsLimit
case errors.Is(err, user.ErrPasswordHashInvalid), errors.Is(err, user.ErrPasswordHashWeak):
return errHTTPBadRequestPasswordHashInvalid
case errors.Is(err, user.ErrInvalidArgument):
return errHTTPBadRequest
}
return err
}
// isAdminSocketRequest reports whether the request came in over the local admin socket. Those
// requests may do what the network API refuses: change roles, and change or delete admins.
func isAdminSocketRequest(r *http.Request) bool {
socket, err := fromContext[bool](r, contextAdminSocket)
return err == nil && socket
}
// adminLog returns the audit log event for a mutating admin request, with how it arrived
func adminLog(v *visitor, r *http.Request) *log.Event {
via := "http"
if isAdminSocketRequest(r) {
via = "socket"
}
return logvr(v, r).Tag(tagAdmin).Field("admin_via", via)
}
+920 -5
View File
@@ -1,15 +1,33 @@
package server
import (
"bytes"
"context"
"encoding/json"
"github.com/stretchr/testify/require"
dbtest "heckel.io/ntfy/v2/db/test"
"heckel.io/ntfy/v2/model"
"heckel.io/ntfy/v2/user"
"heckel.io/ntfy/v2/util"
"errors"
"fmt"
"io"
"io/fs"
"net"
"net/http"
"net/http/httptest"
"net/netip"
"os"
"path/filepath"
"runtime"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/stretchr/testify/require"
"heckel.io/ntfy/v2/ban"
dbtest "heckel.io/ntfy/v2/db/test"
"heckel.io/ntfy/v2/log"
"heckel.io/ntfy/v2/model"
"heckel.io/ntfy/v2/user"
"heckel.io/ntfy/v2/util"
)
func TestVersion_Admin(t *testing.T) {
@@ -672,3 +690,900 @@ func TestAccess_RevokeBetweenAuthorizeAndSubscribe(t *testing.T) {
t.Fatal("the stream subscribed after the revoke stayed open")
}
}
func TestUser_GetWithDetails(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
conf := newTestConfigWithAuthFile(t, databaseURL)
conf.AuthDefault = user.PermissionRead
conf.AuthUsers = []*user.User{
{Name: "philuser", Hash: "$2a$10$U4WSIYY6evyGmZaraavM2e2JeVG6EMGUKN1uUwufUeeRd4Jpg6cGC", Role: user.RoleUser}, // philuser:philpass
}
conf.AuthAccess = map[string][]*user.Grant{
"philuser": {{TopicPattern: "provisioned-topic", Permission: user.PermissionReadWrite}},
}
s := newTestServer(t, conf)
defer s.closeDatabases()
// Admin, a user with a tier and a manual grant
require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin, false))
require.Nil(t, s.userManager.AddTier(&user.Tier{Code: "pro", Name: "Pro"}))
require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser, false))
require.Nil(t, s.userManager.ChangeTier("ben", "pro"))
require.Nil(t, s.userManager.AllowAccess("ben", "bens-topic", user.PermissionRead))
rr := request(t, s, "GET", "/v1/users", "", map[string]string{
"Authorization": util.BasicAuth("phil", "phil"),
})
require.Equal(t, 200, rr.Code)
var users []*apiUserResponse
require.Nil(t, json.NewDecoder(rr.Body).Decode(&users))
byName := make(map[string]*apiUserResponse)
for _, u := range users {
byName[u.Username] = u
}
// Tier code and display name
require.Equal(t, "pro", byName["ben"].Tier)
require.Equal(t, "Pro", byName["ben"].TierName)
require.False(t, byName["ben"].Provisioned)
require.Equal(t, []*apiUserGrantResponse{{Topic: "bens-topic", Permission: "read-only"}}, byName["ben"].Grants)
// Provisioned user and grant
require.True(t, byName["philuser"].Provisioned)
require.Equal(t, []*apiUserGrantResponse{{Topic: "provisioned-topic", Permission: "read-write", Provisioned: true}}, byName["philuser"].Grants)
// Default access only on the everyone entry
require.Equal(t, "read-only", byName[user.Everyone].DefaultAccess)
require.Equal(t, "", byName["ben"].DefaultAccess)
require.Equal(t, "", byName["phil"].DefaultAccess)
})
}
func TestUser_AddWithRole(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
s := newTestServer(t, newTestConfigWithAuthFile(t, databaseURL))
defer s.closeDatabases()
require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin, false))
// Role over HTTP is refused, even for admins
rr := request(t, s, "POST", "/v1/users", `{"username": "ben", "password":"ben", "role":"admin"}`, map[string]string{
"Authorization": util.BasicAuth("phil", "phil"),
})
require.Equal(t, 403, rr.Code)
require.Equal(t, 40302, toHTTPError(t, rr.Body.String()).Code)
_, err := s.userManager.User("ben")
require.ErrorIs(t, err, user.ErrUserNotFound)
// Invalid role over the socket
rr = requestAdminSocket(t, s, "POST", "/v1/users", `{"username": "ben", "password":"ben", "role":"superuser"}`)
require.Equal(t, 400, rr.Code)
require.Equal(t, 40066, toHTTPError(t, rr.Body.String()).Code)
// Role over the socket works
rr = requestAdminSocket(t, s, "POST", "/v1/users", `{"username": "ben", "password":"ben", "role":"admin"}`)
require.Equal(t, 200, rr.Code)
u, err := s.userManager.User("ben")
require.Nil(t, err)
require.Equal(t, user.RoleAdmin, u.Role)
})
}
func TestUser_AddWithInvalidHash(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
s := newTestServer(t, newTestConfigWithAuthFile(t, databaseURL))
defer s.closeDatabases()
require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin, false))
rr := request(t, s, "POST", "/v1/users", `{"username": "ben", "hash":"not-a-bcrypt-hash"}`, map[string]string{
"Authorization": util.BasicAuth("phil", "phil"),
})
require.Equal(t, 400, rr.Code)
require.Equal(t, 40069, toHTTPError(t, rr.Body.String()).Code)
// The error names the JSON fields as they are spelled
rr = request(t, s, "POST", "/v1/users", `{"username": "ben"}`, map[string]string{
"Authorization": util.BasicAuth("phil", "phil"),
})
require.Equal(t, 400, rr.Code)
require.Contains(t, toHTTPError(t, rr.Body.String()).Message, `"password" or "hash" missing`)
rr = request(t, s, "PUT", "/v1/users", `{"username": "ben"}`, map[string]string{
"Authorization": util.BasicAuth("phil", "phil"),
})
require.Equal(t, 400, rr.Code)
require.Contains(t, toHTTPError(t, rr.Body.String()).Message, `"password", "hash", "tier" or "role"`)
})
}
func TestUser_ChangeTierReset(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
s := newTestServer(t, newTestConfigWithAuthFile(t, databaseURL))
defer s.closeDatabases()
require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin, false))
require.Nil(t, s.userManager.AddTier(&user.Tier{Code: "pro", Name: "Pro"}))
require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser, false))
require.Nil(t, s.userManager.ChangeTier("ben", "pro"))
rr := request(t, s, "PUT", "/v1/users", `{"username": "ben", "tier":"-"}`, map[string]string{
"Authorization": util.BasicAuth("phil", "phil"),
})
require.Equal(t, 200, rr.Code)
u, err := s.userManager.User("ben")
require.Nil(t, err)
require.Nil(t, u.Tier)
})
}
func TestUser_ChangeTierTooManyReservations(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
s := newTestServer(t, newTestConfigWithAuthFile(t, databaseURL))
defer s.closeDatabases()
require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin, false))
require.Nil(t, s.userManager.AddTier(&user.Tier{Code: "big", Name: "Big", ReservationLimit: 5}))
require.Nil(t, s.userManager.AddTier(&user.Tier{Code: "small", Name: "Small", ReservationLimit: 1}))
require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser, false))
require.Nil(t, s.userManager.ChangeTier("ben", "big"))
require.Nil(t, s.userManager.AddReservation("ben", "topic1", user.PermissionDenyAll, 5))
require.Nil(t, s.userManager.AddReservation("ben", "topic2", user.PermissionDenyAll, 5))
rr := request(t, s, "PUT", "/v1/users", `{"username": "ben", "tier":"small"}`, map[string]string{
"Authorization": util.BasicAuth("phil", "phil"),
})
require.Equal(t, 400, rr.Code)
require.Equal(t, 40063, toHTTPError(t, rr.Body.String()).Code)
})
}
func TestUser_ChangeRole(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
s := newTestServer(t, newTestConfigWithAuthFile(t, databaseURL))
defer s.closeDatabases()
require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin, false))
require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser, false))
// Refused over HTTP
rr := request(t, s, "PUT", "/v1/users", `{"username": "ben", "role":"admin"}`, map[string]string{
"Authorization": util.BasicAuth("phil", "phil"),
})
require.Equal(t, 403, rr.Code)
require.Equal(t, 40302, toHTTPError(t, rr.Body.String()).Code)
// Invalid role, and missing user over the socket
rr = requestAdminSocket(t, s, "PUT", "/v1/users", `{"username": "ben", "role":"superuser"}`)
require.Equal(t, 40066, toHTTPError(t, rr.Body.String()).Code)
rr = requestAdminSocket(t, s, "PUT", "/v1/users", `{"username": "emma", "role":"admin"}`)
require.Equal(t, 40031, toHTTPError(t, rr.Body.String()).Code)
_, err := s.userManager.User("emma")
require.ErrorIs(t, err, user.ErrUserNotFound)
// Promote and demote over the socket
rr = requestAdminSocket(t, s, "PUT", "/v1/users", `{"username": "ben", "role":"admin"}`)
require.Equal(t, 200, rr.Code)
u, err := s.userManager.User("ben")
require.Nil(t, err)
require.Equal(t, user.RoleAdmin, u.Role)
rr = requestAdminSocket(t, s, "PUT", "/v1/users", `{"username": "ben", "role":"user"}`)
require.Equal(t, 200, rr.Code)
u, err = s.userManager.User("ben")
require.Nil(t, err)
require.Equal(t, user.RoleUser, u.Role)
})
}
func TestUser_ChangeAdminPasswordOverSocket(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
s := newTestServer(t, newTestConfigWithAuthFile(t, databaseURL))
defer s.closeDatabases()
require.Nil(t, s.userManager.AddUser("admin", "admin", user.RoleAdmin, false))
rr := requestAdminSocket(t, s, "PUT", "/v1/users", `{"username": "admin", "password": "admin-new"}`)
require.Equal(t, 200, rr.Code)
_, err := s.userManager.Authenticate("admin", "admin-new")
require.Nil(t, err)
})
}
func TestUser_DeleteAdmin(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
s := newTestServer(t, newTestConfigWithAuthFile(t, databaseURL))
defer s.closeDatabases()
require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin, false))
require.Nil(t, s.userManager.AddUser("admin", "admin", user.RoleAdmin, false))
// Refused over HTTP
rr := request(t, s, "DELETE", "/v1/users", `{"username": "admin"}`, map[string]string{
"Authorization": util.BasicAuth("phil", "phil"),
})
require.Equal(t, 401, rr.Code)
// Allowed over the socket
rr = requestAdminSocket(t, s, "DELETE", "/v1/users", `{"username": "admin"}`)
require.Equal(t, 200, rr.Code)
_, err := s.userManager.User("admin")
require.ErrorIs(t, err, user.ErrUserNotFound)
})
}
func TestUser_DeleteRemovesWebPushSubscriptions(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
s := newTestServer(t, configureAuth(t, newTestConfigWithWebPush(t, databaseURL)))
defer s.closeDatabases()
require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser, false))
u, err := s.userManager.User("ben")
require.Nil(t, err)
require.Nil(t, s.webPush.UpsertSubscription(testWebPushEndpoint, "kSC3T8aN1JCQxxPdrFLrZg", "BMKKbxdUU_xLS7G1Wh5AN8PvWOjCzkCuKZYb8apcqYrDxjOF_2piggBnoJLQYx9IeSD70fNuwawI3e9Y8m3S3PE", u.ID, netip.MustParseAddr("1.2.3.4"), []string{"test-topic"}))
requireSubscriptionCount(t, s, "test-topic", 1)
rr := requestAdminSocket(t, s, "DELETE", "/v1/users", `{"username": "ben"}`)
require.Equal(t, 200, rr.Code)
requireSubscriptionCount(t, s, "test-topic", 0)
})
}
func TestAccess_AllowAdminUser(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
s := newTestServer(t, newTestConfigWithAuthFile(t, databaseURL))
defer s.closeDatabases()
require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin, false))
rr := requestAdminSocket(t, s, "PUT", "/v1/users/access", `{"username": "phil", "topic":"gold", "permission":"ro"}`)
require.Equal(t, 400, rr.Code)
require.Equal(t, 40064, toHTTPError(t, rr.Body.String()).Code)
})
}
func TestAccess_ResetAll(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
s := newTestServer(t, newTestConfigWithAuthFile(t, databaseURL))
defer s.closeDatabases()
require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser, false))
require.Nil(t, s.userManager.AddUser("emma", "emma", user.RoleUser, false))
require.Nil(t, s.userManager.AllowAccess("ben", "gold", user.PermissionRead))
require.Nil(t, s.userManager.AllowAccess("emma", "silver", user.PermissionRead))
require.Nil(t, s.userManager.AllowAccess(user.Everyone, "public", user.PermissionRead))
require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin, false))
// Over HTTP, resetting the whole ACL is refused, even for admins
rr := request(t, s, "DELETE", "/v1/users/access", `{}`, map[string]string{
"Authorization": util.BasicAuth("phil", "phil"),
})
require.Equal(t, 403, rr.Code)
require.Equal(t, 40302, toHTTPError(t, rr.Body.String()).Code)
grants, err := s.userManager.AllGrants()
require.Nil(t, err)
require.Len(t, grants, 3)
// Over the socket, it works
rr = requestAdminSocket(t, s, "DELETE", "/v1/users/access", `{}`)
require.Equal(t, 200, rr.Code)
grants, err = s.userManager.AllGrants()
require.Nil(t, err)
require.Empty(t, grants)
})
}
func TestAccess_ResetUserAllTopics(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
s := newTestServer(t, newTestConfigWithAuthFile(t, databaseURL))
defer s.closeDatabases()
require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser, false))
require.Nil(t, s.userManager.AddUser("emma", "emma", user.RoleUser, false))
require.Nil(t, s.userManager.AllowAccess("ben", "gold", user.PermissionRead))
require.Nil(t, s.userManager.AllowAccess("ben", "silver", user.PermissionRead))
require.Nil(t, s.userManager.AllowAccess("emma", "silver", user.PermissionRead))
rr := requestAdminSocket(t, s, "DELETE", "/v1/users/access", `{"username":"ben"}`)
require.Equal(t, 200, rr.Code)
bensGrants, err := s.userManager.Grants("ben")
require.Nil(t, err)
require.Empty(t, bensGrants)
emmasGrants, err := s.userManager.Grants("emma")
require.Nil(t, err)
require.Len(t, emmasGrants, 1)
// Unknown user
rr = requestAdminSocket(t, s, "DELETE", "/v1/users/access", `{"username":"nobody"}`)
require.Equal(t, 40031, toHTTPError(t, rr.Body.String()).Code)
})
}
func TestTokens_AddListRemove(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
s := newTestServer(t, newTestConfigWithAuthFile(t, databaseURL))
defer s.closeDatabases()
require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin, false))
require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser, false))
require.Nil(t, s.userManager.AddUser("emma", "emma", user.RoleUser, false))
auth := map[string]string{"Authorization": util.BasicAuth("phil", "phil")}
// No tokens yet: empty array, not null
rr := request(t, s, "GET", "/v1/users/tokens?username=ben", "", auth)
require.Equal(t, 200, rr.Code)
require.Equal(t, "[]", strings.TrimSpace(rr.Body.String()))
// Add a token that never expires, and one that does
rr = request(t, s, "POST", "/v1/users/tokens", `{"username":"ben","label":"forever"}`, auth)
require.Equal(t, 200, rr.Code)
var forever apiUserTokenResponse
require.Nil(t, json.NewDecoder(rr.Body).Decode(&forever))
require.Equal(t, "ben", forever.Username)
require.Equal(t, "forever", forever.Label)
require.Equal(t, int64(0), forever.Expires)
require.True(t, user.ValidToken(forever.Token))
require.Equal(t, int64(0), forever.LastAccess) // Never used
require.Equal(t, "", forever.LastOrigin)
expires := time.Now().Add(time.Hour).Unix()
rr = request(t, s, "POST", "/v1/users/tokens", fmt.Sprintf(`{"username":"emma","expires":%d}`, expires), auth)
require.Equal(t, 200, rr.Code)
var expiring apiUserTokenResponse
require.Nil(t, json.NewDecoder(rr.Body).Decode(&expiring))
require.Equal(t, expires, expiring.Expires)
// The new token works
rr = request(t, s, "GET", "/v1/account", "", map[string]string{"Authorization": util.BearerAuth(forever.Token)})
require.Equal(t, 200, rr.Code)
// Listing all users' tokens is refused over HTTP, but works over the socket
rr = request(t, s, "GET", "/v1/users/tokens", "", auth)
require.Equal(t, 403, rr.Code)
require.Equal(t, 40302, toHTTPError(t, rr.Body.String()).Code)
rr = requestAdminSocket(t, s, "GET", "/v1/users/tokens", "")
require.Equal(t, 200, rr.Code)
var tokens []*apiUserTokenResponse
require.Nil(t, json.NewDecoder(rr.Body).Decode(&tokens))
require.Len(t, tokens, 2)
// List one user
rr = request(t, s, "GET", "/v1/users/tokens?username=ben", "", auth)
require.Equal(t, 200, rr.Code)
tokens = nil
require.Nil(t, json.NewDecoder(rr.Body).Decode(&tokens))
require.Len(t, tokens, 1)
require.Equal(t, forever.Token, tokens[0].Token)
// Remove
rr = request(t, s, "DELETE", "/v1/users/tokens", fmt.Sprintf(`{"username":"ben","token":"%s"}`, forever.Token), auth)
require.Equal(t, 200, rr.Code)
bensTokens, err := s.userManager.Tokens(mustUserID(t, s, "ben"))
require.Nil(t, err)
require.Empty(t, bensTokens)
})
}
func TestTokens_Failures(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
conf := newTestConfigWithAuthFile(t, databaseURL)
conf.AuthUsers = []*user.User{
{Name: "philuser", Hash: "$2a$10$U4WSIYY6evyGmZaraavM2e2JeVG6EMGUKN1uUwufUeeRd4Jpg6cGC", Role: user.RoleUser}, // philuser:philpass
}
conf.AuthTokens = map[string][]*user.Token{
"philuser": {{Value: "tk_op56p8lz5bf3cxkz9je99v9oc37lo", Label: "provisioned"}},
}
s := newTestServer(t, conf)
defer s.closeDatabases()
require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin, false))
require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser, false))
auth := map[string]string{"Authorization": util.BasicAuth("phil", "phil")}
// Expiry in the past
rr := request(t, s, "POST", "/v1/users/tokens", fmt.Sprintf(`{"username":"ben","expires":%d}`, time.Now().Add(-time.Hour).Unix()), auth)
require.Equal(t, 400, rr.Code)
// Unknown user, everyone
rr = request(t, s, "POST", "/v1/users/tokens", `{"username":"nobody"}`, auth)
require.Equal(t, 40031, toHTTPError(t, rr.Body.String()).Code)
rr = request(t, s, "POST", "/v1/users/tokens", `{"username":"*"}`, auth)
require.Equal(t, 400, rr.Code)
rr = request(t, s, "GET", "/v1/users/tokens?username=nobody", "", auth)
require.Equal(t, 40031, toHTTPError(t, rr.Body.String()).Code)
// Unknown token, provisioned token
rr = request(t, s, "DELETE", "/v1/users/tokens", `{"username":"ben","token":"tk_doesnotexistdoesnotexist12345"}`, auth)
require.Equal(t, 40065, toHTTPError(t, rr.Body.String()).Code)
rr = request(t, s, "DELETE", "/v1/users/tokens", `{"username":"philuser","token":"tk_op56p8lz5bf3cxkz9je99v9oc37lo"}`, auth)
require.Equal(t, 40906, toHTTPError(t, rr.Body.String()).Code)
// Non-admins cannot use the API
rr = request(t, s, "GET", "/v1/users/tokens?username=ben", "", map[string]string{"Authorization": util.BasicAuth("ben", "ben")})
require.Equal(t, 401, rr.Code)
})
}
func TestTokens_AdminTokensOnlyOverSocket(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
s := newTestServer(t, newTestConfigWithAuthFile(t, databaseURL))
defer s.closeDatabases()
require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin, false))
require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser, false))
_, err := s.userManager.CreateToken(mustUserID(t, s, "phil"), "admin token", time.Unix(0, 0), netip.IPv4Unspecified(), false)
require.Nil(t, err)
_, err = s.userManager.CreateToken(mustUserID(t, s, "ben"), "user token", time.Unix(0, 0), netip.IPv4Unspecified(), false)
require.Nil(t, err)
auth := map[string]string{"Authorization": util.BasicAuth("phil", "phil")}
// Over HTTP, admin tokens cannot be listed, created or removed
rr := request(t, s, "GET", "/v1/users/tokens?username=phil", "", auth)
require.Equal(t, 403, rr.Code)
require.Equal(t, 40302, toHTTPError(t, rr.Body.String()).Code)
rr = request(t, s, "GET", "/v1/users/tokens?username=ben", "", auth)
require.Equal(t, 200, rr.Code)
var tokens []*apiUserTokenResponse
require.Nil(t, json.NewDecoder(rr.Body).Decode(&tokens))
require.Len(t, tokens, 1)
require.Equal(t, "ben", tokens[0].Username)
rr = request(t, s, "POST", "/v1/users/tokens", `{"username":"phil"}`, auth)
require.Equal(t, 403, rr.Code)
require.Equal(t, 40302, toHTTPError(t, rr.Body.String()).Code)
// Over the socket, they are visible and can be created
rr = requestAdminSocket(t, s, "GET", "/v1/users/tokens", "")
require.Equal(t, 200, rr.Code)
tokens = nil
require.Nil(t, json.NewDecoder(rr.Body).Decode(&tokens))
require.Len(t, tokens, 2)
rr = requestAdminSocket(t, s, "POST", "/v1/users/tokens", `{"username":"phil"}`)
require.Equal(t, 200, rr.Code)
})
}
func TestPasswordReset_Link(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
s, mailer, _ := newEmailTestServer(t, databaseURL)
defer s.closeDatabases()
ben, err := s.userManager.User("ben")
require.Nil(t, err)
// Link only
rr := requestAdminSocket(t, s, "POST", "/v1/users/password/reset", `{"username":"ben"}`)
require.Equal(t, 200, rr.Code)
var resp apiUsersPasswordResetResponse
require.Nil(t, json.NewDecoder(rr.Body).Decode(&resp))
require.Equal(t, "", resp.Email)
token := tokenFromLink(t, resp.Link, "https://ntfy.example.com/account/password/reset/")
link, err := s.userManager.MagicLinkByToken(token)
require.Nil(t, err)
require.Equal(t, ben.ID, link.UserID)
require.Empty(t, mailer.resetLinks)
// Send email: no primary email yet
rr = requestAdminSocket(t, s, "POST", "/v1/users/password/reset", `{"username":"ben","send_email":true}`)
require.Equal(t, 400, rr.Code)
require.Equal(t, 40067, toHTTPError(t, rr.Body.String()).Code)
// Send email to the primary email
require.Nil(t, s.userManager.AddEmail(ben.ID, "ben@example.com"))
require.Nil(t, s.userManager.SetPrimaryEmail(ben.ID, "ben@example.com"))
rr = requestAdminSocket(t, s, "POST", "/v1/users/password/reset", `{"username":"ben","send_email":true}`)
require.Equal(t, 200, rr.Code)
resp = apiUsersPasswordResetResponse{}
require.Nil(t, json.NewDecoder(rr.Body).Decode(&resp))
require.Equal(t, "ben@example.com", resp.Email)
require.Equal(t, resp.Link, mailer.resetLinks["ben@example.com"])
// Unknown user
rr = requestAdminSocket(t, s, "POST", "/v1/users/password/reset", `{"username":"nobody"}`)
require.Equal(t, 40031, toHTTPError(t, rr.Body.String()).Code)
})
}
func TestPasswordReset_Failures(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
conf := newTestConfigWithAuthFile(t, databaseURL)
conf.AuthUsers = []*user.User{
{Name: "philuser", Hash: "$2a$10$U4WSIYY6evyGmZaraavM2e2JeVG6EMGUKN1uUwufUeeRd4Jpg6cGC", Role: user.RoleUser}, // philuser:philpass
}
s := newTestServer(t, conf)
defer s.closeDatabases()
require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin, false))
require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser, false))
auth := map[string]string{"Authorization": util.BasicAuth("phil", "phil")}
// No mailer configured
rr := request(t, s, "POST", "/v1/users/password/reset", `{"username":"ben","send_email":true}`, auth)
require.Equal(t, 400, rr.Code)
// Provisioned user, admin over HTTP
rr = request(t, s, "POST", "/v1/users/password/reset", `{"username":"philuser"}`, auth)
require.Equal(t, 40905, toHTTPError(t, rr.Body.String()).Code)
rr = request(t, s, "POST", "/v1/users/password/reset", `{"username":"phil"}`, auth)
require.Equal(t, 403, rr.Code)
// No base URL
s.config.BaseURL = ""
rr = request(t, s, "POST", "/v1/users/password/reset", `{"username":"ben"}`, auth)
require.Equal(t, 40068, toHTTPError(t, rr.Body.String()).Code)
})
}
func TestTiers_AddListUpdateRemove(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
s := newTestServer(t, newTestConfigWithAuthFile(t, databaseURL))
defer s.closeDatabases()
require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin, false))
require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser, false))
auth := map[string]string{"Authorization": util.BasicAuth("phil", "phil")}
// Add
rr := request(t, s, "POST", "/v1/tiers", `{"code":"pro","name":"Pro","messages_limit":1000,"messages_expiry_duration":86400,"attachment_file_size_limit":1048576}`, auth)
require.Equal(t, 200, rr.Code)
tier, err := s.userManager.Tier("pro")
require.Nil(t, err)
require.Equal(t, int64(1000), tier.MessageLimit)
require.Equal(t, 24*time.Hour, tier.MessageExpiryDuration)
require.Nil(t, s.userManager.ChangeTier("ben", "pro"))
// List
rr = request(t, s, "GET", "/v1/tiers", "", auth)
require.Equal(t, 200, rr.Code)
var tiers []*apiAdminTier
require.Nil(t, json.NewDecoder(rr.Body).Decode(&tiers))
require.Len(t, tiers, 1)
require.Equal(t, tier.ID, tiers[0].ID)
require.Equal(t, "Pro", tiers[0].Name)
require.Equal(t, int64(86400), tiers[0].MessagesExpiryDuration)
require.Equal(t, int64(1048576), tiers[0].AttachmentFileSizeLimit)
require.Equal(t, int64(1), tiers[0].Users)
// Update
rr = request(t, s, "PUT", "/v1/tiers", `{"code":"pro","name":"Pro Plus","messages_limit":2000}`, auth)
require.Equal(t, 200, rr.Code)
tier, err = s.userManager.Tier("pro")
require.Nil(t, err)
require.Equal(t, "Pro Plus", tier.Name)
require.Equal(t, int64(2000), tier.MessageLimit)
// Remove: refused while in use, works after moving the user
rr = request(t, s, "DELETE", "/v1/tiers", `{"code":"pro"}`, auth)
require.Equal(t, 409, rr.Code)
require.Equal(t, 40910, toHTTPError(t, rr.Body.String()).Code)
require.Nil(t, s.userManager.ResetTier("ben"))
rr = request(t, s, "DELETE", "/v1/tiers", `{"code":"pro"}`, auth)
require.Equal(t, 200, rr.Code)
_, err = s.userManager.Tier("pro")
require.ErrorIs(t, err, user.ErrTierNotFound)
})
}
func TestTiers_Failures(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
s := newTestServer(t, newTestConfigWithAuthFile(t, databaseURL))
defer s.closeDatabases()
require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin, false))
require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser, false))
require.Nil(t, s.userManager.AddTier(&user.Tier{Code: "pro", Name: "Pro"}))
auth := map[string]string{"Authorization": util.BasicAuth("phil", "phil")}
// Duplicate, bad code, missing name
rr := request(t, s, "POST", "/v1/tiers", `{"code":"pro","name":"Pro"}`, auth)
require.Equal(t, 40909, toHTTPError(t, rr.Body.String()).Code)
rr = request(t, s, "POST", "/v1/tiers", `{"code":"not valid","name":"X"}`, auth)
require.Equal(t, 40062, toHTTPError(t, rr.Body.String()).Code)
rr = request(t, s, "POST", "/v1/tiers", `{"code":"x"}`, auth)
require.Equal(t, 40062, toHTTPError(t, rr.Body.String()).Code)
// Update or remove a missing tier
rr = request(t, s, "PUT", "/v1/tiers", `{"code":"missing","name":"Missing"}`, auth)
require.Equal(t, 40030, toHTTPError(t, rr.Body.String()).Code)
rr = request(t, s, "DELETE", "/v1/tiers", `{"code":"missing"}`, auth)
require.Equal(t, 40030, toHTTPError(t, rr.Body.String()).Code)
// Non-admins cannot change tiers; listing them is the billing list (for older web apps), which needs payments
ben := map[string]string{"Authorization": util.BasicAuth("ben", "ben")}
rr = request(t, s, "POST", "/v1/tiers", `{"code":"ben","name":"Ben"}`, ben)
require.Equal(t, 401, rr.Code)
rr = request(t, s, "DELETE", "/v1/tiers", `{"code":"pro"}`, ben)
require.Equal(t, 401, rr.Code)
rr = request(t, s, "GET", "/v1/tiers", "", ben)
require.Equal(t, 404, rr.Code)
rr = request(t, s, "GET", "/v1/tiers", "", nil)
require.Equal(t, 404, rr.Code)
rr = request(t, s, "GET", "/v1/admin/tiers", "", auth) // Gone, never released
require.Equal(t, 404, rr.Code)
})
}
func TestAdminSocket_ListenAndRemove(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
s, socket := newTestServerWithAdminSocket(t, newTestConfigWithAuthFile(t, databaseURL))
// Socket exists with the right mode
stat, err := os.Stat(socket)
require.Nil(t, err)
require.Equal(t, os.ModeSocket, stat.Mode().Type())
require.Equal(t, DefaultListenAdminUnixMode, stat.Mode().Perm())
// Removed on stop
s.Stop()
_, err = os.Stat(socket)
require.ErrorIs(t, err, os.ErrNotExist)
})
}
func TestAdminSocket_UsersWithoutCredentials(t *testing.T) {
forEachBackend(t, func(t *testing.T, databaseURL string) {
s, socket := newTestServerWithAdminSocket(t, newTestConfigWithAuthFile(t, databaseURL))
require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser, false))
client := adminSocketClient(socket)
// List users, no Authorization header
resp, err := client.Get("http://ntfy/v1/users")
require.Nil(t, err)
require.Equal(t, 200, resp.StatusCode)
var users []*apiUserResponse
require.Nil(t, json.NewDecoder(resp.Body).Decode(&users))
resp.Body.Close()
require.Len(t, users, 2) // ben and everyone
// Create an admin: role changes work end to end
resp, err = client.Post("http://ntfy/v1/users", "application/json", strings.NewReader(`{"username":"phil","password":"phil","role":"admin"}`))
require.Nil(t, err)
require.Equal(t, 200, resp.StatusCode)
resp.Body.Close()
u, err := s.userManager.User("phil")
require.Nil(t, err)
require.Equal(t, user.RoleAdmin, u.Role)
})
}
func TestAdminSocket_NonAdminRoutes(t *testing.T) {
_, socket := newTestServerWithAdminSocket(t, newTestConfigWithAuthFile(t, ""))
client := adminSocketClient(socket)
resp, err := client.Get("http://ntfy/mytopic/json?poll=1")
require.Nil(t, err)
require.Equal(t, 404, resp.StatusCode)
resp.Body.Close()
resp, err = client.Post("http://ntfy/mytopic", "text/plain", strings.NewReader("hi"))
require.Nil(t, err)
require.Equal(t, 404, resp.StatusCode)
resp.Body.Close()
resp, err = client.Get("http://ntfy/v1/account") // A real route in handleInternal, but not an admin route
require.Nil(t, err)
require.Equal(t, 404, resp.StatusCode)
resp.Body.Close()
}
func TestAdminSocket_StaleSocketFile(t *testing.T) {
conf := newTestConfigWithAuthFile(t, "")
conf.ListenAdminUnix = filepath.Join(shortTempDir(t), "admin.sock")
require.Nil(t, os.WriteFile(conf.ListenAdminUnix, []byte("stale"), 0600))
s := newTestServer(t, conf)
runTestServer(t, s)
waitFor(t, func() bool { return adminSocketListening(s) })
resp, err := adminSocketClient(conf.ListenAdminUnix).Get("http://ntfy/v1/users")
require.Nil(t, err)
require.Equal(t, 200, resp.StatusCode)
resp.Body.Close()
}
func TestAdminSocket_AnotherServerListening(t *testing.T) {
conf := newTestConfigWithAuthFile(t, "")
conf.ListenAdminUnix = filepath.Join(shortTempDir(t), "admin.sock")
other, err := net.Listen("unix", conf.ListenAdminUnix)
require.Nil(t, err)
defer other.Close()
s := newTestServer(t, conf)
output := captureLogUntil(t, "another server is listening", func() {
runTestServer(t, s)
})
require.Contains(t, output, "another server is listening")
require.False(t, adminSocketListening(s))
// The other listener is untouched
_, err = os.Stat(conf.ListenAdminUnix)
require.Nil(t, err)
}
func TestAdminSocket_UnwritableDirectory(t *testing.T) {
if os.Getuid() == 0 {
t.Skip("root can write anywhere")
}
dir := shortTempDir(t)
require.Nil(t, os.Chmod(dir, 0500))
t.Cleanup(func() { os.Chmod(dir, 0700) })
conf := newTestConfigWithAuthFile(t, "")
conf.ListenAdminUnix = filepath.Join(dir, "sub", "admin.sock")
s := newTestServer(t, conf)
output := captureLogUntil(t, "not available", func() {
runTestServer(t, s)
})
require.Contains(t, output, "Admin socket")
require.Contains(t, output, "directly, which is deprecated, see https://ntfy.sh/docs/deprecations/")
require.False(t, adminSocketListening(s))
}
func TestAdminSocket_InsecureDirectoryWarning(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("unix file modes")
}
for _, tt := range []struct {
mode fs.FileMode
warn bool
}{
{0750, false},
{0777, true}, // Anyone can replace the socket
{0777 | fs.ModeSticky, false}, // Like /tmp: only the owner can remove or replace it
} {
dir := shortTempDir(t)
require.Nil(t, os.Chmod(dir, tt.mode))
conf := newTestConfigWithAuthFile(t, "")
conf.ListenAdminUnix = filepath.Join(dir, "admin.sock")
s := newTestServer(t, conf)
output := captureLog(t, func() {
runTestServer(t, s)
waitFor(t, func() bool { return adminSocketListening(s) }) // Warned before it listens
})
if tt.warn {
require.Contains(t, output, "Admin socket directory "+dir+" is writable by all users", tt.mode)
} else {
require.NotContains(t, output, "Admin socket directory", tt.mode)
}
}
}
func TestAdminSocket_DisabledWithoutPathOrAuth(t *testing.T) {
// No path: no listener, no warning
s := newTestServer(t, newTestConfigWithAuthFile(t, ""))
output := captureLog(t, func() {
runTestServer(t, s)
})
require.False(t, adminSocketListening(s)) // Run does not even try, so there is nothing to wait for
require.NotContains(t, output, "Admin socket")
// No auth configured: nothing to administer, so no listener either
conf := newTestConfig(t, "")
conf.ListenAdminUnix = filepath.Join(shortTempDir(t), "admin.sock")
s = newTestServer(t, conf)
runTestServer(t, s)
require.False(t, adminSocketListening(s))
_, err := os.Stat(conf.ListenAdminUnix)
require.ErrorIs(t, err, os.ErrNotExist)
}
func TestAdminSocket_AuditLog(t *testing.T) {
_, socket := newTestServerWithAdminSocket(t, newTestConfigWithAuthFile(t, ""))
output := captureLog(t, func() {
resp, err := adminSocketClient(socket).Post("http://ntfy/v1/users", "application/json", strings.NewReader(`{"username":"ben","password":"ben"}`))
require.Nil(t, err)
require.Equal(t, 200, resp.StatusCode)
resp.Body.Close()
})
require.Contains(t, output, "Adding user ben")
require.Contains(t, output, "admin_via=socket")
}
func TestAdminSocket_NoBanFeed(t *testing.T) {
banFile := filepath.Join(t.TempDir(), "ban.log")
conf := newTestConfigWithAuthFile(t, "")
conf.BanFile = banFile
conf.BanWindow = time.Minute
conf.BanThreshold = 1
conf.BanWeights = ban.Weights{"*": 1}
s, socket := newTestServerWithAdminSocket(t, conf)
require.NotNil(t, s.ban)
client := adminSocketClient(socket)
for i := 0; i < 3; i++ {
resp, err := client.Get("http://ntfy/nope")
require.Nil(t, err)
require.Equal(t, 404, resp.StatusCode)
resp.Body.Close()
}
// Writes are async; Close flushes. Nothing may have been recorded.
s.ban.Close()
data, err := os.ReadFile(banFile)
if !errors.Is(err, os.ErrNotExist) {
require.Nil(t, err)
require.Empty(t, strings.TrimSpace(string(data)))
}
}
// requestAdminSocket sends a request through the admin socket handler, i.e. with socket semantics
// (no Authorization header, contextAdminSocket set), without a listener.
func requestAdminSocket(t *testing.T, s *Server, method, url, body string) *httptest.ResponseRecorder {
rr := httptest.NewRecorder()
r, err := http.NewRequest(method, url, strings.NewReader(body))
require.Nil(t, err)
s.handleAdminSocket(rr, r)
return rr
}
func mustUserID(t *testing.T, s *Server, username string) string {
u, err := s.userManager.User(username)
require.Nil(t, err)
return u.ID
}
// newTestServerWithAdminSocket starts a test server with the admin socket listening in a short
// temp dir (unix socket paths are limited to ~104 bytes, t.TempDir() can be longer on macOS)
func newTestServerWithAdminSocket(t *testing.T, conf *Config) (*Server, string) {
conf.ListenAdminUnix = filepath.Join(shortTempDir(t), "admin.sock")
s := newTestServer(t, conf)
runTestServer(t, s)
waitFor(t, func() bool { return adminSocketListening(s) })
return s, conf.ListenAdminUnix
}
// adminSocketListening reports whether Run's admin socket goroutine is listening
func adminSocketListening(s *Server) bool {
s.mu.RLock()
defer s.mu.RUnlock()
return s.adminUnixListener != nil
}
// runTestServer runs the server in the background without network listeners, and waits until
// Run has started its listeners (it holds the lock while doing so)
func runTestServer(t *testing.T, s *Server) {
s.config.ListenHTTP = nil
go s.Run()
waitFor(t, func() bool {
s.mu.RLock()
defer s.mu.RUnlock()
return s.closeChan != nil
})
}
// adminSocketClient returns an HTTP client that sends every request to the given unix socket
func adminSocketClient(socket string) *http.Client {
return &http.Client{
Timeout: 5 * time.Second,
Transport: &http.Transport{
DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
var d net.Dialer
return d.DialContext(ctx, "unix", socket)
},
},
}
}
func shortTempDir(t *testing.T) string {
dir, err := os.MkdirTemp("", "ntfy-admin")
require.Nil(t, err)
t.Cleanup(func() { os.RemoveAll(dir) })
return dir
}
// captureLog runs f and returns what was logged meanwhile (at info level or above)
// captureLogUntil is captureLog for log lines written by goroutines: it waits until the output
// contains until
func captureLogUntil(t *testing.T, until string, f func()) string {
buf := &syncBuffer{}
log.SetOutput(buf)
log.SetLevel(log.InfoLevel)
defer func() {
log.SetLevel(log.ErrorLevel) // As in TestMain
log.SetOutput(struct{ io.Writer }{os.Stderr}) // Wrapped, so the log package does not treat stderr as a log file
}()
f()
waitFor(t, func() bool { return strings.Contains(buf.String(), until) })
return buf.String()
}
// syncBuffer is a bytes.Buffer that the log package and a test can use at the same time
type syncBuffer struct {
buf bytes.Buffer
mu sync.Mutex
}
func (b *syncBuffer) Write(p []byte) (int, error) {
b.mu.Lock()
defer b.mu.Unlock()
return b.buf.Write(p)
}
func (b *syncBuffer) String() string {
b.mu.Lock()
defer b.mu.Unlock()
return b.buf.String()
}
func captureLog(t *testing.T, f func()) string {
buf := &syncBuffer{} // Background goroutines of other tests' servers may log meanwhile, too
log.SetOutput(buf)
log.SetLevel(log.InfoLevel)
defer func() {
log.SetLevel(log.ErrorLevel) // As in TestMain
log.SetOutput(struct{ io.Writer }{os.Stderr}) // Wrapped, so the log package does not treat stderr as a log file
}()
f()
return buf.String()
}
+36
View File
@@ -0,0 +1,36 @@
//go:build !windows
package server
import (
"io/fs"
"os"
"syscall"
)
// insecureAdminSocketDir returns why other users could use or replace an admin socket in dir, or
// "" if they cannot. A directory like /tmp (world-writable, but sticky) is fine: only the owner can
// remove or rename the socket there.
func insecureAdminSocketDir(dir string) string {
info, err := os.Stat(dir)
if err != nil {
return ""
}
st, ok := info.Sys().(*syscall.Stat_t)
if !ok {
return ""
}
mode, sticky := info.Mode(), info.Mode()&fs.ModeSticky != 0
ownGroup := int(st.Gid) == os.Getgid()
switch {
case int(st.Uid) != os.Getuid() && st.Uid != 0:
return "is owned by another user"
case mode&0002 != 0 && !sticky:
return "is writable by all users"
case mode&0020 != 0 && !ownGroup && !sticky:
return "is writable by another group"
case mode&fs.ModeSetgid != 0 && !ownGroup:
return "has the setgid bit for another group, which the socket inherits"
}
return ""
}
+6
View File
@@ -0,0 +1,6 @@
package server
// insecureAdminSocketDir is not implemented on Windows, where file modes do not apply
func insecureAdminSocketDir(dir string) string {
return ""
}
+21
View File
@@ -237,6 +237,27 @@ func TestServer_Cluster_Role(t *testing.T) {
require.Equal(t, cluster.RolePassive, members[0].Role)
}
func TestServer_Cluster_StopClosesCluster(t *testing.T) {
// Stopping the server must stop the cluster: it deregisters the node, so a node starting
// after it no longer sees it, and its heartbeat does not outlive the server's database
schemaDSN := dbtest.CreateTestPostgresSchema(t)
newNode := func(nodeID string) *Server {
conf := newTestConfig(t, schemaDSN)
conf.ClusterNodeID = nodeID
conf.ClusterListen = "127.0.0.1:1" // Enables clustering; not bound since Run() is not called
conf.ClusterSecret = "s3cret"
conf.ClusterAdvertiseURL = "http://127.0.0.1:1"
return newTestServer(t, conf)
}
sA := newNode("node-a")
require.Len(t, sA.cluster.Members(), 1)
sA.Stop()
sB := newNode("node-b")
members := sB.cluster.Members()
require.Len(t, members, 1)
require.Equal(t, cluster.NodeID("node-b"), members[0].NodeID)
}
func TestServer_Cluster_ActiveWithPassivePeer(t *testing.T) {
// The setup for trying clustering beside production: the production server is the only
// active node, a second one runs passive. The active one leads right away, both deliver
+2 -1
View File
@@ -13,7 +13,8 @@ const (
contextRateVisitor contextKey = iota + 2586
contextTopic
contextMatrixPushKey
contextVisitorIP // Client IP extracted in maybeAuthenticate; reused by the abuse ban-feed (see ban.Service.Record)
contextVisitorIP // Client IP extracted in maybeAuthenticate; reused by the abuse ban-feed (see ban.Service.Record)
contextAdminSocket // Set (to true) on requests that arrived over the admin socket, see isAdminSocketRequest
)
func (s *Server) limitRequests(next handleFunc) handleFunc {
+20
View File
@@ -129,6 +129,26 @@ func TestPayments_Tiers(t *testing.T) {
require.Equal(t, int64(999111), tier.Limits.AttachmentFileSize)
require.Equal(t, int64(888111), tier.Limits.AttachmentTotalSize)
require.Equal(t, int64(3600), tier.Limits.AttachmentExpiryDuration)
// The billing list moved to /v1/account/billing/tiers; /v1/tiers still returns it to non-admins, for older web apps
require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser, false))
billing := request(t, s, "GET", "/v1/account/billing/tiers", "", nil)
require.Equal(t, 200, billing.Code)
legacy := request(t, s, "GET", "/v1/tiers", "", nil)
require.Equal(t, billing.Body.String(), legacy.Body.String())
legacy = request(t, s, "GET", "/v1/tiers", "", map[string]string{"Authorization": util.BasicAuth("ben", "ben")})
require.Equal(t, billing.Body.String(), legacy.Body.String())
// Admins get all tiers, as stored
require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin, false))
admin := request(t, s, "GET", "/v1/tiers", "", map[string]string{"Authorization": util.BasicAuth("phil", "phil")})
require.Equal(t, 200, admin.Code)
var adminTiers []apiAdminTier
require.Nil(t, json.NewDecoder(admin.Body).Decode(&adminTiers))
require.Equal(t, 3, len(adminTiers))
require.Equal(t, "ti_1", adminTiers[0].ID)
require.Equal(t, "admin", adminTiers[0].Code)
require.Equal(t, "price_123", adminTiers[2].StripeMonthlyPriceID)
})
}
+1
View File
@@ -5372,6 +5372,7 @@ func forEachBackend(t *testing.T, f func(t *testing.T, databaseURL string)) {
func newTestConfig(t *testing.T, databaseURL string) *Config {
conf := NewConfig()
conf.BaseURL = "http://127.0.0.1:12345"
conf.ListenAdminUnix = "" // Tests that need it use newTestServerWithAdminSocket
if databaseURL != "" {
conf.DatabaseURL = databaseURL
} else {
+67 -8
View File
@@ -171,20 +171,24 @@ type apiUserAddOrUpdateRequest struct {
Username string `json:"username"`
Password string `json:"password"`
Hash string `json:"hash"`
Tier string `json:"tier"`
// Do not add 'role' here. We don't want to add admins via the API.
Tier string `json:"tier"` // Tier code; apiTierReset ("-") removes the tier
Role string `json:"role"` // Only honored over the admin socket; the network API refuses it (no admins via the API)
}
type apiUserResponse struct {
Username string `json:"username"`
Role string `json:"role"`
Tier string `json:"tier,omitempty"`
Grants []*apiUserGrantResponse `json:"grants,omitempty"`
Username string `json:"username"`
Role string `json:"role"`
Tier string `json:"tier,omitempty"` // Tier code
TierName string `json:"tier_name,omitempty"` // Display name, what the CLI prints
Provisioned bool `json:"provisioned,omitempty"`
DefaultAccess string `json:"default_access,omitempty"` // Only on the "*" (everyone) entry: the server's auth-default-access
Grants []*apiUserGrantResponse `json:"grants,omitempty"`
}
type apiUserGrantResponse struct {
Topic string `json:"topic"` // This may be a pattern
Permission string `json:"permission"`
Topic string `json:"topic"` // This may be a pattern
Permission string `json:"permission"`
Provisioned bool `json:"provisioned,omitempty"`
}
type apiUserDeleteRequest struct {
@@ -202,6 +206,61 @@ type apiAccessResetRequest struct {
Topic string `json:"topic"`
}
type apiUsersTokensAddRequest struct {
Username string `json:"username"`
Label string `json:"label"`
Expires int64 `json:"expires"` // Unix seconds; 0 = never
}
type apiUsersTokensDeleteRequest struct {
Username string `json:"username"`
Token string `json:"token"`
}
type apiUserTokenResponse struct {
Username string `json:"username"`
Token string `json:"token"`
Label string `json:"label,omitempty"`
Expires int64 `json:"expires"` // Unix seconds; 0 = never
LastAccess int64 `json:"last_access,omitempty"` // Unix seconds; omitted if never used
LastOrigin string `json:"last_origin,omitempty"` // IP; omitted if never used
Provisioned bool `json:"provisioned,omitempty"`
}
type apiUsersPasswordResetRequest struct {
Username string `json:"username"`
SendEmail bool `json:"send_email"`
}
type apiUsersPasswordResetResponse struct {
Link string `json:"link"`
Email string `json:"email,omitempty"` // Address the link was sent to, if send_email was set
}
// apiAdminTier is used for both requests and responses of the admin tier API. Durations are in seconds.
// The field names match the admin UI branch (user-management), so it can rebase onto this.
type apiAdminTier struct {
ID string `json:"id,omitempty"` // Response only
Code string `json:"code"`
Name string `json:"name"`
MessagesLimit int64 `json:"messages_limit"`
MessagesExpiryDuration int64 `json:"messages_expiry_duration"`
EmailsLimit int64 `json:"emails_limit"`
CallsLimit int64 `json:"calls_limit"`
ReservationsLimit int64 `json:"reservations_limit"`
AttachmentFileSizeLimit int64 `json:"attachment_file_size_limit"`
AttachmentTotalSizeLimit int64 `json:"attachment_total_size_limit"`
AttachmentExpiryDuration int64 `json:"attachment_expiry_duration"`
AttachmentBandwidthLimit int64 `json:"attachment_bandwidth_limit"`
StripeMonthlyPriceID string `json:"stripe_monthly_price_id,omitempty"`
StripeYearlyPriceID string `json:"stripe_yearly_price_id,omitempty"`
Users int64 `json:"users"` // Response only: number of users on this tier
}
type apiAdminTierDeleteRequest struct {
Code string `json:"code"`
}
type apiAccountCreateRequest struct {
Username string `json:"username"`
Password string `json:"password"`
+22
View File
@@ -1317,6 +1317,28 @@ func (a *Manager) Tier(code string) (*Tier, error) {
return a.readTier(rows)
}
// TierUsersCount returns the number of users per tier code, including tiers without users
func (a *Manager) TierUsersCount() (map[string]int64, error) {
rows, err := a.db.Query(a.queries.selectTierUsersCount)
if err != nil {
return nil, err
}
defer rows.Close()
counts := make(map[string]int64)
for rows.Next() {
var code string
var count int64
if err := rows.Scan(&code, &count); err != nil {
return nil, err
}
counts[code] = count
}
if err := rows.Err(); err != nil {
return nil, err
}
return counts, nil
}
// TierByStripePrice returns a Tier based on the Stripe price ID, or ErrTierNotFound if it does not exist
func (a *Manager) TierByStripePrice(priceID string) (*Tier, error) {
rows, err := a.db.Query(a.queries.selectTierByPriceID, priceID, priceID)
+7
View File
@@ -214,6 +214,12 @@ const (
FROM tier
WHERE (stripe_monthly_price_id = $1 OR stripe_yearly_price_id = $2)
`
postgresSelectTierUsersCountQuery = `
SELECT t.code, COUNT(u.id)
FROM tier t
LEFT JOIN "user" u ON u.tier_id = t.id
GROUP BY t.code
`
postgresDeleteTierQuery = `DELETE FROM tier WHERE code = $1`
// Phone queries
@@ -302,6 +308,7 @@ var postgresQueries = queries{
selectTierByCode: postgresSelectTierByCodeQuery,
selectTierByPriceID: postgresSelectTierByPriceIDQuery,
updateTier: postgresUpdateTierQuery,
selectTierUsersCount: postgresSelectTierUsersCountQuery,
deleteTier: postgresDeleteTierQuery,
selectPhoneNumbers: postgresSelectPhoneNumbersQuery,
insertPhoneNumber: postgresInsertPhoneNumberQuery,
+7
View File
@@ -213,6 +213,12 @@ const (
FROM tier
WHERE (stripe_monthly_price_id = ? OR stripe_yearly_price_id = ?)
`
sqliteSelectTierUsersCountQuery = `
SELECT t.code, COUNT(u.id)
FROM tier t
LEFT JOIN user u ON u.tier_id = t.id
GROUP BY t.code
`
sqliteDeleteTierQuery = `DELETE FROM tier WHERE code = ?`
// Phone queries
@@ -300,6 +306,7 @@ var sqliteQueries = queries{
selectTierByCode: sqliteSelectTierByCodeQuery,
selectTierByPriceID: sqliteSelectTierByPriceIDQuery,
updateTier: sqliteUpdateTierQuery,
selectTierUsersCount: sqliteSelectTierUsersCountQuery,
deleteTier: sqliteDeleteTierQuery,
selectPhoneNumbers: sqliteSelectPhoneNumbersQuery,
insertPhoneNumber: sqliteInsertPhoneNumberQuery,
+16
View File
@@ -1019,6 +1019,22 @@ func TestAccount_Tier_Create_With_ID(t *testing.T) {
})
}
func TestManager_TierUsersCount(t *testing.T) {
forEachBackend(t, func(t *testing.T, newManager newManagerFunc) {
a := newTestManager(t, newManager, PermissionDenyAll)
require.Nil(t, a.AddTier(&Tier{Code: "pro", Name: "Pro"}))
require.Nil(t, a.AddTier(&Tier{Code: "empty", Name: "Empty"}))
require.Nil(t, a.AddUser("ben", "ben", RoleUser, false))
require.Nil(t, a.AddUser("emma", "emma", RoleUser, false))
require.Nil(t, a.ChangeTier("ben", "pro"))
require.Nil(t, a.ChangeTier("emma", "pro"))
counts, err := a.TierUsersCount()
require.Nil(t, err)
require.Equal(t, map[string]int64{"pro": 2, "empty": 0}, counts)
})
}
func TestManager_Tier_Change_And_Reset(t *testing.T) {
forEachBackend(t, func(t *testing.T, newManager newManagerFunc) {
a := newTestManager(t, newManager, PermissionDenyAll)
+7 -6
View File
@@ -408,12 +408,13 @@ type queries struct {
deleteExcessTokens string
// Tier queries
insertTier string
selectTiers string
selectTierByCode string
selectTierByPriceID string
updateTier string
deleteTier string
insertTier string
selectTiers string
selectTierByCode string
selectTierByPriceID string
selectTierUsersCount string
updateTier string
deleteTier string
// Phone queries
selectPhoneNumbers string
+1 -1
View File
@@ -12,7 +12,7 @@ test.describe("upgrade banner", () => {
await route.fulfill({ response, body });
});
const limits = { messages: 250, messages_expiry_duration: 43200, emails: 5, calls: 0, reservations: 0 };
await page.route("**/v1/tiers", (route) => route.fulfill({ json: [{ limits }] }));
await page.route("**/v1/account/billing/tiers", (route) => route.fulfill({ json: [{ limits }] }));
});
test("does not cover the bottom of a long topic list", async ({ page }) => {
+68
View File
@@ -9,6 +9,31 @@ let topic;
// Browser notifications the app showed, recorded by the init script in beforeEach
const shown = (page) => page.evaluate(() => window.shownNotifications.map((n) => n.body));
// Publishes messages within one second, right after a second starts. A since=<id> marker replays
// every message from its second, so the earlier ones come back to a client that has the last one.
const publishInOneSecond = async (request, topic, messages) => {
await new Promise((resolve) => setTimeout(resolve, 1000 - (Date.now() % 1000)));
const times = [];
for (const message of messages) {
times.push((await (await request.post(`/${topic}`, { data: message })).json()).time);
}
expect(new Set(times).size, `published in one second: ${times}`).toBe(1);
};
// Messages the topic's WebSocket received from the server, from now on
const liveMessages = (page, topic) => {
const received = [];
page.on("websocket", (ws) => {
if (new URL(ws.url()).pathname.endsWith(`/${topic}/ws`)) {
ws.on("framereceived", (frame) => {
const event = JSON.parse(frame.payload);
if (event.event === "message") received.push(event.message);
});
}
});
return received;
};
test.beforeEach(async ({ page }) => {
// Record browser notifications instead of showing them; headless Chrome can't show them anyway
await page.addInitScript(() => {
@@ -268,6 +293,49 @@ test.describe("sequences", () => {
});
});
test.describe("replays", () => {
test("deleted messages don't come back when a reconnect replays them", async ({ page, request }) => {
await publishInOneSecond(request, topic, ["first", "second", "third"]);
const items = page.getByRole("listitem", { name: "Notification" });
await expect(items).toHaveCount(3);
for (const message of ["first", "second", "third"]) {
await items.filter({ hasText: message }).getByRole("button", { name: "Delete" }).click();
}
await expect(items).toHaveCount(0);
// Reconnecting and the startup poll ask for since=<last>, and get the rest of its second again
const live = liveMessages(page, topic);
const polled = page.waitForResponse((r) => new URL(r.url()).pathname.endsWith(`/${topic}/json`));
await page.reload();
await page.clock.runFor(2000);
expect(await (await polled).text()).toContain("first");
await expect.poll(() => live).toContain("first");
await page.waitForTimeout(1000); // Let the app store what it got
await expect(items).toHaveCount(0);
await expect(page.getByText("You haven't received any notifications for this topic yet.")).toBeVisible();
});
test("a message replayed on reconnect stays read and doesn't notify again", async ({ page, request }) => {
await publishInOneSecond(request, topic, ["first", "second"]);
await expect(page.getByRole("listitem", { name: "Notification" })).toHaveCount(2);
// Read them by opening the topic, then reconnect from another one, so one that comes back unread is counted
const other = unique("other");
await subscribe(page, other);
await expect(page).toHaveTitle("(2) ntfy");
await page.getByRole("button", { name: topic, exact: true }).click();
await expect(page).toHaveTitle("ntfy");
await page.getByRole("button", { name: other, exact: true }).click();
const live = liveMessages(page, topic);
await page.reload();
await page.clock.runFor(2000);
await expect.poll(() => live).toContain("first");
await page.waitForTimeout(1000); // Let the app store what it got
await expect(page).toHaveTitle("ntfy");
expect(await shown(page)).toEqual([]);
});
});
test.describe("browser notifications", () => {
test("a new message is shown", async ({ page, request }) => {
await request.post(`/${topic}`, { data: "pop up" });
+4 -4
View File
@@ -17,7 +17,7 @@
"dexie-react-hooks": "^4.4.0",
"i18next": "^26.4.4",
"i18next-browser-languagedetector": "^8.2.1",
"i18next-http-backend": "^4.0.0",
"i18next-http-backend": "^4.0.4",
"react": "*",
"react-dom": "*",
"react-i18next": "^17.0.16",
@@ -5642,9 +5642,9 @@
}
},
"node_modules/i18next-http-backend": {
"version": "4.0.2",
"resolved": "https://registry.npmjs.org/i18next-http-backend/-/i18next-http-backend-4.0.2.tgz",
"integrity": "sha512-oay62dIB2kL7+WHzoUXBjWfL3+mwijD3pQkQkIEaRLhy6kjXxUhrRenV0gInwlCASAaJaDy94+XvYizK+cAGdA==",
"version": "4.0.4",
"resolved": "https://registry.npmjs.org/i18next-http-backend/-/i18next-http-backend-4.0.4.tgz",
"integrity": "sha512-nG9OQKvUsPiTVUBg1rCGmvmWhzxC9GOxVbGdTYyejHVMSUjutt1hiBNVSzhB/ZE7keL9Xkb880kUfhDovV5T5g==",
"license": "MIT",
"engines": {
"node": ">=18"
+1 -1
View File
@@ -23,7 +23,7 @@
"dexie-react-hooks": "^4.4.0",
"i18next": "^26.4.4",
"i18next-browser-languagedetector": "^8.2.1",
"i18next-http-backend": "^4.0.0",
"i18next-http-backend": "^4.0.4",
"react": "latest",
"react-dom": "latest",
"react-i18next": "^17.0.16",
+5
View File
@@ -64,6 +64,11 @@ const handlePushMessage = async (data) => {
}),
);
// Don't store a message again that the user deleted (e.g. delivered via WebSocket first)
if ((await db.notifications.get(message.id))?.deleted) {
return;
}
// Delete existing notification with same sequence ID (if any)
const sequenceId = message.sequence_id || message.id;
if (sequenceId) {
+34 -6
View File
@@ -176,7 +176,7 @@ export class SubscriptionManager {
return this.db.notifications
.orderBy("time") // Sort by time
.filter((n) => n.subscriptionId === subscriptionId)
.filter((n) => n.subscriptionId === subscriptionId && !n.deleted)
.reverse()
.toArray();
}
@@ -184,6 +184,7 @@ export class SubscriptionManager {
async getAllNotifications() {
return this.db.notifications
.orderBy("time") // Efficient, see docs
.filter((n) => !n.deleted)
.reverse()
.toArray();
}
@@ -200,9 +201,24 @@ export class SubscriptionManager {
// Add notification to database; check and add in one transaction, so a poll storing the
// same message concurrently (see addNotifications) can't slip in between
const added = await this.db.transaction("rw", this.db.notifications, async () => {
if (await this.db.notifications.get(notification.id)) {
// A replayed message keeps its local state (e.g. read or deleted), but one that only a poll
// stored (no "new" marker, see addNotifications) wasn't delivered live yet, so it's new
const existing = await this.db.notifications.get(notification.id);
if (existing && existing.new === undefined) {
await this.db.notifications.update(notification.id, { new: 1 });
return true;
} else if (existing) {
return false;
}
// An update replaces the stored versions of its sequence, unless a newer one is stored
// already (e.g. a reconnect replayed an older version), or was deleted
const sequenceId = notification.sequence_id || notification.id;
const versions = await this.db.notifications.where({ subscriptionId, sequenceId }).toArray();
if (versions.some((v) => v.time > notification.time)) {
return false;
}
await this.db.notifications.bulkDelete(versions.map((v) => v.id));
await this.db.notifications.add({
...messageWithSequenceId(notification),
subscriptionId,
@@ -230,9 +246,9 @@ export class SubscriptionManager {
* older versions of the same sequence; will not throw if they exist
*/
async addNotifications(subscriptionId, notifications) {
await this.db.transaction("rw", this.db.notifications, async () => {
const added = await this.db.transaction("rw", this.db.notifications, async () => {
// Skip notifications that are already stored (e.g. delivered via WebSocket while this poll
// was in flight), so overwriting them doesn't drop their "new" marker
// was in flight), so overwriting them doesn't drop their "new" marker, or deleted
const existing = await this.db.notifications.bulkGet(notifications.map((n) => n.id));
const notificationsWithSubscriptionId = notifications
.filter((_, i) => !existing[i])
@@ -252,7 +268,14 @@ export class SubscriptionManager {
const replaced = versions.flat().filter((v) => updates.some((n) => n.sequenceId === v.sequenceId));
await this.db.notifications.bulkDelete(replaced.map((v) => v.id));
await this.db.notifications.bulkPut(updates);
return notificationsWithSubscriptionId.length > 0;
});
// Don't move the marker if the poll only replayed known messages: a replay of the marker's second
// ends with a message older than the marker, and moving back to it replays the marker again
if (!added) {
return;
}
const lastNotificationId = notifications.at(-1).id;
await this.db.subscriptions.update(subscriptionId, {
last: lastNotificationId,
@@ -272,8 +295,13 @@ export class SubscriptionManager {
return true;
}
/**
* Marks a notification as deleted instead of removing it: the server replays the whole second of a
* since=<id> marker, so a poll can return it again, and the stored row keeps it from being re-added.
* Deleted notifications are pruned with the others.
*/
async deleteNotification(notificationId) {
await this.db.notifications.delete(notificationId);
await this.db.notifications.where({ id: notificationId }).modify({ deleted: 1, new: 0 });
}
async deleteNotificationBySequenceId(subscriptionId, sequenceId) {
@@ -281,7 +309,7 @@ export class SubscriptionManager {
}
async deleteNotifications(subscriptionId) {
await this.db.notifications.where({ subscriptionId }).delete();
await this.db.notifications.where({ subscriptionId }).modify({ deleted: 1, new: 0 });
}
async markNotificationRead(notificationId) {
+115
View File
@@ -158,6 +158,37 @@ describe("SubscriptionManager.addNotification", () => {
expect((await db.subscriptions.get(subscriptionId)).last).toBe("m2");
});
it("replaces an older version of the same sequence", async () => {
const { db, manager } = await setup();
await manager.addNotification(subscriptionId, message("m1", { sequence_id: "s1", message: "Downloading" }));
expect(
await manager.addNotification(subscriptionId, message("m2", { sequence_id: "s1", message: "Download complete", time: 1700000001 })),
).toBe(true);
expect((await db.notifications.toArray()).map((n) => [n.id, n.new])).toEqual([["m2", 1]]);
});
it("does not replace a newer version of the same sequence when the WebSocket replays an older one", async () => {
const { db, manager } = await setup();
await manager.addNotification(subscriptionId, message("m2", { sequence_id: "s1", message: "Download complete", time: 1700000001 }));
expect(await manager.addNotification(subscriptionId, message("m1", { sequence_id: "s1", message: "Downloading" }))).toBe(false);
expect((await db.notifications.toArray()).map((n) => n.message)).toEqual(["Download complete"]);
expect((await db.subscriptions.get(subscriptionId)).last).toBe("m2");
});
it("marks a message the poll stored as unread when the WebSocket delivers it", async () => {
const { db, manager } = await setup();
await manager.addNotifications(subscriptionId, [message("m1")]);
// The poll stored it before the live delivery arrived; the user hasn't seen it yet
expect(await manager.addNotification(subscriptionId, message("m1"))).toBe(true);
expect((await db.notifications.get("m1")).new).toBe(1);
});
it("does not store delete and clear events", async () => {
const { db, manager } = await setup();
expect(await manager.addNotification(subscriptionId, message("d1", { event: "message_delete", sequence_id: "s1" }))).toBe(false);
@@ -233,3 +264,87 @@ describe("SubscriptionManager.addNotifications", () => {
expect((await db.notifications.toArray()).map((n) => n.message)).toEqual(["Download complete"]);
});
});
const visibleIds = async (manager) => (await manager.getNotifications(subscriptionId)).map((n) => n.id).sort();
// The server replays the whole second of a since=<id> marker, so a poll returns messages the
// client already has, or had and deleted (#2098)
describe("SubscriptionManager deleted notifications", () => {
it("does not add a deleted message again when a poll replays it", async () => {
const { manager } = await setup();
await manager.addNotifications(subscriptionId, [message("m1"), message("m2"), message("m3")]);
await manager.deleteNotification("m1");
await manager.deleteNotification("m2");
await manager.addNotifications(subscriptionId, [message("m1"), message("m2")]);
expect(await visibleIds(manager)).toEqual(["m3"]);
});
it("does not add a deleted message again when the WebSocket replays it", async () => {
const { manager } = await setup();
await manager.addNotification(subscriptionId, message("m1"));
await manager.deleteNotification("m1");
expect(await manager.addNotification(subscriptionId, message("m1"))).toBe(false);
expect(await visibleIds(manager)).toEqual([]);
});
it("does not add older versions of a deleted sequence again, but newer ones", async () => {
const { manager } = await setup();
await manager.addNotifications(subscriptionId, [message("m2", { sequence_id: "s1", message: "Download complete", time: 1700000001 })]);
await manager.deleteNotification("m2");
await manager.addNotifications(subscriptionId, [message("m1", { sequence_id: "s1", message: "Downloading" })]);
expect(await visibleIds(manager)).toEqual([]);
await manager.addNotification(subscriptionId, message("m3", { sequence_id: "s1", message: "Download failed", time: 1700000002 }));
expect(await visibleIds(manager)).toEqual(["m3"]);
});
it("does not add messages again after deleting all of them", async () => {
const { manager } = await setup();
await manager.addNotifications(subscriptionId, [message("m1"), message("m2")]);
await manager.deleteNotifications(subscriptionId);
await manager.addNotifications(subscriptionId, [message("m1"), message("m2")]);
expect(await visibleIds(manager)).toEqual([]);
});
it("keeps the last message when a poll only replays known or deleted messages", async () => {
const { db, manager } = await setup();
await manager.addNotifications(subscriptionId, [message("m1"), message("m2"), message("m3")]);
await manager.deleteNotification("m1");
await manager.addNotifications(subscriptionId, [message("m1"), message("m2")]);
expect((await db.subscriptions.get(subscriptionId)).last).toBe("m3");
});
it("does not list deleted messages or count them as new", async () => {
const { manager } = await setup();
await manager.addNotification(subscriptionId, message("m1"));
await manager.addNotification(subscriptionId, message("m2"));
await manager.deleteNotification("m1");
expect((await manager.getAllNotifications()).map((n) => n.id)).toEqual(["m2"]);
expect((await manager.all()).find((s) => s.id === subscriptionId).new).toBe(1);
});
it("forgets deleted messages when the subscription is removed or they are pruned", async () => {
const { manager } = await setup();
await manager.addNotifications(subscriptionId, [message("old", { time: 1600000000 }), message("m1")]);
await manager.deleteNotifications(subscriptionId);
await manager.pruneNotifications(1650000000);
await manager.addNotifications(subscriptionId, [message("old", { time: 1600000000 })]);
expect(await visibleIds(manager)).toEqual(["old"]);
await manager.remove({ id: subscriptionId });
await manager.upsert(baseUrl, "mytopic");
await manager.addNotifications(subscriptionId, [message("m1")]);
expect(await visibleIds(manager)).toEqual(["m1"]);
});
});
+1 -1
View File
@@ -9,7 +9,7 @@ import config from "./config";
import emojisMapped from "./emojisMapped";
import { THEME, DATE_FORMAT, TIME_FORMAT } from "./Prefs";
export const tiersUrl = (baseUrl) => `${baseUrl}/v1/tiers`;
export const tiersUrl = (baseUrl) => `${baseUrl}/v1/account/billing/tiers`;
export const shortUrl = (url) => url.replaceAll(/https?:\/\//g, "");
export const expandUrl = (url) => [`https://${url}`, `http://${url}`];
export const expandSecureUrl = (url) => `https://${url}`;
+2 -5
View File
@@ -65,11 +65,8 @@ export const useConnectionListeners = (account, subscriptions, users, webPushTop
await subscriptionManager.markNotificationReadBySequenceId(subscription.id, notification.sequence_id);
await notifier.cancel(subscription, notification);
} else {
// Regular message: delete existing and add new
const sequenceId = notification.sequence_id || notification.id;
if (sequenceId) {
await subscriptionManager.deleteNotificationBySequenceId(subscription.id, sequenceId);
}
// Regular message: add it, replacing older versions of its sequence; a replayed duplicate
// keeps its local state (e.g. read) and doesn't notify again
const added = await subscriptionManager.addNotification(subscription.id, notification);
if (added) {
await subscriptionManager.notify(subscription.id, notification);