added superuser ips whitelist

core/base.go

@@ -38,8 +38,9 @@ const (
 
 	LocalStorageDirName       string = "storage"
 	LocalBackupsDirName       string = "backups"
-	LocalTempDirName          string = ".pb_temp_to_delete" // temp pb_data sub directory that will be deleted on each app.Bootstrap()
 	LocalAutocertCacheDirName string = ".autocert_cache"
+	LocalNotifyDirName        string = ".notify"            // optional watched directory that is used as a cross-platform workaround for synchronizing various runtime states between multiple PocketBase instances pointing to the same pb_data
+	LocalTempDirName          string = ".pb_temp_to_delete" // temp pb_data sub directory that will be deleted on each app.Bootstrap()
 
 	// @todo consider removing after backups refactoring
 	lostFoundDirName string = "lost+found"
@@ -1382,6 +1383,7 @@ func (app *BaseApp) registerBaseHooks() {
 	app.registerMFAHooks()
 	app.registerOTPHooks()
 	app.registerAuthOriginHooks()
+	app.registerNotifyWatcherHooks()
 }
 
 // getLoggerMinLevel returns the logger min level based on the

core/notify_watcher.go

@@ -0,0 +1,210 @@
+package core
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/fatih/color"
+	"github.com/fsnotify/fsnotify"
+	"github.com/pocketbase/pocketbase/tools/hook"
+	"github.com/pocketbase/pocketbase/tools/security"
+)
+
+const systemHookIdNotifyWatcher = "__pbNotifyWatcherSystemHook__"
+
+func (app *BaseApp) registerNotifyWatcherHooks() {
+	var notifyWatcher *fsnotify.Watcher
+
+	instanceId := "@" + security.PseudorandomString(10)
+
+	localNotifyDirPath := filepath.Join(app.DataDir(), LocalNotifyDirName)
+	settingsFile := filepath.Join(localNotifyDirPath, "settings"+instanceId)
+	collectionsFile := filepath.Join(localNotifyDirPath, "collections"+instanceId)
+
+	// init
+	app.OnBootstrap().Bind(&hook.Handler[*BootstrapEvent]{
+		Id: systemHookIdNotifyWatcher,
+		Func: func(e *BootstrapEvent) error {
+			err := e.Next()
+			if err != nil {
+				return err
+			}
+
+			if notifyWatcher != nil {
+				_ = notifyWatcher.Close()
+			}
+
+			notifyWatcher, err = createNotifyDirWatcher(e.App, instanceId, localNotifyDirPath)
+			if err != nil {
+				e.App.Logger().Warn("Notify dir watcher failure.", "error", err)
+			}
+
+			return nil
+		},
+		Priority: -998,
+	})
+
+	// cleanup
+	app.OnTerminate().Bind(&hook.Handler[*TerminateEvent]{
+		Id: systemHookIdNotifyWatcher,
+		Func: func(e *TerminateEvent) error {
+			if notifyWatcher != nil {
+				_ = notifyWatcher.Close()
+			}
+
+			_ = os.Remove(settingsFile)
+			_ = os.Remove(collectionsFile)
+
+			return e.Next()
+		},
+		Priority: -998,
+	})
+
+	// ---------------------------------------------------------------
+
+	settingsNotify := func(e *ModelEvent) error {
+		err := e.Next()
+		if err != nil || e.Model.PK() != paramsKeySettings {
+			return err
+		}
+
+		if notifyWatcher != nil {
+			if err := os.WriteFile(settingsFile, nil, 0644); err != nil {
+				e.App.Logger().Warn("Failed to write watcher file", "error", err, "file", settingsFile)
+			}
+			_ = os.Remove(settingsFile)
+		}
+
+		return nil
+	}
+	app.OnModelAfterCreateSuccess(paramsTable).Bind(&hook.Handler[*ModelEvent]{
+		Id:       systemHookIdNotifyWatcher,
+		Func:     settingsNotify,
+		Priority: 999,
+	})
+	app.OnModelAfterUpdateSuccess(paramsTable).Bind(&hook.Handler[*ModelEvent]{
+		Id:       systemHookIdNotifyWatcher,
+		Func:     settingsNotify,
+		Priority: 999,
+	})
+
+	// ---------------------------------------------------------------
+
+	collectionsNotify := func(e *CollectionEvent) error {
+		if err := e.Next(); err != nil {
+			return err
+		}
+
+		if notifyWatcher != nil {
+			if err := os.WriteFile(collectionsFile, nil, 0644); err != nil {
+				e.App.Logger().Warn("Failed to write watcher file", "error", err, "file", collectionsFile)
+			}
+			_ = os.Remove(collectionsFile)
+		}
+
+		return nil
+	}
+	app.OnCollectionAfterCreateSuccess().Bind(&hook.Handler[*CollectionEvent]{
+		Id:       systemHookIdNotifyWatcher,
+		Func:     collectionsNotify,
+		Priority: 999,
+	})
+	app.OnCollectionAfterUpdateSuccess().Bind(&hook.Handler[*CollectionEvent]{
+		Id:       systemHookIdNotifyWatcher,
+		Func:     collectionsNotify,
+		Priority: 999,
+	})
+	app.OnCollectionAfterDeleteSuccess().Bind(&hook.Handler[*CollectionEvent]{
+		Id:       systemHookIdNotifyWatcher,
+		Func:     collectionsNotify,
+		Priority: 999,
+	})
+}
+
+func createNotifyDirWatcher(app App, instanceId string, localNotifyDirPath string) (*fsnotify.Watcher, error) {
+	// create the notify dir (if not already)
+	err := os.MkdirAll(localNotifyDirPath, os.ModePerm)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create a notify dir: %w", err)
+	}
+
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		return nil, fmt.Errorf("failed to init notify dir watcher: %w", err)
+	}
+
+	err = watcher.Add(localNotifyDirPath)
+	if err != nil {
+		_ = watcher.Close()
+		return nil, fmt.Errorf("unable to watch notify dir: %w", err)
+	}
+
+	var debounceTimer *time.Timer
+
+	stopDebounceTimer := func() {
+		if debounceTimer != nil {
+			debounceTimer.Stop()
+			debounceTimer = nil
+		}
+	}
+
+	// watch
+	go func() {
+		defer stopDebounceTimer()
+
+		for {
+			select {
+			case event, ok := <-watcher.Events:
+				if !ok {
+					return
+				}
+
+				// modified from within the current app instance or cleanup event
+				if strings.HasSuffix(event.Name, instanceId) || event.Has(fsnotify.Remove) || !app.IsBootstrapped() {
+					continue
+				}
+
+				stopDebounceTimer()
+
+				debounceTimer = time.AfterFunc(50*time.Millisecond, func() {
+					filename := filepath.Base(event.Name)
+
+					// settings changed
+					if strings.HasPrefix(filename, "settings@") {
+						app.Logger().Debug("Reloading settings after notify event")
+
+						err := app.ReloadSettings()
+						if err != nil {
+							app.Logger().Warn("Failed to reload app settings after notify", "error", err)
+						}
+						return
+					}
+
+					// collections changed
+					if strings.HasPrefix(filename, "collections@") {
+						app.Logger().Debug("Reloading cached collections after notify event")
+
+						err := app.ReloadCachedCollections()
+						if err != nil {
+							app.Logger().Warn("Failed to reload cached collections after notify", "error", err)
+						}
+						return
+					}
+				})
+			case err, ok := <-watcher.Errors:
+				if app.IsDev() && err != nil {
+					color.Red("Notify dir watch error:", err)
+				}
+
+				if !ok {
+					return
+				}
+			}
+		}
+	}()
+
+	return watcher, err
+}

