verstak/verstak-sync-server
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: isolate sync operations by user and vault

Browse Source
This commit is contained in:
mirivlad
2026-07-10 03:13:17 +08:00
parent 5c8644d5e1
commit 086bc4ea4a
8 changed files with 712 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/server/handlers_api.go
+46 -21
View File
@@ -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
internal/server/middleware.go
+55 -27
View File
@@ -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 {
internal/server/schema.go
+210 -3
View File
@@ -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, &notNull, &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
}
internal/server/server.go
+4
View File
@@ -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 {
internal/server/server_test.go
+343 -2
View File
@@ -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)