From 086bc4ea4a4b4968d8a2f4579e5d610f057cf3a2 Mon Sep 17 00:00:00 2001 From: mirivlad Date: Fri, 10 Jul 2026 03:13:17 +0800 Subject: [PATCH] fix: isolate sync operations by user and vault --- README.md | 15 +- .../plans/2026-07-10-sync-tenant-isolation.md | 50 +-- ...2026-07-10-sync-tenant-isolation-design.md | 19 +- internal/server/handlers_api.go | 67 ++-- internal/server/middleware.go | 82 +++-- internal/server/schema.go | 213 ++++++++++- internal/server/server.go | 4 + internal/server/server_test.go | 345 +++++++++++++++++- 8 files changed, 712 insertions(+), 83 deletions(-) diff --git a/README.md b/README.md index a286dc2..a831056 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ Standalone sync server for Verstak2 platform. This server provides synchronization between devices running Verstak2. It handles: - Device registration and authentication -- Operation log sync with server sequence numbers and conflict detection +- Vault-scoped operation log sync with server sequence numbers and conflict detection - Blob storage for attachments - User management with email confirmation @@ -139,7 +139,7 @@ internal/server/ - Server implementation Desktop sync client: -- `POST /api/client/pair` - Pair a desktop client with username/password and return a device token +- `POST /api/client/pair` - Pair a desktop client with username/password and its persistent `vault_id`, then return a device token - `POST /api/auth/test` - Validate username/password from the desktop client - `GET /api/client/me` - Return current authenticated client/device details - `POST /api/client/revoke-current` - Revoke the current desktop device token @@ -165,8 +165,15 @@ Operational endpoints: - `/register`, `/login`, `/dashboard`, `/forgot`, `/reset`, `/logout` - User web UI Sync operations are generic records with `entity_type`, `entity_id`, `op_type`, -`payload_json`, `device_id`, and sequencing metadata. The server stores and -orders operations; Verstak desktop owns the v2 payload semantics. +`payload_json`, `device_id`, and sequencing metadata. A pairing token is bound +to one user and vault. The server derives the stored device ID and operation +scope from that token, ignores a caller-supplied `device_id` for authorization, +and returns only operations and cursors from the authenticated user/vault. +Verstak desktop owns the v2 payload semantics. + +New device enrollment requires a non-empty `vault_id`. The `legacy:` prefix is +reserved for server-side migration of older records and cannot be selected by +new clients. ## Development diff --git a/docs/superpowers/plans/2026-07-10-sync-tenant-isolation.md b/docs/superpowers/plans/2026-07-10-sync-tenant-isolation.md index 48ae305..960811a 100644 --- a/docs/superpowers/plans/2026-07-10-sync-tenant-isolation.md +++ b/docs/superpowers/plans/2026-07-10-sync-tenant-isolation.md @@ -18,16 +18,16 @@ migrated into a deterministic legacy scope. **Files:** - Modify: `internal/server/server_test.go` -- [ ] Add helpers that create confirmed users and token-authenticated devices +- [x] Add helpers that create confirmed users and token-authenticated devices with a specified vault ID. -- [ ] Add a failing test where separate users push and pull from the same +- [x] Add a failing test where separate users push and pull from the same vault ID; each pull must contain only its own operation and cursor. -- [ ] Add a failing test where one user owns devices in two vaults; pulls must +- [x] Add a failing test where one user owns devices in two vaults; pulls must remain vault-local. -- [ ] Add a failing test that sends another device's ID in `push`; assert the +- [x] Add a failing test that sends another device's ID in `push`; assert the stored and returned operation uses the authenticated device ID. -- [ ] Add a failing test for identical idempotency keys in different scopes. -- [ ] Run: `go test ./internal/server -run 'TestSync.*Isolation|TestSyncPush'` +- [x] Add a failing test for identical idempotency keys in different scopes. +- [x] Run: `go test ./internal/server -run 'TestSync.*Isolation|TestSyncPush'` and confirm the new assertions fail for the intended missing behaviour. ## Task 2: Add idempotent SQLite scope migration @@ -37,15 +37,15 @@ migrated into a deterministic legacy scope. - Modify: `internal/server/server.go` - Test: `internal/server/server_test.go` -- [ ] Define `vault_id` on new devices and `user_id`/`vault_id` on new +- [x] Define `vault_id` on new devices and `user_id`/`vault_id` on new operations; define scoped tombstone and idempotency primary keys. -- [ ] Add startup migration helpers that inspect columns, add compatible +- [x] Add startup migration helpers that inspect columns, add compatible columns, backfill owner IDs, assign `legacy:` to old scopes, and rebuild the two tables whose primary keys change. -- [ ] Add a failing legacy-schema fixture test, then make it pass by opening +- [x] Add a failing legacy-schema fixture test, then make it pass by opening the database through `NewServer` and asserting its operation has the expected owner and legacy scope. -- [ ] Run: `go test ./internal/server -run 'Test.*Migration|TestSync.*'`. +- [x] Run: `go test ./internal/server -run 'Test.*Migration|TestSync.*'`. ## Task 3: Apply authenticated scope to sync handlers @@ -54,15 +54,15 @@ migrated into a deterministic legacy scope. - Modify: `internal/server/handlers_api.go` - Test: `internal/server/server_test.go` -- [ ] Extend authenticated device lookup to provide the effective vault scope; +- [x] Extend authenticated device lookup to provide the effective vault scope; missing user ownership must not authorize sync operations. -- [ ] Require `vault_id` when creating a new client pairing and store it with +- [x] Require `vault_id` when creating a new client pairing and store it with the device. -- [ ] Make push use authenticated device/user/vault values for inserts, +- [x] Make push use authenticated device/user/vault values for inserts, conflicts, revisions, tombstones, and idempotency lookup/storage. -- [ ] Make pull filter operations and its reported cursor by authenticated +- [x] Make pull filter operations and its reported cursor by authenticated user/vault. -- [ ] Run the focused tests from Task 1 until green, then +- [x] Run the focused tests from Task 1 until green, then `go test ./internal/server`. ## Task 4: Send the current vault ID while pairing @@ -73,13 +73,17 @@ migrated into a deterministic legacy scope. - Modify: `../verstak-desktop/internal/api/app.go` - Modify: `../verstak-desktop/internal/api/app_test.go` -- [ ] Add `vault_id` to the pair request and expose it in the pairing client +- [x] Add `vault_id` to the pair request and expose it in the pairing client method without changing push/pull wire compatibility. -- [ ] Read the open vault metadata in `syncConfigure`; reject configuration if +- [x] Read the open vault metadata in `syncConfigure`; reject configuration if the vault ID is absent. -- [ ] Add a failing client/API test that captures the pair request and asserts +- [x] Add a failing client/API test that captures the pair request and asserts the persistent vault ID is sent. -- [ ] Run: `go test ./internal/core/sync ./internal/api`. +- [x] Rebind desktop sync state, cursor, and persisted device identity whenever + a vault is created, opened, or switched. +- [x] Hydrate missing legacy vault device IDs from the authenticated sync + server before a token-based sync can use a global fallback. +- [x] Run: `go test ./internal/core/sync ./internal/api`. ## Task 5: Document, verify, and publish @@ -87,9 +91,9 @@ migrated into a deterministic legacy scope. - Modify: `README.md` - Modify: `docs/superpowers/specs/2026-07-10-sync-tenant-isolation-design.md` -- [ ] Document that pairing is vault-bound and that sync cursors are scoped. -- [ ] Run `gofmt` on all changed Go files. -- [ ] Run `go test ./...` in both `verstak-sync-server` and +- [x] Document that pairing is vault-bound and that sync cursors are scoped. +- [x] Run `gofmt` on all changed Go files. +- [x] Run `go test ./...` in both `verstak-sync-server` and `verstak-desktop`, then `git diff --check` in both repositories. -- [ ] Commit and push the sync-server and desktop changes as coordinated +- [x] Commit and push the sync-server and desktop changes as coordinated security commits. diff --git a/docs/superpowers/specs/2026-07-10-sync-tenant-isolation-design.md b/docs/superpowers/specs/2026-07-10-sync-tenant-isolation-design.md index 0cafe81..dc64f86 100644 --- a/docs/superpowers/specs/2026-07-10-sync-tenant-isolation-design.md +++ b/docs/superpowers/specs/2026-07-10-sync-tenant-isolation-design.md @@ -23,9 +23,10 @@ in this change. ## Data model -`server_devices` gains a nullable `vault_id`. A device created through -`/api/client/pair` must have a non-empty vault ID. The authenticated device -therefore identifies one user and one vault. +`server_devices` gains a nullable `vault_id`. A device created through either +enrollment endpoint must have a non-empty vault ID that does not use the +reserved `legacy:` prefix. The authenticated device therefore identifies one +user and one vault. `server_ops` gains `user_id` and `vault_id`. New writes always set both from the authenticated device. Pull and conflict queries filter both fields. @@ -40,6 +41,12 @@ legacy scope during startup migration. This preserves existing single-vault accounts while preventing data from crossing account boundaries. New pairings never use the legacy scope. +Desktop sync state, operation queues, cursors, and persisted device IDs are +vault-local. The desktop recreates its sync service whenever the active vault +is created, opened, or switched. When it opens a legacy vault state without a +stored device ID, it obtains the authenticated ID from `/api/client/me` before +syncing and persists it locally. + ## API contract `POST /api/client/pair` accepts a required `vault_id`. The desktop gets it @@ -61,6 +68,10 @@ new operation/device fields, backfills `user_id` from each operation's device, and assigns the explicit legacy scope when an old device has no vault ID. Tables whose primary key must change are rebuilt transactionally. +The prior global idempotency cache is intentionally discarded during migration: +it is only a replay cache and retaining it could replay one tenant's response +for another tenant. + If an operation cannot be associated with a user, it remains unscoped and is not readable through sync APIs. The server must not guess an owner from a request body. @@ -75,5 +86,7 @@ Focused server tests must prove: 4. scoped idempotency does not replay another tenant's response; 5. a legacy SQLite database is upgraded with its existing operation retained in the matching legacy scope. +6. switching the active desktop vault rebinds the sync queue, cursor, and + device identity to the new vault. Desktop tests must prove that pairing sends the opened vault's persistent ID. diff --git a/internal/server/handlers_api.go b/internal/server/handlers_api.go index ffc30b7..ab13d10 100644 --- a/internal/server/handlers_api.go +++ b/internal/server/handlers_api.go @@ -50,6 +50,7 @@ func (s *Server) handleClientPair(w http.ResponseWriter, r *http.Request) { Password string `json:"password"` DeviceName string `json:"device_name"` ClientVersion string `json:"client_version"` + VaultID string `json:"vault_id"` } if err := json.NewDecoder(r.Body).Decode(&req); err != nil { jsonErr(w, 400, "bad json") @@ -59,6 +60,15 @@ func (s *Server) handleClientPair(w http.ResponseWriter, r *http.Request) { jsonErr(w, 400, "login and password required") return } + req.VaultID = strings.TrimSpace(req.VaultID) + if req.VaultID == "" { + jsonErr(w, 400, "vault_id required") + return + } + if strings.HasPrefix(req.VaultID, "legacy:") { + jsonErr(w, 400, "vault_id uses reserved prefix") + return + } if req.DeviceName == "" { req.DeviceName = "unknown" } @@ -95,10 +105,10 @@ func (s *Server) handleClientPair(w http.ResponseWriter, r *http.Request) { apiKey := make([]byte, 20) rand.Read(apiKey) _, err = s.db.Exec(`INSERT INTO server_devices - (id, name, api_key, token_hash, token_prefix, token_suffix, user_id, client_version, last_ip, last_seen, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + (id, name, api_key, token_hash, token_prefix, token_suffix, user_id, vault_id, client_version, last_ip, last_seen, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, deviceID, req.DeviceName, hex.EncodeToString(apiKey), tokenHash, prefix, suffix, - userID, req.ClientVersion, ip, now, now) + userID, req.VaultID, req.ClientVersion, ip, now, now) if err != nil { jsonErr(w, 500, err.Error()) return @@ -272,6 +282,7 @@ func (s *Server) handleDeviceRegister(w http.ResponseWriter, r *http.Request) { Name string `json:"name"` Username string `json:"username"` Password string `json:"password"` + VaultID string `json:"vault_id"` } if err := json.NewDecoder(r.Body).Decode(&req); err != nil { jsonErr(w, 400, "invalid JSON") @@ -285,6 +296,15 @@ func (s *Server) handleDeviceRegister(w http.ResponseWriter, r *http.Request) { jsonErr(w, 401, "username and password required") return } + req.VaultID = strings.TrimSpace(req.VaultID) + if req.VaultID == "" { + jsonErr(w, 400, "vault_id required") + return + } + if strings.HasPrefix(req.VaultID, "legacy:") { + jsonErr(w, 400, "vault_id uses reserved prefix") + return + } var userID, hash string var confirmed, blocked int err := s.db.QueryRow("SELECT id, password_hash, confirmed, blocked FROM server_users WHERE username=? OR email=?", @@ -311,8 +331,8 @@ func (s *Server) handleDeviceRegister(w http.ResponseWriter, r *http.Request) { deviceID := apiKey[:12] now := time.Now().UTC().Format(time.RFC3339) _, err = s.db.Exec( - "INSERT INTO server_devices (id, name, api_key, last_seen, created_at) VALUES (?, ?, ?, ?, ?)", - deviceID, req.Name, apiKey, now, now, + "INSERT INTO server_devices (id, name, api_key, user_id, vault_id, last_seen, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)", + deviceID, req.Name, apiKey, userID, req.VaultID, now, now, ) if err != nil { jsonErr(w, 500, err.Error()) @@ -326,7 +346,7 @@ func (s *Server) handleDeviceRegister(w http.ResponseWriter, r *http.Request) { } func (s *Server) handleSyncPush(w http.ResponseWriter, r *http.Request) { - _, _, ok := s.requireAuth(w, r) + scope, ok := s.requireSyncScope(w, r) if !ok { return } @@ -354,7 +374,9 @@ func (s *Server) handleSyncPush(w http.ResponseWriter, r *http.Request) { } if req.IdempotencyKey != "" { var cachedJSON string - err := s.db.QueryRow("SELECT response_json FROM server_idempotency_keys WHERE idempotency_key=?", req.IdempotencyKey).Scan(&cachedJSON) + err := s.db.QueryRow(`SELECT response_json FROM server_idempotency_keys + WHERE user_id=? AND vault_id=? AND idempotency_key=?`, + scope.UserID, scope.VaultID, req.IdempotencyKey).Scan(&cachedJSON) if err == nil { w.Header().Set("Content-Type", "application/json") w.Write([]byte(cachedJSON)) @@ -371,9 +393,9 @@ func (s *Server) handleSyncPush(w http.ResponseWriter, r *http.Request) { if op.LastSeenServerSeq > 0 { conflictRows, err := s.db.Query(` SELECT op_id, device_id, op_type, server_sequence FROM server_ops - WHERE entity_type=? AND entity_id=? AND device_id!=? + WHERE user_id=? AND vault_id=? AND entity_type=? AND entity_id=? AND device_id!=? AND server_sequence > ? AND op_type != 'delete' - ORDER BY server_sequence`, op.EntityType, op.EntityID, req.DeviceID, op.LastSeenServerSeq) + ORDER BY server_sequence`, scope.UserID, scope.VaultID, op.EntityType, op.EntityID, scope.DeviceID, op.LastSeenServerSeq) if err == nil { for conflictRows.Next() { var cOpID, cDevID, cOpType string @@ -392,9 +414,9 @@ func (s *Server) handleSyncPush(w http.ResponseWriter, r *http.Request) { } } res, err := s.db.Exec( - `INSERT OR IGNORE INTO server_ops (op_id, server_sequence, device_id, entity_type, entity_id, op_type, payload_json, idempotency_key, client_sequence, last_seen_server_seq, created_at, pushed_at) - VALUES (?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, - op.OpID, req.DeviceID, op.EntityType, op.EntityID, op.OpType, op.PayloadJSON, + `INSERT OR IGNORE INTO server_ops (op_id, server_sequence, user_id, vault_id, device_id, entity_type, entity_id, op_type, payload_json, idempotency_key, client_sequence, last_seen_server_seq, created_at, pushed_at) + VALUES (?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + op.OpID, scope.UserID, scope.VaultID, scope.DeviceID, op.EntityType, op.EntityID, op.OpType, op.PayloadJSON, req.IdempotencyKey, op.ClientSequence, op.LastSeenServerSeq, op.CreatedAt, now, ) if err != nil { @@ -404,15 +426,16 @@ func (s *Server) handleSyncPush(w http.ResponseWriter, r *http.Request) { if n == 0 { continue } - seqRes, err := s.db.Exec("INSERT INTO server_revisions (op_id, device_id) VALUES (?, ?)", op.OpID, req.DeviceID) + seqRes, err := s.db.Exec("INSERT INTO server_revisions (op_id, device_id) VALUES (?, ?)", op.OpID, scope.DeviceID) if err != nil { continue } seq, _ := seqRes.LastInsertId() s.db.Exec("UPDATE server_ops SET server_sequence=? WHERE op_id=?", seq, op.OpID) if op.OpType == "delete" { - s.db.Exec(`INSERT OR REPLACE INTO server_tombstones (entity_type, entity_id, op_id, deleted_at) VALUES (?, ?, ?, ?)`, - op.EntityType, op.EntityID, op.OpID, now) + s.db.Exec(`INSERT OR REPLACE INTO server_tombstones + (user_id, vault_id, entity_type, entity_id, op_id, deleted_at) VALUES (?, ?, ?, ?, ?, ?)`, + scope.UserID, scope.VaultID, op.EntityType, op.EntityID, op.OpID, now) } accepted = append(accepted, op.OpID) } @@ -423,15 +446,16 @@ func (s *Server) handleSyncPush(w http.ResponseWriter, r *http.Request) { } if req.IdempotencyKey != "" { if respJSON, err := json.Marshal(resp); err == nil { - s.db.Exec("INSERT OR IGNORE INTO server_idempotency_keys (idempotency_key, response_json, created_at) VALUES (?, ?, ?)", - req.IdempotencyKey, string(respJSON), now) + s.db.Exec(`INSERT OR IGNORE INTO server_idempotency_keys + (user_id, vault_id, idempotency_key, response_json, created_at) VALUES (?, ?, ?, ?, ?)`, + scope.UserID, scope.VaultID, req.IdempotencyKey, string(respJSON), now) } } jsonOK(w, resp) } func (s *Server) handleSyncPull(w http.ResponseWriter, r *http.Request) { - _, _, ok := s.requireAuth(w, r) + scope, ok := s.requireSyncScope(w, r) if !ok { return } @@ -447,12 +471,13 @@ func (s *Server) handleSyncPull(w http.ResponseWriter, r *http.Request) { return } var serverSeq int - s.db.QueryRow("SELECT COALESCE(MAX(server_sequence), 0) FROM server_ops").Scan(&serverSeq) + s.db.QueryRow(`SELECT COALESCE(MAX(server_sequence), 0) FROM server_ops + WHERE user_id=? AND vault_id=?`, scope.UserID, scope.VaultID).Scan(&serverSeq) rows, err := s.db.Query(` SELECT op_id, server_sequence, device_id, entity_type, entity_id, op_type, payload_json, created_at FROM server_ops - WHERE server_sequence > ? AND server_sequence IS NOT NULL - ORDER BY server_sequence`, req.SinceSequence) + WHERE user_id=? AND vault_id=? AND server_sequence > ? AND server_sequence IS NOT NULL + ORDER BY server_sequence`, scope.UserID, scope.VaultID, req.SinceSequence) if err != nil { jsonErr(w, 500, err.Error()) return diff --git a/internal/server/middleware.go b/internal/server/middleware.go index 950ca97..96bdee1 100644 --- a/internal/server/middleware.go +++ b/internal/server/middleware.go @@ -7,7 +7,38 @@ import ( "time" ) +type authenticatedDevice struct { + DeviceID string + UserID string + VaultID string +} + func (s *Server) requireAuth(w http.ResponseWriter, r *http.Request) (deviceID, userID string, ok bool) { + device, ok := s.authenticateDevice(w, r) + if !ok { + return "", "", false + } + return device.DeviceID, device.UserID, true +} + +func (s *Server) requireSyncScope(w http.ResponseWriter, r *http.Request) (authenticatedDevice, bool) { + device, ok := s.authenticateDevice(w, r) + if !ok { + return authenticatedDevice{}, false + } + if device.UserID == "" { + jsonErr(w, http.StatusForbidden, "device is not associated with a user") + return authenticatedDevice{}, false + } + device.VaultID = effectiveVaultScope(device.UserID, device.VaultID) + if device.VaultID == "" { + jsonErr(w, http.StatusForbidden, "device is not associated with a vault") + return authenticatedDevice{}, false + } + return device, true +} + +func (s *Server) authenticateDevice(w http.ResponseWriter, r *http.Request) (authenticatedDevice, bool) { key := r.Header.Get("Authorization") key = strings.TrimPrefix(key, "Bearer ") if key == "" { @@ -15,46 +46,43 @@ func (s *Server) requireAuth(w http.ResponseWriter, r *http.Request) (deviceID, } if key == "" { jsonErr(w, 401, "API key required") - return "", "", false + return authenticatedDevice{}, false } - hash := sha256Hex(key) - var deviceIDVal, userIDVal, revokedAt sql.NullString - err := s.db.QueryRow("SELECT id, user_id, revoked_at FROM server_devices WHERE token_hash=?", hash).Scan(&deviceIDVal, &userIDVal, &revokedAt) - if err == nil { - if revokedAt.Valid && revokedAt.String != "" { - jsonErr(w, 401, "device revoked") - return "", "", false - } - if userIDVal.Valid && userIDVal.String != "" { - var blocked int - s.db.QueryRow("SELECT blocked FROM server_users WHERE id=?", userIDVal.String).Scan(&blocked) - if blocked != 0 { - jsonErr(w, 403, "user blocked") - return "", "", false - } - } - s.db.Exec("UPDATE server_devices SET last_seen=? WHERE id=?", time.Now().UTC().Format(time.RFC3339), deviceIDVal.String) - return deviceIDVal.String, userIDVal.String, true + + var device authenticatedDevice + var userID, vaultID, revokedAt sql.NullString + err := s.db.QueryRow(`SELECT id, user_id, vault_id, revoked_at + FROM server_devices WHERE token_hash=?`, sha256Hex(key)). + Scan(&device.DeviceID, &userID, &vaultID, &revokedAt) + if err != nil { + err = s.db.QueryRow(`SELECT id, user_id, vault_id, revoked_at + FROM server_devices WHERE api_key=?`, key). + Scan(&device.DeviceID, &userID, &vaultID, &revokedAt) } - err = s.db.QueryRow("SELECT id, user_id, revoked_at FROM server_devices WHERE api_key=?", key).Scan(&deviceIDVal, &userIDVal, &revokedAt) if err != nil { jsonErr(w, 401, "invalid API key") - return "", "", false + return authenticatedDevice{}, false } if revokedAt.Valid && revokedAt.String != "" { jsonErr(w, 401, "device revoked") - return "", "", false + return authenticatedDevice{}, false } - if userIDVal.Valid && userIDVal.String != "" { + if userID.Valid { + device.UserID = userID.String + } + if vaultID.Valid { + device.VaultID = vaultID.String + } + if device.UserID != "" { var blocked int - s.db.QueryRow("SELECT blocked FROM server_users WHERE id=?", userIDVal.String).Scan(&blocked) + s.db.QueryRow("SELECT blocked FROM server_users WHERE id=?", device.UserID).Scan(&blocked) if blocked != 0 { jsonErr(w, 403, "user blocked") - return "", "", false + return authenticatedDevice{}, false } } - s.db.Exec("UPDATE server_devices SET last_seen=? WHERE id=?", time.Now().UTC().Format(time.RFC3339), deviceIDVal.String) - return deviceIDVal.String, userIDVal.String, true + s.db.Exec("UPDATE server_devices SET last_seen=? WHERE id=?", time.Now().UTC().Format(time.RFC3339), device.DeviceID) + return device, true } func (s *Server) requireAdmin(w http.ResponseWriter, r *http.Request) bool { diff --git a/internal/server/schema.go b/internal/server/schema.go index 8d8a6dc..8c3a6c1 100644 --- a/internal/server/schema.go +++ b/internal/server/schema.go @@ -1,5 +1,10 @@ package server +import ( + "database/sql" + "fmt" +) + const serverSchema = ` CREATE TABLE IF NOT EXISTS server_users ( id TEXT PRIMARY KEY, @@ -20,6 +25,7 @@ CREATE TABLE IF NOT EXISTS server_devices ( token_prefix TEXT, token_suffix TEXT, user_id TEXT, + vault_id TEXT, client_version TEXT, last_ip TEXT, last_seen TEXT, @@ -36,6 +42,8 @@ CREATE TABLE IF NOT EXISTS server_user_devices ( CREATE TABLE IF NOT EXISTS server_ops ( op_id TEXT PRIMARY KEY, server_sequence INTEGER, + user_id TEXT NOT NULL, + vault_id TEXT NOT NULL, device_id TEXT NOT NULL, entity_type TEXT NOT NULL, entity_id TEXT NOT NULL, @@ -49,17 +57,22 @@ CREATE TABLE IF NOT EXISTS server_ops ( ); CREATE TABLE IF NOT EXISTS server_tombstones ( + user_id TEXT NOT NULL, + vault_id TEXT NOT NULL, entity_type TEXT NOT NULL, entity_id TEXT NOT NULL, op_id TEXT NOT NULL, deleted_at TEXT NOT NULL, - PRIMARY KEY (entity_type, entity_id) + PRIMARY KEY (user_id, vault_id, entity_type, entity_id) ); CREATE TABLE IF NOT EXISTS server_idempotency_keys ( - idempotency_key TEXT PRIMARY KEY, + user_id TEXT NOT NULL, + vault_id TEXT NOT NULL, + idempotency_key TEXT NOT NULL, response_json TEXT NOT NULL, - created_at TEXT NOT NULL + created_at TEXT NOT NULL, + PRIMARY KEY (user_id, vault_id, idempotency_key) ); CREATE TABLE IF NOT EXISTS server_email_tokens ( @@ -107,3 +120,197 @@ CREATE TABLE IF NOT EXISTS server_smtp_config ( value TEXT NOT NULL ); ` + +type sqliteColumn struct { + primaryKeyOrder int +} + +func migrateServerSchema(db *sql.DB) error { + tx, err := db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + + for _, change := range []struct { + table string + column string + definition string + }{ + {"server_devices", "user_id", "TEXT"}, + {"server_devices", "vault_id", "TEXT"}, + {"server_ops", "user_id", "TEXT"}, + {"server_ops", "vault_id", "TEXT"}, + } { + if err := ensureSQLiteColumn(tx, change.table, change.column, change.definition); err != nil { + return err + } + } + + if err := backfillDeviceOwners(tx); err != nil { + return err + } + if err := backfillOperationScope(tx); err != nil { + return err + } + if err := migrateTombstoneScope(tx); err != nil { + return err + } + if err := migrateIdempotencyScope(tx); err != nil { + return err + } + if _, err := tx.Exec(`CREATE INDEX IF NOT EXISTS idx_server_ops_scope_sequence + ON server_ops(user_id, vault_id, server_sequence)`); err != nil { + return err + } + if _, err := tx.Exec(`CREATE INDEX IF NOT EXISTS idx_server_devices_scope + ON server_devices(user_id, vault_id)`); err != nil { + return err + } + + return tx.Commit() +} + +func ensureSQLiteColumn(tx *sql.Tx, table, column, definition string) error { + columns, err := sqliteTableColumns(tx, table) + if err != nil { + return err + } + if _, ok := columns[column]; ok { + return nil + } + if _, err := tx.Exec(fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s %s", table, column, definition)); err != nil { + return fmt.Errorf("add %s.%s: %w", table, column, err) + } + return nil +} + +func sqliteTableColumns(tx *sql.Tx, table string) (map[string]sqliteColumn, error) { + rows, err := tx.Query(fmt.Sprintf("PRAGMA table_info(%s)", table)) + if err != nil { + return nil, err + } + defer rows.Close() + + columns := make(map[string]sqliteColumn) + for rows.Next() { + var cid, notNull, primaryKeyOrder int + var name, dataType string + var defaultValue sql.NullString + if err := rows.Scan(&cid, &name, &dataType, ¬Null, &defaultValue, &primaryKeyOrder); err != nil { + return nil, err + } + columns[name] = sqliteColumn{primaryKeyOrder: primaryKeyOrder} + } + return columns, rows.Err() +} + +func backfillDeviceOwners(tx *sql.Tx) error { + _, err := tx.Exec(`UPDATE server_devices + SET user_id = ( + SELECT user_id FROM server_user_devices + WHERE device_id = server_devices.id + ) + WHERE COALESCE(user_id, '') = '' + AND 1 = ( + SELECT COUNT(*) FROM server_user_devices + WHERE device_id = server_devices.id + )`) + return err +} + +func backfillOperationScope(tx *sql.Tx) error { + if _, err := tx.Exec(`UPDATE server_ops + SET user_id = ( + SELECT user_id FROM server_devices + WHERE id = server_ops.device_id + ) + WHERE COALESCE(user_id, '') = ''`); err != nil { + return err + } + _, err := tx.Exec(`UPDATE server_ops + SET vault_id = COALESCE( + NULLIF((SELECT vault_id FROM server_devices WHERE id = server_ops.device_id), ''), + 'legacy:' || user_id + ) + WHERE COALESCE(vault_id, '') = '' + AND COALESCE(user_id, '') != ''`) + return err +} + +func migrateTombstoneScope(tx *sql.Tx) error { + if hasScopedPrimaryKey(tx, "server_tombstones", "user_id", "vault_id", "entity_type", "entity_id") { + return nil + } + if _, err := tx.Exec("ALTER TABLE server_tombstones RENAME TO server_tombstones_legacy"); err != nil { + return err + } + if _, err := tx.Exec(`CREATE TABLE server_tombstones ( + user_id TEXT NOT NULL, + vault_id TEXT NOT NULL, + entity_type TEXT NOT NULL, + entity_id TEXT NOT NULL, + op_id TEXT NOT NULL, + deleted_at TEXT NOT NULL, + PRIMARY KEY (user_id, vault_id, entity_type, entity_id) + )`); err != nil { + return err + } + if _, err := tx.Exec(`INSERT OR IGNORE INTO server_tombstones + (user_id, vault_id, entity_type, entity_id, op_id, deleted_at) + SELECT o.user_id, o.vault_id, legacy.entity_type, legacy.entity_id, legacy.op_id, legacy.deleted_at + FROM server_tombstones_legacy AS legacy + JOIN server_ops AS o ON o.op_id = legacy.op_id + WHERE COALESCE(o.user_id, '') != '' AND COALESCE(o.vault_id, '') != ''`); err != nil { + return err + } + _, err := tx.Exec("DROP TABLE server_tombstones_legacy") + return err +} + +func migrateIdempotencyScope(tx *sql.Tx) error { + if hasScopedPrimaryKey(tx, "server_idempotency_keys", "user_id", "vault_id", "idempotency_key") { + return nil + } + if _, err := tx.Exec("ALTER TABLE server_idempotency_keys RENAME TO server_idempotency_keys_legacy"); err != nil { + return err + } + if _, err := tx.Exec(`CREATE TABLE server_idempotency_keys ( + user_id TEXT NOT NULL, + vault_id TEXT NOT NULL, + idempotency_key TEXT NOT NULL, + response_json TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY (user_id, vault_id, idempotency_key) + )`); err != nil { + return err + } + // The previous cache was global. Discard it rather than risk replaying a + // response for another user or vault after the migration. + _, err := tx.Exec("DROP TABLE server_idempotency_keys_legacy") + return err +} + +func hasScopedPrimaryKey(tx *sql.Tx, table string, want ...string) bool { + columns, err := sqliteTableColumns(tx, table) + if err != nil { + return false + } + for index, name := range want { + column, ok := columns[name] + if !ok || column.primaryKeyOrder != index+1 { + return false + } + } + return true +} + +func effectiveVaultScope(userID, vaultID string) string { + if vaultID != "" { + return vaultID + } + if userID == "" { + return "" + } + return "legacy:" + userID +} diff --git a/internal/server/server.go b/internal/server/server.go index 3cd306a..2f84cf9 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -139,6 +139,10 @@ func NewServer(dbPath, dataDir string, cfg *Config) (*Server, error) { return nil, fmt.Errorf("schema: %w", err) } } + if err := migrateServerSchema(db); err != nil { + db.Close() + return nil, fmt.Errorf("migrate schema: %w", err) + } blobsDir := filepath.Join(dataDir, "blobs") if err := os.MkdirAll(blobsDir, 0750); err != nil { diff --git a/internal/server/server_test.go b/internal/server/server_test.go index 49a40d8..2723af5 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -2,6 +2,7 @@ package server import ( "bytes" + "database/sql" "encoding/json" "net/http" "net/http/httptest" @@ -9,6 +10,8 @@ import ( "path/filepath" "testing" "time" + + "golang.org/x/crypto/bcrypt" ) func TestNewServer(t *testing.T) { @@ -67,10 +70,11 @@ func TestSyncPushPullStoresSequencedOps(t *testing.T) { defer s.Close() s.SetupRoutes() + insertSyncUser(t, s, "user-a") now := time.Now().UTC().Format(time.RFC3339) if _, err := s.db.Exec( - "INSERT INTO server_devices (id, name, api_key, last_seen, created_at) VALUES (?, ?, ?, ?, ?)", - "device-a", "Device A", "api-key", now, now, + "INSERT INTO server_devices (id, name, api_key, user_id, vault_id, last_seen, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)", + "device-a", "Device A", "api-key", "user-a", "vault-a", now, now, ); err != nil { t.Fatalf("insert device: %v", err) } @@ -167,6 +171,343 @@ func TestRevokedLegacyAPIKeyCannotPushOrPull(t *testing.T) { } } +func TestSyncPullDoesNotExposeOtherUserOperations(t *testing.T) { + s, ts := newSyncHTTPServer(t) + defer s.Close() + defer ts.Close() + + insertSyncUser(t, s, "user-a") + insertSyncUser(t, s, "user-b") + insertSyncDevice(t, s, "device-a", "user-a", "token-a") + insertSyncDevice(t, s, "device-b", "user-b", "token-b") + + postJSON(t, ts.URL+"/api/v1/sync/push", "token-a", syncPushBody("device-a", "op-user-a", "")) + + pull := postJSON(t, ts.URL+"/api/v1/sync/pull", "token-b", map[string]interface{}{ + "since_sequence": 0, + }) + if got := len(pull["ops"].([]interface{})); got != 0 { + t.Fatalf("other user pulled %d operation(s), want 0: %#v", got, pull) + } + if got := int(pull["server_sequence"].(float64)); got != 0 { + t.Fatalf("other user cursor = %d, want 0: %#v", got, pull) + } +} + +func TestSyncPushUsesAuthenticatedDeviceIdentity(t *testing.T) { + s, ts := newSyncHTTPServer(t) + defer s.Close() + defer ts.Close() + + insertSyncUser(t, s, "user-a") + insertSyncUser(t, s, "user-b") + insertSyncDevice(t, s, "device-authenticated", "user-a", "token-a") + insertSyncDevice(t, s, "device-forged", "user-b", "token-b") + + postJSON(t, ts.URL+"/api/v1/sync/push", "token-a", syncPushBody("device-forged", "op-auth-device", "")) + + var storedDeviceID string + if err := s.db.QueryRow("SELECT device_id FROM server_ops WHERE op_id=?", "op-auth-device").Scan(&storedDeviceID); err != nil { + t.Fatalf("read stored operation: %v", err) + } + if storedDeviceID != "device-authenticated" { + t.Fatalf("stored device = %q, want authenticated device", storedDeviceID) + } +} + +func TestSyncPushScopesIdempotencyResponsesByUser(t *testing.T) { + s, ts := newSyncHTTPServer(t) + defer s.Close() + defer ts.Close() + + insertSyncUser(t, s, "user-a") + insertSyncUser(t, s, "user-b") + insertSyncDevice(t, s, "device-a", "user-a", "token-a") + insertSyncDevice(t, s, "device-b", "user-b", "token-b") + + postJSON(t, ts.URL+"/api/v1/sync/push", "token-a", syncPushBody("device-a", "op-user-a", "same-key")) + response := postJSON(t, ts.URL+"/api/v1/sync/push", "token-b", syncPushBody("device-b", "op-user-b", "same-key")) + + accepted := response["accepted"].([]interface{}) + if len(accepted) != 1 || accepted[0] != "op-user-b" { + t.Fatalf("user-b received idempotency response %#v, want only op-user-b", response) + } +} + +func TestSyncPushDoesNotReportOtherTenantConflicts(t *testing.T) { + s, ts := newSyncHTTPServer(t) + defer s.Close() + defer ts.Close() + + insertSyncUser(t, s, "user-a") + insertSyncUser(t, s, "user-b") + insertSyncDevice(t, s, "device-a", "user-a", "token-a") + insertSyncDevice(t, s, "device-b", "user-b", "token-b") + + postJSON(t, ts.URL+"/api/v1/sync/push", "token-a", syncPushBody("device-a", "op-user-a-1", "")) + postJSON(t, ts.URL+"/api/v1/sync/push", "token-a", syncPushBody("device-a", "op-user-a-2", "")) + + body := syncPushBody("device-b", "op-user-b", "") + body["ops"].([]map[string]interface{})[0]["last_seen_server_seq"] = 1 + response := postJSON(t, ts.URL+"/api/v1/sync/push", "token-b", body) + if response["conflicts"] != nil { + t.Fatalf("other tenant conflict leaked into response: %#v", response) + } +} + +func TestSyncPullDoesNotExposeOtherVaultOperations(t *testing.T) { + s, ts := newSyncHTTPServer(t) + defer s.Close() + defer ts.Close() + + password := "correct horse battery staple" + insertPairableUser(t, s, "user-a", "alice", password) + + deviceA := pairSyncDevice(t, ts.URL, "alice", password, "vault-a") + deviceB := pairSyncDevice(t, ts.URL, "alice", password, "vault-b") + + postJSON(t, ts.URL+"/api/v1/sync/push", deviceA.token, syncPushBody(deviceA.id, "op-vault-a", "")) + postJSON(t, ts.URL+"/api/v1/sync/push", deviceB.token, syncPushBody(deviceB.id, "op-vault-b", "")) + + pull := postJSON(t, ts.URL+"/api/v1/sync/pull", deviceA.token, map[string]interface{}{ + "since_sequence": 0, + }) + ops := pull["ops"].([]interface{}) + if len(ops) != 1 || ops[0].(map[string]interface{})["op_id"] != "op-vault-a" { + t.Fatalf("vault-a pulled %#v, want only op-vault-a", pull) + } +} + +func TestClientPairRequiresVaultID(t *testing.T) { + s, ts := newSyncHTTPServer(t) + defer s.Close() + defer ts.Close() + + password := "correct horse battery staple" + insertPairableUser(t, s, "user-a", "alice", password) + + status, response := postJSONStatus(t, ts.URL+"/api/client/pair", "", map[string]interface{}{ + "login": "alice", + "password": password, + "device_name": "Desktop", + }) + if status != http.StatusBadRequest || response["error"] != "vault_id required" { + t.Fatalf("pair without vault ID status=%d response=%#v, want 400 vault_id required", status, response) + } +} + +func TestClientPairRejectsReservedLegacyVaultID(t *testing.T) { + s, ts := newSyncHTTPServer(t) + defer s.Close() + defer ts.Close() + + password := "correct horse battery staple" + insertPairableUser(t, s, "user-a", "alice", password) + + status, response := postJSONStatus(t, ts.URL+"/api/client/pair", "", map[string]interface{}{ + "login": "alice", + "password": password, + "device_name": "Desktop", + "vault_id": "legacy:user-a", + }) + if status != http.StatusBadRequest || response["error"] != "vault_id uses reserved prefix" { + t.Fatalf("pair with reserved vault ID status=%d response=%#v, want 400 reserved prefix", status, response) + } +} + +func TestDeviceRegisterRequiresValidVaultID(t *testing.T) { + s, ts := newSyncHTTPServer(t) + defer s.Close() + defer ts.Close() + + password := "correct horse battery staple" + insertPairableUser(t, s, "user-a", "alice", password) + + status, response := postJSONStatus(t, ts.URL+"/api/v1/device/register", "", map[string]interface{}{ + "name": "Desktop", + "username": "alice", + "password": password, + }) + if status != http.StatusBadRequest || response["error"] != "vault_id required" { + t.Fatalf("register without vault ID status=%d response=%#v, want 400 vault_id required", status, response) + } + + status, response = postJSONStatus(t, ts.URL+"/api/v1/device/register", "", map[string]interface{}{ + "name": "Desktop", + "username": "alice", + "password": password, + "vault_id": "legacy:user-a", + }) + if status != http.StatusBadRequest || response["error"] != "vault_id uses reserved prefix" { + t.Fatalf("register with reserved vault ID status=%d response=%#v, want 400 reserved prefix", status, response) + } +} + +func TestNewServerMigratesLegacyOperationScope(t *testing.T) { + dir := t.TempDir() + dbPath := filepath.Join(dir, "legacy.db") + db, err := sql.Open("sqlite3", dbPath) + if err != nil { + t.Fatalf("open legacy db: %v", err) + } + + legacySchema := []string{ + `CREATE TABLE server_devices ( + id TEXT PRIMARY KEY, name TEXT NOT NULL, api_key TEXT NOT NULL UNIQUE, + token_hash TEXT, token_prefix TEXT, token_suffix TEXT, user_id TEXT, + client_version TEXT, last_ip TEXT, last_seen TEXT, revoked_at TEXT, + created_at TEXT NOT NULL + )`, + `CREATE TABLE server_user_devices ( + user_id TEXT NOT NULL, device_id TEXT NOT NULL, + PRIMARY KEY (user_id, device_id) + )`, + `CREATE TABLE server_ops ( + op_id TEXT PRIMARY KEY, server_sequence INTEGER, device_id TEXT NOT NULL, + entity_type TEXT NOT NULL, entity_id TEXT NOT NULL, op_type TEXT NOT NULL, + payload_json TEXT NOT NULL, idempotency_key TEXT, client_sequence INTEGER DEFAULT 0, + last_seen_server_seq INTEGER DEFAULT 0, created_at TEXT NOT NULL, + pushed_at TEXT NOT NULL + )`, + `CREATE TABLE server_tombstones ( + entity_type TEXT NOT NULL, entity_id TEXT NOT NULL, op_id TEXT NOT NULL, + deleted_at TEXT NOT NULL, PRIMARY KEY (entity_type, entity_id) + )`, + `CREATE TABLE server_idempotency_keys ( + idempotency_key TEXT PRIMARY KEY, response_json TEXT NOT NULL, created_at TEXT NOT NULL + )`, + } + for _, stmt := range legacySchema { + if _, err := db.Exec(stmt); err != nil { + t.Fatalf("create legacy schema: %v", err) + } + } + now := time.Now().UTC().Format(time.RFC3339) + if _, err := db.Exec(`INSERT INTO server_devices (id, name, api_key, last_seen, created_at) + VALUES (?, ?, ?, ?, ?)`, "legacy-device", "Legacy", "legacy-key", now, now); err != nil { + t.Fatalf("insert legacy device: %v", err) + } + if _, err := db.Exec(`INSERT INTO server_user_devices (user_id, device_id) VALUES (?, ?)`, "legacy-user", "legacy-device"); err != nil { + t.Fatalf("insert legacy device owner: %v", err) + } + if _, err := db.Exec(`INSERT INTO server_ops + (op_id, server_sequence, device_id, entity_type, entity_id, op_type, payload_json, created_at, pushed_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, "legacy-op", 1, "legacy-device", "file", "Docs/one.txt", "create", `{}`, now, now); err != nil { + t.Fatalf("insert legacy operation: %v", err) + } + if err := db.Close(); err != nil { + t.Fatalf("close legacy db: %v", err) + } + + s, err := NewServer(dbPath, filepath.Join(dir, "data"), &Config{Port: 47732}) + if err != nil { + t.Fatalf("NewServer migration: %v", err) + } + assertLegacyOperationScope(t, s) + if err := s.Close(); err != nil { + t.Fatalf("close migrated db: %v", err) + } + + s, err = NewServer(dbPath, filepath.Join(dir, "data"), &Config{Port: 47732}) + if err != nil { + t.Fatalf("NewServer repeated migration: %v", err) + } + defer s.Close() + assertLegacyOperationScope(t, s) +} + +func assertLegacyOperationScope(t *testing.T, s *Server) { + t.Helper() + var userID, vaultID string + if err := s.db.QueryRow("SELECT user_id, vault_id FROM server_ops WHERE op_id=?", "legacy-op").Scan(&userID, &vaultID); err != nil { + t.Fatalf("read migrated operation: %v", err) + } + if userID != "legacy-user" || vaultID != "legacy:legacy-user" { + t.Fatalf("migrated scope = %q/%q, want legacy-user/legacy:legacy-user", userID, vaultID) + } +} + +func newSyncHTTPServer(t *testing.T) (*Server, *httptest.Server) { + t.Helper() + dir := t.TempDir() + s, err := NewServer(filepath.Join(dir, "test.db"), filepath.Join(dir, "data"), &Config{Port: 47732}) + if err != nil { + t.Fatalf("NewServer: %v", err) + } + s.SetupRoutes() + return s, httptest.NewServer(s.mux) +} + +func insertSyncUser(t *testing.T, s *Server, userID string) { + t.Helper() + now := time.Now().UTC().Format(time.RFC3339) + if _, err := s.db.Exec(`INSERT INTO server_users + (id, username, email, password_hash, confirmed, created_at) + VALUES (?, ?, ?, ?, 1, ?)`, userID, userID, userID+"@example.test", "unused", now); err != nil { + t.Fatalf("insert sync user %s: %v", userID, err) + } +} + +func insertPairableUser(t *testing.T, s *Server, userID, username, password string) { + t.Helper() + passwordHash, err := bcrypt.GenerateFromPassword([]byte(password), bcrypt.MinCost) + if err != nil { + t.Fatalf("hash password: %v", err) + } + now := time.Now().UTC().Format(time.RFC3339) + if _, err := s.db.Exec(`INSERT INTO server_users + (id, username, email, password_hash, confirmed, created_at) + VALUES (?, ?, ?, ?, 1, ?)`, userID, username, username+"@example.test", string(passwordHash), now); err != nil { + t.Fatalf("insert pairable user: %v", err) + } +} + +func insertSyncDevice(t *testing.T, s *Server, deviceID, userID, token string) { + t.Helper() + now := time.Now().UTC().Format(time.RFC3339) + if _, err := s.db.Exec(`INSERT INTO server_devices + (id, name, api_key, token_hash, user_id, last_seen, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?)`, deviceID, deviceID, "legacy-"+deviceID, sha256Hex(token), userID, now, now); err != nil { + t.Fatalf("insert sync device %s: %v", deviceID, err) + } +} + +func syncPushBody(deviceID, opID, idempotencyKey string) map[string]interface{} { + return map[string]interface{}{ + "device_id": deviceID, + "idempotency_key": idempotencyKey, + "ops": []map[string]interface{}{ + { + "op_id": opID, + "entity_type": "file", + "entity_id": "Docs/one.txt", + "op_type": "create", + "payload_json": `{"path":"Docs/one.txt","content":"hello"}`, + "created_at": "2026-07-10T00:00:00Z", + }, + }, + } +} + +type pairedSyncDevice struct { + id string + token string +} + +func pairSyncDevice(t *testing.T, serverURL, username, password, vaultID string) pairedSyncDevice { + t.Helper() + response := postJSON(t, serverURL+"/api/client/pair", "", map[string]interface{}{ + "login": username, + "password": password, + "device_name": "Desktop " + vaultID, + "vault_id": vaultID, + }) + return pairedSyncDevice{ + id: response["device_id"].(string), + token: response["device_token"].(string), + } +} + func postJSON(t *testing.T, url, token string, body interface{}) map[string]interface{} { t.Helper() status, out := postJSONStatus(t, url, token, body)