core/notify_watcher_test.go

@@ -0,0 +1,185 @@
+package core_test
+
+import (
+	"context"
+	"database/sql"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/pocketbase/dbx"
+	"github.com/pocketbase/pocketbase/core"
+	"github.com/pocketbase/pocketbase/tools/store"
+	"golang.org/x/sync/semaphore"
+)
+
+func TestNotifyWatcher_SettingsUpdate(t *testing.T) {
+	t.Parallel()
+
+	testEvents := store.New[core.App, int](nil)
+
+	tmpDir, err := os.MkdirTemp("", "pb_notify_test*")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(tmpDir)
+
+	app1 := core.NewBaseApp(core.BaseAppConfig{
+		DataDir: tmpDir,
+	})
+	if err := app1.Bootstrap(); err != nil {
+		t.Fatal(err)
+	}
+
+	app2 := core.NewBaseApp(core.BaseAppConfig{
+		DataDir: tmpDir,
+	})
+	if err := app2.Bootstrap(); err != nil {
+		t.Fatal(err)
+	}
+
+	ctx, cancelCtx := context.WithTimeout(context.Background(), 1*time.Second)
+	defer cancelCtx()
+
+	sem := semaphore.NewWeighted(1)
+	sem.Acquire(ctx, 1)
+
+	app1.OnSettingsReload().BindFunc(func(e *core.SettingsReloadEvent) error {
+		testEvents.SetFunc(app1, func(old int) int {
+			return old + 1
+		})
+		return e.Next()
+	})
+
+	app2.OnSettingsReload().BindFunc(func(e *core.SettingsReloadEvent) error {
+		testEvents.SetFunc(app2, func(old int) int {
+			sem.Release(1)
+
+			return old + 1
+		})
+		return e.Next()
+	})
+
+	// updating app1 settings should trigger a reload in app2
+	app1.Settings().SuperuserIPs = []string{"127.0.0.1"}
+	if err := app1.Save(app1.Settings()); err != nil {
+		t.Fatal(err)
+	}
+
+	// block until released or timeouted
+	sem.Acquire(ctx, 1)
+
+	if app1Total := testEvents.Get(app1); app1Total != 1 {
+		t.Fatalf("Expected 1 app1 event, got %d", app1Total)
+	}
+
+	if app2Total := testEvents.Get(app2); app2Total != 1 {
+		t.Fatalf("Expected 1 app2 event, got %d", app2Total)
+	}
+
+	app2SuperuserIPs := app2.Settings().SuperuserIPs
+	if len(app2SuperuserIPs) != 1 || app2SuperuserIPs[0] != "127.0.0.1" {
+		t.Fatalf("Expected exactly 127.0.0.1 superuser IP in app2 settings event, got %v", app2SuperuserIPs)
+	}
+}
+
+func TestNotifyWatcher_CollectionsUpdate(t *testing.T) {
+	t.Parallel()
+
+	tmpDir, err := os.MkdirTemp("", "pb_notify_test*")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(tmpDir)
+
+	app1 := core.NewBaseApp(core.BaseAppConfig{
+		DataDir: tmpDir,
+	})
+	if err := app1.Bootstrap(); err != nil {
+		t.Fatal(err)
+	}
+
+	app2 := core.NewBaseApp(core.BaseAppConfig{
+		DataDir: tmpDir,
+	})
+	if err := app2.Bootstrap(); err != nil {
+		t.Fatal(err)
+	}
+
+	testQueries := store.New[string, []string](nil)
+	app2.ConcurrentDB().(*dbx.DB).QueryLogFunc = func(ctx context.Context, t time.Duration, sql string, rows *sql.Rows, err error) {
+		testQueries.SetFunc("concurrent", func(old []string) []string {
+			return append(old, sql)
+		})
+	}
+	app2.ConcurrentDB().(*dbx.DB).ExecLogFunc = func(ctx context.Context, t time.Duration, sql string, result sql.Result, err error) {
+		testQueries.SetFunc("concurrent", func(old []string) []string {
+			return append(old, sql)
+		})
+	}
+	app2.NonconcurrentDB().(*dbx.DB).QueryLogFunc = func(ctx context.Context, t time.Duration, sql string, rows *sql.Rows, err error) {
+		testQueries.SetFunc("nonconcurrent", func(old []string) []string {
+			return append(old, sql)
+		})
+	}
+	app2.NonconcurrentDB().(*dbx.DB).ExecLogFunc = func(ctx context.Context, t time.Duration, sql string, result sql.Result, err error) {
+		testQueries.SetFunc("nonconcurrent", func(old []string) []string {
+			return append(old, sql)
+		})
+	}
+
+	ctx, cancelCtx := context.WithTimeout(context.Background(), 1*time.Second)
+	defer cancelCtx()
+
+	sem := semaphore.NewWeighted(1)
+	sem.Acquire(ctx, 1)
+
+	// currently there is no hook for the collections cache reload so we pool instead
+	done := make(chan bool, 1)
+	ticker := time.NewTicker(100 * time.Millisecond)
+	go func() {
+		for {
+			select {
+			case <-ticker.C:
+				if len(testQueries.Get("concurrent")) == 1 {
+					sem.Release(1)
+					return
+				}
+			case <-done:
+				return
+			}
+		}
+	}()
+
+	// create/update/delete app1 collections should trigger a reload in app2
+	dummyCollection := core.NewBaseCollection("test")
+	if err := app1.Save(dummyCollection); err != nil {
+		t.Fatal(err)
+	}
+	dummyCollection.Fields.Add(&core.TextField{Name: "test"})
+	if err := app1.Save(dummyCollection); err != nil {
+		t.Fatal(err)
+	}
+	if err := app1.Delete(dummyCollection); err != nil {
+	}
+
+	// block until released or timeouted
+	sem.Acquire(ctx, 1)
+	ticker.Stop()
+	done <- true
+
+	nonconcurrentQueries := testQueries.Get("nonconcurrent")
+	concurrentQueries := testQueries.Get("concurrent")
+
+	if len(nonconcurrentQueries) != 0 {
+		t.Fatalf("Expected 0 concurrent queries, got %d (%v)", len(nonconcurrentQueries), nonconcurrentQueries)
+	}
+	if len(concurrentQueries) != 1 {
+		t.Fatalf("Expected 1 concurrent query, got %d (%v)", len(concurrentQueries), concurrentQueries)
+	}
+
+	expectedQuery := "SELECT {{_collections}}.* FROM `_collections` ORDER BY `rowid` ASC"
+	if concurrentQueries[0] != expectedQuery {
+		t.Fatalf("Expected query\n%s\ngot\n%s", expectedQuery, concurrentQueries[0])
+	}
+}

