diff --git a/cmd/publish_unix.go b/cmd/publish_unix.go index d2b49a5e..560c3f64 100644 --- a/cmd/publish_unix.go +++ b/cmd/publish_unix.go @@ -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) } diff --git a/cmd/publish_unix_test.go b/cmd/publish_unix_test.go new file mode 100644 index 00000000..16eea94b --- /dev/null +++ b/cmd/publish_unix_test.go @@ -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)) +} diff --git a/cmd/serve.go b/cmd/serve.go index b468be3a..f9b20164 100644 --- a/cmd/serve.go +++ b/cmd/serve.go @@ -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 diff --git a/cmd/serve_test.go b/cmd/serve_test.go index 87c0abc1..388835bd 100644 --- a/cmd/serve_test.go +++ b/cmd/serve_test.go @@ -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)) diff --git a/db/db.go b/db/db.go index 695cd0e2..e93d477a 100644 --- a/db/db.go +++ b/db/db.go @@ -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) } } } diff --git a/db/db_test.go b/db/db_test.go index c2105f92..4537f6a8 100644 --- a/db/db_test.go +++ b/db/db_test.go @@ -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) +} diff --git a/db/types.go b/db/types.go index e9309d04..c849fcec 100644 --- a/db/types.go +++ b/db/types.go @@ -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 } diff --git a/docs/config.md b/docs/config.md index 72281c3b..5e7669d0 100644 --- a/docs/config.md +++ b/docs/config.md @@ -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] diff --git a/docs/releases.md b/docs/releases.md index ea347d38..e026969e 100644 --- a/docs/releases.md +++ b/docs/releases.md @@ -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:** diff --git a/docs/subscribe/cli.md b/docs/subscribe/cli.md index 36388c71..ae104886 100644 --- a/docs/subscribe/cli.md +++ b/docs/subscribe/cli.md @@ -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, diff --git a/go.mod b/go.mod index de789760..55e33131 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index 9453eac8..e65b342e 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/metrics/metrics.go b/metrics/metrics.go index fdf38d4a..3287aa30 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -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) } } } diff --git a/metrics/metrics_test.go b/metrics/metrics_test.go index 45bfdfea..0563622d 100644 --- a/metrics/metrics_test.go +++ b/metrics/metrics_test.go @@ -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) diff --git a/server/config.go b/server/config.go index 33728b85..05b4d54a 100644 --- a/server/config.go +++ b/server/config.go @@ -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: "", diff --git a/server/errors.go b/server/errors.go index a4474487..79bc8cbd 100644 --- a/server/errors.go +++ b/server/errors.go @@ -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} diff --git a/server/log.go b/server/log.go index 8fbeab31..7c87d6cc 100644 --- a/server/log.go +++ b/server/log.go @@ -30,6 +30,7 @@ const ( tagWebsocket = "websocket" tagMatrix = "matrix" tagWebPush = "webpush" + tagAdmin = "admin" ) var ( diff --git a/server/ntfy.service b/server/ntfy.service index df420404..8f95a716 100644 --- a/server/ntfy.service +++ b/server/ntfy.service @@ -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 diff --git a/server/server.go b/server/server.go index cc8592aa..452a253b 100644 --- a/server/server.go +++ b/server/server.go @@ -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 { diff --git a/server/server.yml b/server/server.yml index ff2a365c..c44c0c16 100644 --- a/server/server.yml +++ b/server/server.yml @@ -31,6 +31,13 @@ # listen-unix: # listen-unix-mode: +# 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: diff --git a/server/server_admin.go b/server/server_admin.go index 00be9bed..a05618e2 100644 --- a/server/server_admin.go +++ b/server/server_admin.go @@ -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) +} diff --git a/server/server_admin_test.go b/server/server_admin_test.go index f5901fbb..40362c0d 100644 --- a/server/server_admin_test.go +++ b/server/server_admin_test.go @@ -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() +} diff --git a/server/server_admin_unix.go b/server/server_admin_unix.go new file mode 100644 index 00000000..5710fb12 --- /dev/null +++ b/server/server_admin_unix.go @@ -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 "" +} diff --git a/server/server_admin_windows.go b/server/server_admin_windows.go new file mode 100644 index 00000000..dba7ccbe --- /dev/null +++ b/server/server_admin_windows.go @@ -0,0 +1,6 @@ +package server + +// insecureAdminSocketDir is not implemented on Windows, where file modes do not apply +func insecureAdminSocketDir(dir string) string { + return "" +} diff --git a/server/server_cluster_test.go b/server/server_cluster_test.go index 47fe9804..c3e4b08e 100644 --- a/server/server_cluster_test.go +++ b/server/server_cluster_test.go @@ -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 diff --git a/server/server_middleware.go b/server/server_middleware.go index 117be70d..4f41e264 100644 --- a/server/server_middleware.go +++ b/server/server_middleware.go @@ -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 { diff --git a/server/server_payments_test.go b/server/server_payments_test.go index e1df26d7..88243490 100644 --- a/server/server_payments_test.go +++ b/server/server_payments_test.go @@ -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) }) } diff --git a/server/server_test.go b/server/server_test.go index d5d01048..cb901196 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -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 { diff --git a/server/types.go b/server/types.go index c4b20e05..f5b5ebd8 100644 --- a/server/types.go +++ b/server/types.go @@ -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"` diff --git a/user/manager.go b/user/manager.go index 41b068a5..9cc21f3b 100644 --- a/user/manager.go +++ b/user/manager.go @@ -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) diff --git a/user/manager_postgres.go b/user/manager_postgres.go index 645495f4..7adaabc8 100644 --- a/user/manager_postgres.go +++ b/user/manager_postgres.go @@ -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, diff --git a/user/manager_sqlite.go b/user/manager_sqlite.go index 73fe8377..277245bc 100644 --- a/user/manager_sqlite.go +++ b/user/manager_sqlite.go @@ -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, diff --git a/user/manager_test.go b/user/manager_test.go index 90e7cdb3..f5f44c00 100644 --- a/user/manager_test.go +++ b/user/manager_test.go @@ -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) diff --git a/user/types.go b/user/types.go index a69041fe..6abb8e3a 100644 --- a/user/types.go +++ b/user/types.go @@ -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 diff --git a/web/e2e/navigation.spec.js b/web/e2e/navigation.spec.js index 8e33fab8..ded2dfbe 100644 --- a/web/e2e/navigation.spec.js +++ b/web/e2e/navigation.spec.js @@ -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 }) => { diff --git a/web/e2e/notifications.spec.js b/web/e2e/notifications.spec.js index 9ce83ea7..5f422100 100644 --- a/web/e2e/notifications.spec.js +++ b/web/e2e/notifications.spec.js @@ -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= 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=, 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" }); diff --git a/web/package-lock.json b/web/package-lock.json index a79534df..c8dba3a5 100644 --- a/web/package-lock.json +++ b/web/package-lock.json @@ -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" diff --git a/web/package.json b/web/package.json index 90381300..5ca69a02 100644 --- a/web/package.json +++ b/web/package.json @@ -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", diff --git a/web/public/sw.js b/web/public/sw.js index 1c0235d3..2730ec5a 100644 --- a/web/public/sw.js +++ b/web/public/sw.js @@ -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) { diff --git a/web/src/app/SubscriptionManager.js b/web/src/app/SubscriptionManager.js index e1776833..6014a3a4 100644 --- a/web/src/app/SubscriptionManager.js +++ b/web/src/app/SubscriptionManager.js @@ -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= 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) { diff --git a/web/src/app/SubscriptionManager.test.js b/web/src/app/SubscriptionManager.test.js index 15098463..82f3b169 100644 --- a/web/src/app/SubscriptionManager.test.js +++ b/web/src/app/SubscriptionManager.test.js @@ -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= 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"]); + }); +}); diff --git a/web/src/app/utils.js b/web/src/app/utils.js index 58b9118e..cb343ebd 100644 --- a/web/src/app/utils.js +++ b/web/src/app/utils.js @@ -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}`; diff --git a/web/src/components/hooks.js b/web/src/components/hooks.js index ba62f54f..150e7540 100644 --- a/web/src/components/hooks.js +++ b/web/src/components/hooks.js @@ -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);