diff --git a/Client/tauri-client/src/lib/pluginBridge.ts b/Client/tauri-client/src/lib/pluginBridge.ts index 16b17e02..dc4a4ce8 100644 --- a/Client/tauri-client/src/lib/pluginBridge.ts +++ b/Client/tauri-client/src/lib/pluginBridge.ts @@ -86,14 +86,37 @@ class PluginBridge { ); } + /** + * Look up the pluginId of an iframe by its contentWindow. Returns null if + * the source is not one of our managed plugin frames. This is the key + * defense against postMessage spoofing: we never trust the pluginId field + * inside the message body, only the e.source pointer. + */ + private pluginIdForSource(source: MessageEventSource | null): number | null { + if (!source) return null; + for (const [pid, frame] of this.frames) { + if (frame.contentWindow === source) return pid; + } + return null; + } + private onMessage = (e: MessageEvent): void => { const data = e.data; if (!data || typeof data !== "object") return; if ((data as { source?: unknown }).source === HOST_ORIGIN_PREFIX) return; // own echo - const env = data as { pluginId?: unknown; type?: unknown; payload?: unknown }; - if (typeof env.pluginId !== "number" || typeof env.type !== "string") return; + // SECURITY: validate the message originated from one of our managed + // plugin iframes by matching e.source against frame.contentWindow. + // Without this check, any arbitrary frame (including a malicious parent + // frame in an embedding scenario, or any same-origin script that + // obtained a window reference) could spoof messages from any plugin by + // claiming an arbitrary pluginId in the body. The pluginId from the + // message body is intentionally ignored — we use the trusted lookup. + const trustedPluginId = this.pluginIdForSource(e.source); + if (trustedPluginId === null) return; + const env = data as { type?: unknown; payload?: unknown }; + if (typeof env.type !== "string") return; const envelope: PluginMessageEnvelope = { - pluginId: env.pluginId, + pluginId: trustedPluginId, type: env.type, payload: env.payload, }; diff --git a/Server/admin/middleware.go b/Server/admin/middleware.go index bb06fdc4..1f8c4b9b 100644 --- a/Server/admin/middleware.go +++ b/Server/admin/middleware.go @@ -11,6 +11,13 @@ import ( // ─── Middleware ─────────────────────────────────────────────────────────────── +// RequireAdminAuth is the exported form of adminAuthMiddleware. External +// packages (e.g. api/router.go for the plugin admin handler) reuse it so the +// session/permission gate stays in one place. +func RequireAdminAuth(database *db.DB) func(http.Handler) http.Handler { + return adminAuthMiddleware(database) +} + // adminAuthMiddleware validates the Bearer token and requires ADMINISTRATOR. // On success it stores the *db.User and *db.Session in the request context so // downstream handlers can retrieve them without re-querying the database. diff --git a/Server/api/diagnostics_handler_test.go b/Server/api/diagnostics_handler_test.go index 564fe69f..9cb8b6a8 100644 --- a/Server/api/diagnostics_handler_test.go +++ b/Server/api/diagnostics_handler_test.go @@ -33,7 +33,7 @@ func setupDiagnosticsRouter(t *testing.T) (http.Handler, string) { }, } - handler, _, cleanup := api.NewRouter(cfg, database, "1.0.0-test", nil) + handler, _, cleanup := api.NewRouter(cfg, database, "1.0.0-test", nil, nil) t.Cleanup(cleanup) // Create a user and session for authenticated requests. diff --git a/Server/api/router.go b/Server/api/router.go index a80bd59e..404375be 100644 --- a/Server/api/router.go +++ b/Server/api/router.go @@ -16,6 +16,7 @@ import ( "github.com/owncord/server/config" "github.com/owncord/server/db" "github.com/owncord/server/permissions" + "github.com/owncord/server/plugin" "github.com/owncord/server/service" "github.com/owncord/server/storage" dbstore "github.com/owncord/server/store" @@ -27,7 +28,10 @@ import ( // NewRouter builds and returns the fully configured HTTP handler, the // WebSocket hub (so the caller can call hub.GracefulStop on shutdown), and a // cleanup function that stops background goroutines (e.g. rate-limiter cleanup). -func NewRouter(cfg *config.Config, database *db.DB, ver string, logBuf *admin.RingBuffer) (http.Handler, *ws.Hub, func()) { +// +// pluginRegistry may be nil — in that case the plugin admin endpoints respond +// with 503 on lifecycle calls and an empty list on read. +func NewRouter(cfg *config.Config, database *db.DB, ver string, logBuf *admin.RingBuffer, pluginRegistry *plugin.Registry) (http.Handler, *ws.Hub, func()) { r := chi.NewRouter() // Middleware stack. @@ -225,11 +229,17 @@ func NewRouter(cfg *config.Config, database *db.DB, ver string, logBuf *admin.Ri r.Use(AdminIPRestrict(cfg.Server.AdminAllowedCIDRs, cfg.Server.TrustedProxies)) r.Mount("/admin", adminHandler) - // Phase C Step 9 — plugin admin REST surface. Mounted alongside the - // admin panel so it inherits the same network ACL. Plugin runtime is - // owned by main.go; the handler accepts a nil registry and reports - // 503 on lifecycle calls when plugin support is disabled. - r.Mount("/api/v1/admin/plugins", NewPluginAdminHandler(nil, st)) + // Phase C Step 9 — plugin admin REST surface. The IP gate above is + // only the outer perimeter; plugin lifecycle endpoints additionally + // require a valid admin Bearer token via admin.RequireAdminAuth so a + // LAN attacker on the allowed CIDR cannot install/enable plugins + // without a session. The handler is wired with the live registry + // constructed in main.go (nil when plugin support is disabled, in + // which case lifecycle calls return 503 and list returns []). + r.Group(func(r chi.Router) { + r.Use(admin.RequireAdminAuth(database)) + r.Mount("/api/v1/admin/plugins", NewPluginAdminHandler(pluginRegistry, st)) + }) }) // Client auto-update endpoint (unauthenticated). diff --git a/Server/api/router_test.go b/Server/api/router_test.go index 40aebaf0..323c53c1 100644 --- a/Server/api/router_test.go +++ b/Server/api/router_test.go @@ -32,7 +32,7 @@ func setupRouter(t *testing.T) http.Handler { }, } - handler, _, cleanup := api.NewRouter(cfg, database, "test", nil) + handler, _, cleanup := api.NewRouter(cfg, database, "test", nil, nil) t.Cleanup(cleanup) return handler } diff --git a/Server/db/queries/postgres/events.sql b/Server/db/queries/postgres/events.sql index 2c61d7fa..ec332592 100644 --- a/Server/db/queries/postgres/events.sql +++ b/Server/db/queries/postgres/events.sql @@ -1,7 +1,12 @@ --- name: PersistEvent :one -INSERT INTO events (event_type, channel_id, payload) -VALUES ($1, $2, $3) -RETURNING seq; +-- name: PersistEvent :exec +-- seq is supplied by the hub so the row seq matches the wrapped-payload seq. +-- The schema's BIGSERIAL still owns the id column for inserts that omit seq, +-- but PersistEvent always supplies an explicit value. +INSERT INTO events (seq, event_type, channel_id, payload) +VALUES ($1, $2, $3, $4); + +-- name: GetMaxEventSeq :one +SELECT COALESCE(MAX(seq), 0)::BIGINT FROM events; -- name: GetEventsSince :many SELECT seq, event_type, channel_id, payload, created_at diff --git a/Server/db/queries/sqlite/events.sql b/Server/db/queries/sqlite/events.sql index e6a18941..ec23190a 100644 --- a/Server/db/queries/sqlite/events.sql +++ b/Server/db/queries/sqlite/events.sql @@ -1,5 +1,9 @@ --- name: PersistEvent :execresult -INSERT INTO events (event_type, channel_id, payload) VALUES (?, ?, ?); +-- name: PersistEvent :exec +-- seq is supplied by the hub so the row seq matches the wrapped-payload seq. +INSERT INTO events (seq, event_type, channel_id, payload) VALUES (?, ?, ?, ?); + +-- name: GetMaxEventSeq :one +SELECT COALESCE(MAX(seq), 0) FROM events; -- name: GetEventsSince :many SELECT seq, event_type, channel_id, payload, created_at diff --git a/Server/main.go b/Server/main.go index a87c3e6b..e578cb1a 100644 --- a/Server/main.go +++ b/Server/main.go @@ -149,17 +149,56 @@ func run(log *slog.Logger, logBuf *admin.RingBuffer) error { } }() - // ── 5. Build HTTP router ─────────────────────────────────────────────── - router, hub, routerCleanup := api.NewRouter(cfg, database, version, logBuf) + // ── 5. Construct shared store wrapper ────────────────────────────────── + // Used by both the event persistence layer and the plugin runtime. Once + // the Phase A "store everywhere" refactor lands, NewRouter will accept + // store.Store directly and this wrapper goes away. + storeWrapper := store.NewSQLiteStore(database) + + // ── 5a. Construct plugin runtime BEFORE the router so the router can + // wire the live registry into the plugin admin handler. ──────────────── + var pluginRegistry *plugin.Registry + if cfg.Plugins.Enabled { + registry, plugErr := plugin.NewRegistry(plugin.Config{ + Directory: cfg.Plugins.Directory, + MaxMemoryMB: cfg.Plugins.MaxMemoryMB, + CPUBudgetMs: cfg.Plugins.CPUBudgetMs, + HTTPAllowlist: cfg.Plugins.HTTPAllowlist, + Store: storeWrapper, + }) + if plugErr != nil { + log.Warn("plugin runtime init failed; continuing without plugins", "error", plugErr) + } else { + pluginRegistry = registry + if err := registry.LoadAll(context.Background()); err != nil { + log.Warn("plugin loader: failed to scan directory", "error", err) + } + defer func() { + closeCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = registry.Close(closeCtx) + }() + } + } + + // ── 5b. Build HTTP router ────────────────────────────────────────────── + router, hub, routerCleanup := api.NewRouter(cfg, database, version, logBuf, pluginRegistry) defer routerCleanup() - // ── 5b. Wire event persistence (Phase B Step 7) ──────────────────────── - // Construct a Store wrapper for the cold-tier event log + plugin KV. The - // store-everywhere refactor (Phase A pending TODO) will eventually thread - // this through NewRouter directly; for now we attach it after the fact so - // the router signature stays unchanged. - storeWrapper := store.NewSQLiteStore(database) + // ── 5c. Wire event persistence (Phase B Step 7) ──────────────────────── if cfg.EventPersistence.Enabled && hub != nil { + // Seed the hub's in-memory seq counter from the persisted MAX(seq) + // so wrapped-payload seqs stay monotonic across restarts. Without + // this, the events table accumulates rows whose payload seqs reset + // to 1 after every restart, breaking the reconnect "events since + // last_seq" contract. + if maxSeq, seedErr := storeWrapper.GetMaxEventSeq(context.Background()); seedErr != nil { + log.Warn("event persistence: failed to read MAX(events.seq); starting hub seq from 0", "error", seedErr) + } else if maxSeq > 0 { + hub.SeedSeq(uint64(maxSeq)) + log.Info("event persistence: seeded hub seq from persisted events", "seq", maxSeq) + } + persister := ws.NewEventPersister( storeWrapper, 4096, @@ -182,29 +221,6 @@ func run(log *slog.Logger, logBuf *admin.RingBuffer) error { }() } - // ── 5c. Wire plugin runtime (Phase C Step 9) ─────────────────────────── - if cfg.Plugins.Enabled { - registry, plugErr := plugin.NewRegistry(plugin.Config{ - Directory: cfg.Plugins.Directory, - MaxMemoryMB: cfg.Plugins.MaxMemoryMB, - CPUBudgetMs: cfg.Plugins.CPUBudgetMs, - HTTPAllowlist: cfg.Plugins.HTTPAllowlist, - Store: storeWrapper, - }) - if plugErr != nil { - log.Warn("plugin runtime init failed; continuing without plugins", "error", plugErr) - } else { - if err := registry.LoadAll(context.Background()); err != nil { - log.Warn("plugin loader: failed to scan directory", "error", err) - } - defer func() { - closeCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - _ = registry.Close(closeCtx) - }() - } - } - // ── 6. Start server ──────────────────────────────────────────────────── addr := fmt.Sprintf(":%d", cfg.Server.Port) srv := &http.Server{ diff --git a/Server/plugin/host_http.go b/Server/plugin/host_http.go index 39fbd0ee..53dee803 100644 --- a/Server/plugin/host_http.go +++ b/Server/plugin/host_http.go @@ -7,10 +7,14 @@ package plugin import ( + "bytes" "context" + "errors" "fmt" "io" + "net" "net/http" + "net/url" "strings" "time" ) @@ -30,34 +34,100 @@ type HTTPResponse struct { Header map[string]string } -const httpTimeout = 10 * time.Second +const ( + httpTimeout = 10 * time.Second + maxResponseBytes = 5 * 1024 * 1024 // 5 MiB +) + +// ErrHTTPHostDenied is returned when a plugin HTTP request targets a host that +// is not in the allowlist or resolves to a private/loopback/link-local address. +var ErrHTTPHostDenied = errors.New("plugin http: host denied") // HTTPDo executes a plugin-initiated HTTP request after enforcing the host -// allowlist declared in PluginsConfig. +// allowlist declared in PluginsConfig and rejecting requests that resolve to +// private, loopback, or link-local IP ranges (SSRF defense). func (r *Registry) HTTPDo(ctx context.Context, inst *Instance, req HTTPRequest) (*HTTPResponse, error) { if !inst.Manifest.HasCapability(CapHTTP) { return nil, ErrCapabilityNotGranted } - if !r.hostAllowed(req.URL) { - return nil, fmt.Errorf("plugin http: host not in allowlist: %s", req.URL) + parsed, err := url.Parse(req.URL) + if err != nil { + return nil, fmt.Errorf("plugin http: invalid URL: %w", err) } - httpReq, err := http.NewRequestWithContext(ctx, req.Method, req.URL, strings.NewReader(string(req.Body))) + if parsed.Scheme != "http" && parsed.Scheme != "https" { + return nil, fmt.Errorf("plugin http: scheme %q not allowed", parsed.Scheme) + } + host := parsed.Hostname() + if host == "" { + return nil, fmt.Errorf("plugin http: empty host") + } + if !r.hostAllowed(host) { + return nil, fmt.Errorf("%w: %s", ErrHTTPHostDenied, host) + } + if err := rejectPrivateAddrs(ctx, host); err != nil { + return nil, fmt.Errorf("%w: %v", ErrHTTPHostDenied, err) + } + + httpReq, err := http.NewRequestWithContext(ctx, req.Method, req.URL, bytes.NewReader(req.Body)) if err != nil { return nil, fmt.Errorf("plugin http: build request: %w", err) } for k, v := range req.Header { httpReq.Header.Set(k, v) } - client := &http.Client{Timeout: httpTimeout} + // Custom transport with a guarded DialContext: every actual TCP dial + // re-checks the resolved IP, closing the DNS-rebinding TOCTOU window + // between rejectPrivateAddrs above and the underlying dial. + dialer := &net.Dialer{Timeout: httpTimeout} + transport := &http.Transport{ + DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { + h, _, splitErr := net.SplitHostPort(addr) + if splitErr != nil { + return nil, splitErr + } + ip := net.ParseIP(h) + if ip == nil { + // Hostname — resolve and validate every address before dial. + if err := rejectPrivateAddrs(ctx, h); err != nil { + return nil, fmt.Errorf("%w: %v", ErrHTTPHostDenied, err) + } + } else if err := ipAllowed(ip); err != nil { + return nil, fmt.Errorf("%w: %v", ErrHTTPHostDenied, err) + } + return dialer.DialContext(ctx, network, addr) + }, + } + client := &http.Client{ + Timeout: httpTimeout, + Transport: transport, + // Refuse to follow redirects across hosts that the allowlist would + // reject — re-evaluate the new URL through the same checks. + CheckRedirect: func(redirReq *http.Request, via []*http.Request) error { + if len(via) >= 5 { + return fmt.Errorf("too many redirects") + } + h := redirReq.URL.Hostname() + if !r.hostAllowed(h) { + return fmt.Errorf("%w: redirect to %s", ErrHTTPHostDenied, h) + } + return nil + }, + } resp, err := client.Do(httpReq) if err != nil { return nil, fmt.Errorf("plugin http: do: %w", err) } defer resp.Body.Close() - body, err := io.ReadAll(resp.Body) + // Cap body size so a hostile/large response cannot OOM the host. We + // LimitReader to maxResponseBytes+1 so we can detect truncation. + limited := io.LimitReader(resp.Body, maxResponseBytes+1) + body, err := io.ReadAll(limited) if err != nil { return nil, fmt.Errorf("plugin http: read body: %w", err) } + if int64(len(body)) > maxResponseBytes { + return nil, fmt.Errorf("plugin http: response exceeds %d bytes", maxResponseBytes) + } hdr := make(map[string]string, len(resp.Header)) for k, v := range resp.Header { if len(v) > 0 { @@ -71,24 +141,82 @@ func (r *Registry) HTTPDo(ctx context.Context, inst *Instance, req HTTPRequest) }, nil } -// hostAllowed reports whether url's host matches any suffix in the allowlist. -func (r *Registry) hostAllowed(url string) bool { - // Trivial host extraction — full URL parsing would be overkill since the - // allowlist match is suffix-based. - rest := url - for _, prefix := range []string{"https://", "http://"} { - if strings.HasPrefix(rest, prefix) { - rest = rest[len(prefix):] - break +// hostAllowed reports whether host matches any allowlist entry. Matching is +// either exact (host == entry) or proper suffix bounded by a dot +// (host == "api."+entry or host ends with "."+entry). This rejects +// "evilexample.com" against an allowlist of "example.com". +// +// Empty allowlist entries are ignored to prevent the empty-suffix wildcard +// bug. host is expected to already be a clean hostname (no scheme/port/path). +func (r *Registry) hostAllowed(host string) bool { + if host == "" { + return false + } + host = strings.ToLower(strings.TrimSuffix(host, ".")) + for _, entry := range r.cfg.HTTPAllowlist { + entry = strings.ToLower(strings.TrimSpace(entry)) + if entry == "" { + continue } - } - if i := strings.IndexAny(rest, "/?#"); i >= 0 { - rest = rest[:i] - } - for _, suffix := range r.cfg.HTTPAllowlist { - if strings.HasSuffix(rest, suffix) { + if host == entry { + return true + } + if strings.HasSuffix(host, "."+entry) { return true } } return false } + +// rejectPrivateAddrs resolves host and returns an error if any resolved +// address is loopback, link-local, private (RFC1918), or unspecified. +// This prevents an allowlisted hostname from being repointed at internal +// services via DNS. +func rejectPrivateAddrs(ctx context.Context, host string) error { + // If host is already an IP literal, check it directly. + if ip := net.ParseIP(host); ip != nil { + return ipAllowed(ip) + } + resolver := &net.Resolver{} + ips, err := resolver.LookupIPAddr(ctx, host) + if err != nil { + return fmt.Errorf("dns lookup failed: %w", err) + } + if len(ips) == 0 { + return fmt.Errorf("no addresses for %s", host) + } + for _, addr := range ips { + if err := ipAllowed(addr.IP); err != nil { + return err + } + } + return nil +} + +// ipAllowed reports nil if ip is a public, routable address. Loopback, +// link-local, multicast, unspecified, and RFC1918 ranges are rejected. +func ipAllowed(ip net.IP) error { + if ip == nil { + return fmt.Errorf("nil ip") + } + if ip.IsLoopback() { + return fmt.Errorf("loopback address %s", ip) + } + if ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() { + return fmt.Errorf("link-local address %s", ip) + } + if ip.IsPrivate() { + return fmt.Errorf("private address %s", ip) + } + if ip.IsUnspecified() { + return fmt.Errorf("unspecified address %s", ip) + } + if ip.IsMulticast() { + return fmt.Errorf("multicast address %s", ip) + } + // Reject IPv4-mapped IPv6 forms of the same. + if v4 := ip.To4(); v4 != nil && (v4.IsLoopback() || v4.IsPrivate() || v4.IsLinkLocalUnicast()) { + return fmt.Errorf("disallowed v4-mapped address %s", ip) + } + return nil +} diff --git a/Server/plugin/host_ui.go b/Server/plugin/host_ui.go index 73929853..8536d837 100644 --- a/Server/plugin/host_ui.go +++ b/Server/plugin/host_ui.go @@ -38,22 +38,35 @@ func (r *Registry) RegisterUI(inst *Instance) error { } // AssetHandler returns an http.Handler that serves the on-disk assets for -// inst, rooted at the plugin's directory. The handler refuses path traversal -// attempts and only serves files declared by manifest tabs. +// inst, rooted at the plugin's directory. Defense in depth: +// 1. Manifest validation rejects absolute paths and "..". +// 2. The handler only serves files explicitly declared by a manifest tab. +// 3. After resolving the on-disk path we use filepath.Rel and reject any +// result containing ".." or that is absolute, which catches symlink +// escapes and the prefix-without-separator class of bug. func (r *Registry) AssetHandler(inst *Instance) http.Handler { allowed := make(map[string]bool, len(inst.Manifest.UI.Tabs)) for _, t := range inst.Manifest.UI.Tabs { allowed[t.Asset] = true } - pluginDir := filepath.Dir(inst.WASMPath) + pluginDir, dirErr := filepath.Abs(filepath.Dir(inst.WASMPath)) return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + if dirErr != nil { + http.Error(w, "plugin asset root unavailable", http.StatusInternalServerError) + return + } rel := strings.TrimPrefix(req.URL.Path, "/") if !allowed[rel] { http.NotFound(w, req) return } - full := filepath.Join(pluginDir, rel) - if !strings.HasPrefix(full, pluginDir) { + full, absErr := filepath.Abs(filepath.Join(pluginDir, rel)) + if absErr != nil { + http.Error(w, "forbidden", http.StatusForbidden) + return + } + relCheck, relErr := filepath.Rel(pluginDir, full) + if relErr != nil || relCheck == "" || relCheck == "." || strings.HasPrefix(relCheck, "..") || filepath.IsAbs(relCheck) { http.Error(w, "forbidden", http.StatusForbidden) return } diff --git a/Server/plugin/manifest.go b/Server/plugin/manifest.go index 2f6be965..8c4bb82a 100644 --- a/Server/plugin/manifest.go +++ b/Server/plugin/manifest.go @@ -24,19 +24,28 @@ package plugin import ( "encoding/json" "fmt" + "path" + "regexp" "strings" ) +// pluginNameRegexp restricts plugin names to a tight ASCII charset so the +// name can flow safely into URL paths (/api/v1/plugins//...), filesystem +// paths, and log lines without escaping concerns. Mirrors the npm package +// name rules: lowercase, digits, dash and underscore, 1-64 chars, must start +// with a letter or digit. +var pluginNameRegexp = regexp.MustCompile(`^[a-z0-9][a-z0-9_-]{0,63}$`) + // Manifest is the parsed plugin metadata declared in plugin.json (or // plugin.toml in the wazero-tagged build). The on-disk schema is intentionally // flat so the default JSON parser handles it without a TOML dependency. type Manifest struct { - Name string `json:"name"` - Version string `json:"version"` - Author string `json:"author"` - Description string `json:"description"` - Entrypoint string `json:"entrypoint"` // relative .wasm path - Permissions []string `json:"permissions"` + Name string `json:"name"` + Version string `json:"version"` + Author string `json:"author"` + Description string `json:"description"` + Entrypoint string `json:"entrypoint"` // relative .wasm path + Permissions []string `json:"permissions"` Resources Resources `json:"resources"` UI UISpec `json:"ui"` } @@ -98,15 +107,24 @@ func (m *Manifest) Validate() error { if strings.TrimSpace(m.Name) == "" { return fmt.Errorf("plugin manifest: name is required") } + if !pluginNameRegexp.MatchString(m.Name) { + return fmt.Errorf("plugin manifest: name %q must match %s", m.Name, pluginNameRegexp.String()) + } if strings.TrimSpace(m.Version) == "" { return fmt.Errorf("plugin manifest: version is required") } + if len(m.Version) > 64 { + return fmt.Errorf("plugin manifest: version too long (max 64)") + } if strings.TrimSpace(m.Entrypoint) == "" { return fmt.Errorf("plugin manifest: entrypoint is required") } if !strings.HasSuffix(m.Entrypoint, ".wasm") { return fmt.Errorf("plugin manifest: entrypoint %q must end in .wasm", m.Entrypoint) } + if err := validateRelativePath(m.Entrypoint); err != nil { + return fmt.Errorf("plugin manifest: entrypoint: %w", err) + } for _, p := range m.Permissions { if !validCapabilities[Capability(p)] { return fmt.Errorf("plugin manifest: unknown permission %q", p) @@ -115,6 +133,47 @@ func (m *Manifest) Validate() error { if m.Resources.MaxMemoryMB < 0 || m.Resources.CPUBudgetMs < 0 { return fmt.Errorf("plugin manifest: resources must be non-negative") } + for i, t := range m.UI.Tabs { + if strings.TrimSpace(t.ID) == "" { + return fmt.Errorf("plugin manifest: ui.tabs[%d].id is required", i) + } + if strings.TrimSpace(t.Asset) == "" { + return fmt.Errorf("plugin manifest: ui.tabs[%d].asset is required", i) + } + if err := validateRelativePath(t.Asset); err != nil { + return fmt.Errorf("plugin manifest: ui.tabs[%d].asset: %w", i, err) + } + } + return nil +} + +// validateRelativePath rejects absolute paths, paths containing "..", paths +// with NUL bytes, backslashes (Windows separators), and paths that the path +// package's Clean would alter (which catches "./foo", "foo//bar", trailing +// slashes, etc.). Asset and entrypoint paths must be plain forward-slash +// relative segments under the plugin directory. +func validateRelativePath(p string) error { + if p == "" { + return fmt.Errorf("path is empty") + } + if strings.ContainsRune(p, 0) { + return fmt.Errorf("path contains NUL byte") + } + if strings.ContainsRune(p, '\\') { + return fmt.Errorf("path contains backslash; use forward slashes only") + } + if strings.HasPrefix(p, "/") { + return fmt.Errorf("path %q must be relative", p) + } + cleaned := path.Clean(p) + if cleaned != p { + return fmt.Errorf("path %q is not in canonical form (clean: %q)", p, cleaned) + } + for _, seg := range strings.Split(cleaned, "/") { + if seg == ".." { + return fmt.Errorf("path %q contains parent traversal", p) + } + } return nil } diff --git a/Server/store/memstore_events.go b/Server/store/memstore_events.go index 3213cd5d..8a9905b0 100644 --- a/Server/store/memstore_events.go +++ b/Server/store/memstore_events.go @@ -15,7 +15,9 @@ import ( // memEventStore is an in-memory EventStore + PluginStore implementation // embedded into MemStore via the field below. type memEventStore struct { - mu sync.Mutex + mu sync.Mutex + // nextSeq tracks the high-water mark of caller-supplied event seqs so + // GetMaxEventSeq is O(1). PersistEvent updates it after each insert. nextSeq atomic.Int64 events []db.PersistedEvent plugins map[int64]*db.PluginRow @@ -39,11 +41,10 @@ func (m *MemStore) ensureEvents() *memEventStore { return m.eventStore } -func (m *MemStore) PersistEvent(_ context.Context, eventType string, channelID int64, payload []byte) (int64, error) { +func (m *MemStore) PersistEvent(_ context.Context, seq int64, eventType string, channelID int64, payload []byte) error { es := m.ensureEvents() es.mu.Lock() defer es.mu.Unlock() - seq := es.nextSeq.Add(1) cp := make([]byte, len(payload)) copy(cp, payload) es.events = append(es.events, db.PersistedEvent{ @@ -53,7 +54,11 @@ func (m *MemStore) PersistEvent(_ context.Context, eventType string, channelID i Payload: cp, CreatedAt: time.Now().UTC(), }) - return seq, nil + // Track the high water mark so GetMaxEventSeq is O(1). + if seq > es.nextSeq.Load() { + es.nextSeq.Store(seq) + } + return nil } func (m *MemStore) GetEventsSince(_ context.Context, afterSeq int64, limit int) ([]db.PersistedEvent, error) { @@ -95,6 +100,11 @@ func (m *MemStore) GetEventsSinceForChannels(_ context.Context, afterSeq int64, return out, nil } +func (m *MemStore) GetMaxEventSeq(_ context.Context) (int64, error) { + es := m.ensureEvents() + return es.nextSeq.Load(), nil +} + func (m *MemStore) PruneEventsOlderThan(_ context.Context, cutoff time.Time) (int64, error) { es := m.ensureEvents() es.mu.Lock() diff --git a/Server/store/postgres.go b/Server/store/postgres.go index a1dead5f..66a16ea3 100644 --- a/Server/store/postgres.go +++ b/Server/store/postgres.go @@ -576,8 +576,8 @@ func (s *PostgresStore) GetAllSettings() (map[string]string, error) { // ── EventStore (stubs — Phase B Step 7) ───────────────────────────────────── -func (s *PostgresStore) PersistEvent(ctx context.Context, eventType string, channelID int64, payload []byte) (int64, error) { - return 0, ErrPostgresNotImplemented +func (s *PostgresStore) PersistEvent(ctx context.Context, seq int64, eventType string, channelID int64, payload []byte) error { + return ErrPostgresNotImplemented } func (s *PostgresStore) GetEventsSince(ctx context.Context, afterSeq int64, limit int) ([]db.PersistedEvent, error) { @@ -592,6 +592,10 @@ func (s *PostgresStore) PruneEventsOlderThan(ctx context.Context, cutoff time.Ti return 0, ErrPostgresNotImplemented } +func (s *PostgresStore) GetMaxEventSeq(ctx context.Context) (int64, error) { + return 0, ErrPostgresNotImplemented +} + // ── PluginStore (stubs — Phase C Step 9) ──────────────────────────────────── func (s *PostgresStore) InstallPlugin(ctx context.Context, name, version, manifestJSON string) (int64, error) { diff --git a/Server/store/sqlite_events.go b/Server/store/sqlite_events.go index 6b54ff97..08980396 100644 --- a/Server/store/sqlite_events.go +++ b/Server/store/sqlite_events.go @@ -2,6 +2,7 @@ package store import ( "context" + "database/sql" "fmt" "strings" "time" @@ -11,21 +12,19 @@ import ( // ── EventStore (Phase B Step 7) ───────────────────────────────────────────── -// PersistEvent appends a single event to the events table and returns the -// auto-assigned seq. -func (s *SQLiteStore) PersistEvent(ctx context.Context, eventType string, channelID int64, payload []byte) (int64, error) { - res, err := s.db.SQLDb().ExecContext(ctx, - `INSERT INTO events (event_type, channel_id, payload) VALUES (?, ?, ?)`, - eventType, channelID, payload, +// PersistEvent appends a single event to the events table with the +// caller-supplied seq. The hub assigns seq before this is called so the row +// seq always matches the wrapped-payload seq, even if the persister drops +// some events under load. +func (s *SQLiteStore) PersistEvent(ctx context.Context, seq int64, eventType string, channelID int64, payload []byte) error { + _, err := s.db.SQLDb().ExecContext(ctx, + `INSERT INTO events (seq, event_type, channel_id, payload) VALUES (?, ?, ?, ?)`, + seq, eventType, channelID, payload, ) if err != nil { - return 0, fmt.Errorf("PersistEvent: %w", err) + return fmt.Errorf("PersistEvent: %w", err) } - id, err := res.LastInsertId() - if err != nil { - return 0, fmt.Errorf("PersistEvent LastInsertId: %w", err) - } - return id, nil + return nil } // GetEventsSince returns events with seq > afterSeq up to limit, ordered ASC. @@ -92,6 +91,19 @@ func (s *SQLiteStore) GetEventsSinceForChannels(ctx context.Context, afterSeq in return scanEventRows(rows) } +// GetMaxEventSeq returns the largest seq in the events table, or 0 if empty. +func (s *SQLiteStore) GetMaxEventSeq(ctx context.Context) (int64, error) { + var maxSeq sql.NullInt64 + err := s.db.SQLDb().QueryRowContext(ctx, `SELECT MAX(seq) FROM events`).Scan(&maxSeq) + if err != nil { + return 0, fmt.Errorf("GetMaxEventSeq: %w", err) + } + if !maxSeq.Valid { + return 0, nil + } + return maxSeq.Int64, nil +} + // PruneEventsOlderThan deletes events older than cutoff. Returns rows deleted. func (s *SQLiteStore) PruneEventsOlderThan(ctx context.Context, cutoff time.Time) (int64, error) { res, err := s.db.SQLDb().ExecContext(ctx, diff --git a/Server/store/store.go b/Server/store/store.go index 1f968bdc..925a078b 100644 --- a/Server/store/store.go +++ b/Server/store/store.go @@ -202,9 +202,13 @@ type SettingsStore interface { // // Phase B Step 7 — Event Persistence Layer. type EventStore interface { - // PersistEvent appends an event to the persistent log and returns the - // auto-assigned seq. channelID == 0 means the event was a global broadcast. - PersistEvent(ctx context.Context, eventType string, channelID int64, payload []byte) (int64, error) + // PersistEvent appends an event with the hub-assigned seq. seq must be + // the same monotonic counter the wrapped payload exposes to clients so + // that cold-replay queries by seq return rows whose payload seq matches + // the row seq. channelID == 0 means the event was a global broadcast. + // Implementations should be tolerant of out-of-order seq insertion (e.g. + // they SHOULD NOT rely on AUTOINCREMENT semantics). + PersistEvent(ctx context.Context, seq int64, eventType string, channelID int64, payload []byte) error // GetEventsSince returns up to limit events with seq > afterSeq, ordered // by seq ascending. Used as a fallback after the ring buffer misses. @@ -218,6 +222,12 @@ type EventStore interface { // PruneEventsOlderThan deletes events with created_at < cutoff. Returns the // number of deleted rows. Called periodically by the pruner goroutine. PruneEventsOlderThan(ctx context.Context, cutoff time.Time) (int64, error) + + // GetMaxEventSeq returns the largest seq in the events table, or 0 if the + // table is empty. Used at startup to seed the hub's in-memory monotonic + // counter so wrapped-payload seqs stay aligned with row seqs across + // restarts. Returns 0 (without error) when the table is empty. + GetMaxEventSeq(ctx context.Context) (int64, error) } // PluginStore manages installed plugins and per-plugin KV namespaces. diff --git a/Server/ws/event_persister.go b/Server/ws/event_persister.go index 726896e4..e943bcf8 100644 --- a/Server/ws/event_persister.go +++ b/Server/ws/event_persister.go @@ -19,7 +19,10 @@ import ( ) // pendingEvent is a single event waiting to be flushed to the EventStore. +// seq carries the hub-assigned monotonic sequence so the row written to the +// store has the same seq as the wrapped payload sent to clients. type pendingEvent struct { + seq int64 eventType string channelID int64 payload []byte @@ -32,9 +35,11 @@ type EventPersister struct { batchSize int flushEvy time.Duration - stopOnce sync.Once - stop chan struct{} - done chan struct{} + startOnce sync.Once + started atomic.Bool + stopOnce sync.Once + stop chan struct{} + done chan struct{} persisted atomic.Uint64 dropped atomic.Uint64 @@ -65,13 +70,21 @@ func NewEventPersister(s store.EventStore, queueSize, batchSize int, flushEvery } } -// Start launches the background flusher goroutine. +// Start launches the background flusher goroutine. Idempotent — calling +// Start more than once is a no-op so test setups that share a persister +// across cases don't spawn duplicate runners. func (p *EventPersister) Start(ctx context.Context) { - go p.run(ctx) + p.startOnce.Do(func() { + p.started.Store(true) + go p.run(ctx) + }) } // Enqueue queues an event for persistence. Non-blocking; drops on full queue. -func (p *EventPersister) Enqueue(eventType string, channelID int64, payload []byte) { +// seq is the hub-assigned monotonic sequence for this event — it must be the +// same value embedded in the wrapped payload so reconnect replay returns rows +// whose row-seq matches the payload-seq the client tracks. +func (p *EventPersister) Enqueue(seq int64, eventType string, channelID int64, payload []byte) { if p == nil { return } @@ -79,19 +92,25 @@ func (p *EventPersister) Enqueue(eventType string, channelID int64, payload []by cp := make([]byte, len(payload)) copy(cp, payload) select { - case p.queue <- pendingEvent{eventType: eventType, channelID: channelID, payload: cp}: + case p.queue <- pendingEvent{seq: seq, eventType: eventType, channelID: channelID, payload: cp}: default: p.dropped.Add(1) } } // Stop signals the persister to drain remaining events and exit. Blocks until -// the goroutine exits or ctx is cancelled. +// the goroutine exits or ctx is cancelled. Safe to call without a prior +// Start: in that case there's no goroutine to wait for and Stop returns +// immediately after closing the stop channel. func (p *EventPersister) Stop(ctx context.Context) { if p == nil { return } p.stopOnce.Do(func() { close(p.stop) }) + if !p.started.Load() { + // run() was never launched, so done will never be closed. + return + } select { case <-p.done: case <-ctx.Done(): @@ -115,10 +134,10 @@ func (p *EventPersister) run(ctx context.Context) { } p.flushes.Add(1) for _, evt := range batch { - if _, err := p.store.PersistEvent(ctx, evt.eventType, evt.channelID, evt.payload); err != nil { + if err := p.store.PersistEvent(ctx, evt.seq, evt.eventType, evt.channelID, evt.payload); err != nil { p.errors.Add(1) slog.Warn("event persister: PersistEvent failed", - "event_type", evt.eventType, "channel_id", evt.channelID, "err", err) + "seq", evt.seq, "event_type", evt.eventType, "channel_id", evt.channelID, "err", err) continue } p.persisted.Add(1) diff --git a/Server/ws/event_persister_test.go b/Server/ws/event_persister_test.go index 2e54dc07..576a02a9 100644 --- a/Server/ws/event_persister_test.go +++ b/Server/ws/event_persister_test.go @@ -22,7 +22,7 @@ func TestEventPersisterFlushesBatch(t *testing.T) { t.Cleanup(func() { p.Stop(ctx) }) for i := 0; i < 10; i++ { - p.Enqueue("broadcast", 0, []byte(`{"type":"x"}`)) + p.Enqueue(int64(i+1), "broadcast", 0, []byte(`{"type":"x"}`)) } // Wait for at least one flush tick. @@ -62,8 +62,12 @@ func TestEventPersisterDropsOnFullQueue(t *testing.T) { p := NewEventPersister(mem, 2, 1024, time.Hour) // NB: Start is intentionally NOT called so the queue stays full. for i := 0; i < 50; i++ { - p.Enqueue("broadcast", 0, []byte(`{}`)) + p.Enqueue(int64(i+1), "broadcast", 0, []byte(`{}`)) } + // Stop without Start — must not deadlock. + stopCtx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + p.Stop(stopCtx) _, dropped, _, _ := p.Stats() if dropped == 0 { t.Fatal("expected drops with full queue and no consumer") @@ -76,7 +80,7 @@ func TestEventPersisterStopDrains(t *testing.T) { p.Start(context.Background()) for i := 0; i < 5; i++ { - p.Enqueue("broadcast", 0, []byte(`{}`)) + p.Enqueue(int64(i+1), "broadcast", 0, []byte(`{}`)) } stopCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) diff --git a/Server/ws/hub.go b/Server/ws/hub.go index 0856a2b5..a07f0754 100644 --- a/Server/ws/hub.go +++ b/Server/ws/hub.go @@ -42,7 +42,7 @@ type Hub struct { registry *HandlerRegistry permChecker *permissions.Checker - pubsub *PubSub // topic-based pub/sub for O(subscribers) broadcast + pubsub *PubSub // topic-based pub/sub for O(subscribers) broadcast topicLimiter *TopicRateLimiter // per-topic throughput caps seq uint64 // atomic monotonic sequence counter @@ -548,7 +548,7 @@ func (h *Hub) sendSequencedToUsers(channelID int64, userIDs []int64, msg []byte) // Store DM event for reconnect replay; filtering is channel-based and uses // allowed channel IDs computed at auth time (including open DMs). h.replayBuf.Push(seq, channelID, wrapped) - h.persistEvent(channelID, wrapped) + h.persistEvent(seq, channelID, wrapped) for _, userID := range userIDs { h.SendToUser(userID, wrapped) @@ -563,7 +563,7 @@ func (h *Hub) sendSequencedToUsersHigh(channelID int64, userIDs []int64, msg []b seq := h.nextSeq() wrapped := wrapWithSeq(msg, seq) h.replayBuf.Push(seq, channelID, wrapped) - h.persistEvent(channelID, wrapped) + h.persistEvent(seq, channelID, wrapped) for _, userID := range userIDs { h.SendToUserHigh(userID, wrapped) @@ -619,6 +619,22 @@ func (h *Hub) ReplayBuffer() *EventRingBuffer { return h.replayBuf } +// SeedSeq sets the hub's monotonic sequence counter to seed (atomic). Used +// at startup to align in-memory seqs with the persisted MAX(events.seq) so +// wrapped-payload seqs stay monotonic across restarts. Calling SeedSeq with +// a value less than the current seq is a no-op (we never go backwards). +func (h *Hub) SeedSeq(seed uint64) { + for { + cur := atomic.LoadUint64(&h.seq) + if seed <= cur { + return + } + if atomic.CompareAndSwapUint64(&h.seq, cur, seed) { + return + } + } +} + // SetEventPersister attaches a persister so subsequent broadcasts are also // written to the persistent EventStore. Pass nil to disable. func (h *Hub) SetEventPersister(p *EventPersister) { @@ -639,8 +655,11 @@ func (h *Hub) ReconnectTierStats() (buffer, db, full uint64) { } // persistEvent enqueues a broadcast event for cold-storage persistence. Safe -// to call with a nil persister; never blocks the broadcast hot path. -func (h *Hub) persistEvent(channelID int64, payload []byte) { +// to call with a nil persister; never blocks the broadcast hot path. seq is +// the same hub-assigned monotonic counter embedded in payload, so the row +// written to the EventStore has a row-seq that matches the wrapped-payload +// seq the client tracks. +func (h *Hub) persistEvent(seq uint64, channelID int64, payload []byte) { if h.eventPersister == nil { return } @@ -648,7 +667,7 @@ func (h *Hub) persistEvent(channelID int64, payload []byte) { if channelID != 0 { eventType = "channel_broadcast" } - h.eventPersister.Enqueue(eventType, channelID, payload) + h.eventPersister.Enqueue(int64(seq), eventType, channelID, payload) } // wrapWithSeq injects a "seq" field into a JSON message without re-serializing. @@ -800,7 +819,7 @@ func (h *Hub) deliverBroadcast(bm broadcastMsg) { // Store in replay buffer for reconnection recovery. h.replayBuf.Push(seq, bm.channelID, msg) - h.persistEvent(bm.channelID, msg) + h.persistEvent(seq, bm.channelID, msg) if bm.channelID == 0 { // Global broadcast — deliver to every connected client.