core/settings_model.go

@@ -120,6 +120,10 @@ var (
 )
 
 type settings struct {
+	// SuperuserIPs defines an optional list of the superuser allowed
+	// individual IPs and subnets (in CIDR notation).
+	SuperuserIPs []string `form:"superuserIPs" json:"superuserIPs"`
+
 	SMTP         SMTPConfig         `form:"smtp" json:"smtp"`
 	Backups      BackupsConfig      `form:"backups" json:"backups"`
 	S3           S3Config           `form:"s3" json:"s3"`
@@ -253,6 +257,12 @@ func (s *Settings) DBExport(app App) (map[string]any, error) {
 	}
 	result["updated"] = now
 
+	// @todo remove with encoding/json/2
+	// serialize as empty array
+	if s.settings.SuperuserIPs == nil {
+		s.settings.SuperuserIPs = []string{}
+	}
+
 	encoded, err := json.Marshal(s.settings)
 	if err != nil {
 		return nil, err
@@ -280,6 +290,7 @@ func (s *Settings) PostValidate(ctx context.Context, app App) error {
 	defer s.mu.RUnlock()
 
 	return validation.ValidateStructWithContext(ctx, s,
+		validation.Field(&s.SuperuserIPs, validation.Each(validation.By(validators.IPOrSubnet))),
 		validation.Field(&s.Meta),
 		validation.Field(&s.Logs),
 		validation.Field(&s.SMTP),
@@ -343,6 +354,12 @@ func (s *Settings) MarshalJSON() ([]byte, error) {
 		}
 	}
 
+	// @todo remove with encoding/json/2
+	// serialize as empty array
+	if copy.SuperuserIPs == nil {
+		copy.SuperuserIPs = []string{}
+	}
+
 	return json.Marshal(copy)
 }
 

core/settings_model_test.go

@@ -84,7 +84,7 @@ func TestSettings_DBExport(t *testing.T) {
 				valueStr = string(export["value"].([]byte))
 			}
 
-			expected := `{"smtp":{"enabled":false,"port":0,"host":"smtp_host","username":"smtp_username","password":"","authMethod":"","tls":false,"localName":""},"backups":{"cron":"* * * * *","cronMaxKeep":0,"s3":{"enabled":true,"bucket":"","region":"","endpoint":"","accessKey":"","forcePathStyle":false}},"s3":{"enabled":false,"bucket":"","region":"","endpoint":"s3_endpoint","accessKey":"","secret":"s3_secret","forcePathStyle":false},"meta":{"accentColor":"","appName":"test_app_name","appURL":"","senderName":"","senderAddress":"","hideControls":false},"rateLimits":{"rules":[],"enabled":true},"trustedProxy":{"headers":[],"useLeftmostIP":true},"batch":{"enabled":false,"maxRequests":0,"timeout":15,"maxBodySize":0},"logs":{"maxDays":123,"minLevel":0,"logIP":false,"logAuthId":false}}`
+			expected := `{"superuserIPs":[],"smtp":{"enabled":false,"port":0,"host":"smtp_host","username":"smtp_username","password":"","authMethod":"","tls":false,"localName":""},"backups":{"cron":"* * * * *","cronMaxKeep":0,"s3":{"enabled":true,"bucket":"","region":"","endpoint":"","accessKey":"","forcePathStyle":false}},"s3":{"enabled":false,"bucket":"","region":"","endpoint":"s3_endpoint","accessKey":"","secret":"s3_secret","forcePathStyle":false},"meta":{"accentColor":"","appName":"test_app_name","appURL":"","senderName":"","senderAddress":"","hideControls":false},"rateLimits":{"rules":[],"enabled":true},"trustedProxy":{"headers":[],"useLeftmostIP":true},"batch":{"enabled":false,"maxRequests":0,"timeout":15,"maxBodySize":0},"logs":{"maxDays":123,"minLevel":0,"logIP":false,"logAuthId":false}}`
 			if valueStr != expected {
 				t.Fatalf("Expected exported settings\n%s\ngot\n%s", expected, valueStr)
 			}
@@ -180,7 +180,7 @@ func TestSettingsMarshalJSON(t *testing.T) {
 	}
 	rawStr := string(raw)
 
-	expected := `{"smtp":{"enabled":false,"port":0,"host":"","username":"abc","authMethod":"","tls":false,"localName":""},"backups":{"cron":"","cronMaxKeep":0,"s3":{"enabled":false,"bucket":"","region":"","endpoint":"","accessKey":"","forcePathStyle":false}},"s3":{"enabled":false,"bucket":"","region":"","endpoint":"","accessKey":"","forcePathStyle":false},"meta":{"accentColor":"","appName":"test123","appURL":"","senderName":"","senderAddress":"","hideControls":false},"rateLimits":{"rules":[],"enabled":false},"trustedProxy":{"headers":[],"useLeftmostIP":false},"batch":{"enabled":false,"maxRequests":0,"timeout":0,"maxBodySize":0},"logs":{"maxDays":0,"minLevel":0,"logIP":false,"logAuthId":false}}`
+	expected := `{"superuserIPs":[],"smtp":{"enabled":false,"port":0,"host":"","username":"abc","authMethod":"","tls":false,"localName":""},"backups":{"cron":"","cronMaxKeep":0,"s3":{"enabled":false,"bucket":"","region":"","endpoint":"","accessKey":"","forcePathStyle":false}},"s3":{"enabled":false,"bucket":"","region":"","endpoint":"","accessKey":"","forcePathStyle":false},"meta":{"accentColor":"","appName":"test123","appURL":"","senderName":"","senderAddress":"","hideControls":false},"rateLimits":{"rules":[],"enabled":false},"trustedProxy":{"headers":[],"useLeftmostIP":false},"batch":{"enabled":false,"maxRequests":0,"timeout":0,"maxBodySize":0},"logs":{"maxDays":0,"minLevel":0,"logIP":false,"logAuthId":false}}`
 
 	if rawStr != expected {
 		t.Fatalf("Expected\n%v\ngot\n%v", expected, rawStr)
@@ -196,6 +196,7 @@ func TestSettingsValidate(t *testing.T) {
 	s := app.Settings()
 
 	// set invalid settings data
+	s.SuperuserIPs = []string{"127.0.0.1", "invalid"}
 	s.Meta.AppName = ""
 	s.Logs.MaxDays = -10
 	s.SMTP.Enabled = true
@@ -217,6 +218,7 @@ func TestSettingsValidate(t *testing.T) {
 	}
 
 	expectations := []string{
+		`"superuserIPs":{`,
 		`"meta":{`,
 		`"logs":{`,
 		`"smtp":{`,

core/validators/file.go

@@ -31,7 +31,7 @@ func UploadedFileSize(maxBytes int64) validation.RuleFunc {
 				"validation_file_size_limit",
 				"Failed to upload {{.file}} - the maximum allowed file size is {{.maxSize}} bytes.",
 			).SetParams(map[string]any{
-				"file":    v.OriginalName,
+				"file":    cutStr(v.OriginalName, 300),
 				"maxSize": maxBytes,
 			})
 		}
@@ -60,7 +60,7 @@ func UploadedFileMimeType(validTypes []string) validation.RuleFunc {
 
 		baseErr := validation.NewError(
 			"validation_invalid_mime_type",
-			fmt.Sprintf("Failed to upload %q due to unsupported file type.", v.OriginalName),
+			fmt.Sprintf("Failed to upload %q due to unsupported file type.", cutStr(v.OriginalName, 300)),
 		)
 
 		if len(validTypes) == 0 {

core/validators/string.go

@@ -1,6 +1,7 @@
 package validators
 
 import (
+	"net/netip"
 	"regexp"
 
 	validation "github.com/go-ozzo/ozzo-validation/v4"
@@ -27,3 +28,30 @@ func IsRegex(value any) error {
 
 	return nil
 }
+
+// IPOrSubnet checks whether the validated value is an individual
+// IPv4/IPv6 or CIDR subnet.
+func IPOrSubnet(value any) error {
+	v, ok := value.(string)
+	if !ok {
+		return ErrUnsupportedValueType
+	}
+
+	if v == "" {
+		return nil // nothing to check
+	}
+
+	// subnet
+	_, err := netip.ParsePrefix(v)
+	if err == nil {
+		return nil
+	}
+
+	// individual IP
+	_, err = netip.ParseAddr(v)
+	if err == nil {
+		return nil
+	}
+
+	return validation.NewError("validation_invlaid_ip_or_subnet", "invalid IP or CIDR subnet")
+}

core/validators/string_test.go

@@ -31,3 +31,32 @@ func TestIsRegex(t *testing.T) {
 		})
 	}
 }
+
+func TestIPOrSubnet(t *testing.T) {
+	t.Parallel()
+
+	scenarios := []struct {
+		val         string
+		expectError bool
+	}{
+		{"", false},
+		{`invalid`, true},
+		{`127.0`, true}, // incomplete
+		{`127.0.0.1`, false},
+		{`::1`, false},
+		{`0000:0000:0000:0000:0000:0000:0000:0001`, false},
+		{`127.0.0.1/24`, false},
+		{`::/128`, false},
+	}
+
+	for i, s := range scenarios {
+		t.Run(fmt.Sprintf("%d_%#v", i, s.val), func(t *testing.T) {
+			err := validators.IPOrSubnet(s.val)
+
+			hasErr := err != nil
+			if hasErr != s.expectError {
+				t.Fatalf("Expected hasErr to be %v, got %v (%v)", s.expectError, hasErr, err)
+			}
+		})
+	}
+}

core/validators/validators.go

@@ -38,3 +38,10 @@ func JoinValidationErrors(errA, errB error) error {
 
 	return errors.Join(errA, errB)
 }
+
+func cutStr(str string, max int) string {
+	if len(str) > max {
+		return str[:max] + "..."
+	}
+	return str
+}