// Package store keeps the state of the web UI in SQLite: clusters (root // passwords encrypted with AES-256-GCM, pinned TLS certificates and SSH host // keys), synchronization settings, per target mappings and policies, users, // execution journal and audit trail. package store import ( "context" "database/sql" "encoding/json" "errors" "fmt" "log" "strings" "time" _ "modernc.org/sqlite" // pure Go driver: the binary stays static (CGO_ENABLED=0) "pve-confsync/internal/adapter" "pve-confsync/internal/config" "pve-confsync/internal/model" "pve-confsync/internal/sshx" ) // ErrNotFound is returned for an unknown cluster, user or run. var ErrNotFound = errors.New("not found") // Store is the SQLite database. type Store struct { db *sql.DB box *Box } // schemaVersion 2: clusters reached with the root credentials (API + SSH), // roles primary / target / disabled. Version 1 (API tokens) is not migrated: // its clusters table is dropped, the clusters are added again from the UI. const schemaVersion = 2 const schema = ` CREATE TABLE IF NOT EXISTS clusters ( name TEXT PRIMARY KEY, role TEXT NOT NULL DEFAULT 'target', position INTEGER NOT NULL DEFAULT 0, url TEXT NOT NULL, api_user TEXT NOT NULL DEFAULT 'root@pam', password BLOB, tls TEXT NOT NULL DEFAULT 'fp', fingerprint TEXT NOT NULL DEFAULT '', ca_pem TEXT NOT NULL DEFAULT '', timeout TEXT NOT NULL DEFAULT '', ssh_host TEXT NOT NULL DEFAULT '', ssh_port INTEGER NOT NULL DEFAULT 22, ssh_hostkey TEXT NOT NULL DEFAULT '', mapping TEXT NOT NULL DEFAULT '{}', policies TEXT NOT NULL DEFAULT '{}', last_check TEXT NOT NULL DEFAULT '', updated_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, value TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS users ( name TEXT PRIMARY KEY, role TEXT NOT NULL, password TEXT NOT NULL, epoch INTEGER NOT NULL, created_at INTEGER NOT NULL, last_login INTEGER NOT NULL DEFAULT 0 ); CREATE TABLE IF NOT EXISTS runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, started_at INTEGER NOT NULL, finished_at INTEGER NOT NULL DEFAULT 0, user TEXT NOT NULL, mode TEXT NOT NULL, scope TEXT NOT NULL, clients TEXT NOT NULL DEFAULT '', objects TEXT NOT NULL DEFAULT '', status TEXT NOT NULL, summary TEXT NOT NULL DEFAULT '', detail TEXT NOT NULL DEFAULT '{}' ); CREATE INDEX IF NOT EXISTS runs_started ON runs(started_at); CREATE TABLE IF NOT EXISTS audit ( id INTEGER PRIMARY KEY AUTOINCREMENT, at INTEGER NOT NULL, user TEXT NOT NULL, role TEXT NOT NULL DEFAULT '', ip TEXT NOT NULL DEFAULT '', action TEXT NOT NULL, target TEXT NOT NULL DEFAULT '', ok INTEGER NOT NULL DEFAULT 1, detail TEXT NOT NULL DEFAULT '' ); CREATE INDEX IF NOT EXISTS audit_at ON audit(at); ` // Open opens (and creates) the database. key encrypts the token secrets. func Open(path string, key []byte) (*Store, error) { box, err := NewBox(key) if err != nil { return nil, err } // WAL: the UI reads while a synchronization writes the journal. db, err := sql.Open("sqlite", "file:"+path+"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_pragma=foreign_keys(1)") if err != nil { return nil, err } db.SetMaxOpenConns(1) // one writer; SQLite serializes anyway if err := migrate(db); err != nil { db.Close() return nil, fmt.Errorf("%s: %w", path, err) } if _, err := db.Exec(schema + nodeSchema); err != nil { db.Close() return nil, fmt.Errorf("%s: %w", path, err) } s := &Store{db: db, box: box} if err := s.checkKey(); err != nil { db.Close() return nil, err } return s, nil } // migrate drops the clusters of the first schema (API tokens, not reusable). func migrate(db *sql.DB) error { var v int if err := db.QueryRow(`PRAGMA user_version`).Scan(&v); err != nil { return err } if v >= schemaVersion { return nil } var old int if err := db.QueryRow(`SELECT count(*) FROM pragma_table_info('clusters') WHERE name = 'token_id'`).Scan(&old); err != nil { return err } if old > 0 { log.Printf("database: clusters of the previous version (API tokens) dropped, add them again with their root credentials") if _, err := db.Exec(`DROP TABLE clusters; DROP TABLE IF EXISTS settings`); err != nil { return err } } _, err := db.Exec(fmt.Sprintf(`PRAGMA user_version = %d`, schemaVersion)) return err } // Close closes the database. func (s *Store) Close() error { return s.db.Close() } // checkKey refuses to start with a key that does not decrypt the stored secrets // (lost or changed CONFSYNC_SECRET_KEY), instead of failing later on each cluster. func (s *Store) checkKey() error { var blob []byte err := s.db.QueryRow(`SELECT password FROM clusters WHERE password IS NOT NULL AND length(password) > 0 LIMIT 1`).Scan(&blob) if errors.Is(err, sql.ErrNoRows) { return nil } if err != nil { return err } if _, err := s.box.Open(blob); err != nil { return errors.New("the encryption key does not decrypt the stored passwords (CONFSYNC_SECRET_KEY changed?)") } return nil } // ---- clusters ---- // Cluster roles: one primary (the source, never written), targets (replicated) // and disabled clusters (kept, not read). const ( ClusterPrimary = "primary" ClusterTarget = "target" ClusterDisabled = "disabled" ) // TLS modes of a cluster. const ( TLSPin = "fp" // pinned certificate (recorded on the first connection when empty) TLSCA = "ca" // CA bundle TLSSystem = "sys" // system authorities TLSInsecure = "insecure" // no verification (lab only) ) // Check is the result of the last connection test of a cluster. type Check struct { At time.Time `json:"at"` OK bool `json:"ok"` Error string `json:"error,omitempty"` Warnings []string `json:"warnings,omitempty"` Version string `json:"version,omitempty"` Nodes []string `json:"nodes,omitempty"` API bool `json:"api"` SSH bool `json:"ssh"` } // Cluster is a cluster as stored. The password is never returned to the UI: // HasPassword only says whether one is set. type Cluster struct { Name string `json:"name"` Role string `json:"role"` Position int `json:"position"` URL string `json:"url"` User string `json:"user"` Password string `json:"password,omitempty"` // write only HasPassword bool `json:"hasPassword"` TLS string `json:"tls"` Fingerprint string `json:"fingerprint"` CAPEM string `json:"caPem"` Timeout string `json:"timeout"` SSHHost string `json:"sshHost"` SSHPort int `json:"sshPort"` SSHHostKey string `json:"sshHostKey"` // read only (pinned by the connection test) SSHFingerprint string `json:"sshFingerprint"` // read only Mapping map[model.Kind]map[string]map[string]any `json:"mapping"` Policies map[model.Kind]map[string]adapter.Policy `json:"policies"` LastCheck *Check `json:"lastCheck,omitempty"` UpdatedAt time.Time `json:"updatedAt"` } const clusterCols = `name, role, position, url, api_user, password, tls, fingerprint, ca_pem, timeout, ssh_host, ssh_port, ssh_hostkey, mapping, policies, last_check, updated_at` func scanCluster(scan func(...any) error) (Cluster, []byte, error) { var c Cluster var blob []byte var mapping, policies, check string var updated int64 err := scan(&c.Name, &c.Role, &c.Position, &c.URL, &c.User, &blob, &c.TLS, &c.Fingerprint, &c.CAPEM, &c.Timeout, &c.SSHHost, &c.SSHPort, &c.SSHHostKey, &mapping, &policies, &check, &updated) if err != nil { return c, nil, err } c.HasPassword = len(blob) > 0 c.SSHFingerprint = sshx.Fingerprint(c.SSHHostKey) c.UpdatedAt = time.Unix(updated, 0).UTC() _ = json.Unmarshal([]byte(mapping), &c.Mapping) _ = json.Unmarshal([]byte(policies), &c.Policies) if check != "" { c.LastCheck = &Check{} _ = json.Unmarshal([]byte(check), c.LastCheck) } return c, blob, nil } // Clusters returns every cluster, primary first, without passwords. func (s *Store) Clusters(ctx context.Context) ([]Cluster, error) { rows, err := s.db.QueryContext(ctx, `SELECT `+clusterCols+` FROM clusters ORDER BY role = 'primary' DESC, position, name`) if err != nil { return nil, err } defer rows.Close() var out []Cluster for rows.Next() { c, _, err := scanCluster(rows.Scan) if err != nil { return nil, err } out = append(out, c) } return out, rows.Err() } // Cluster returns one cluster with its decrypted connection (for a test). func (s *Store) Cluster(ctx context.Context, name string) (Cluster, config.Cluster, error) { c, blob, err := scanCluster(s.db.QueryRowContext(ctx, `SELECT `+clusterCols+` FROM clusters WHERE name = ?`, name).Scan) if errors.Is(err, sql.ErrNoRows) { return c, config.Cluster{}, ErrNotFound } if err != nil { return c, config.Cluster{}, err } cc, err := s.connection(c, blob) return c, cc, err } func (s *Store) connection(c Cluster, blob []byte) (config.Cluster, error) { pass, err := s.box.Open(blob) if err != nil { return config.Cluster{}, fmt.Errorf("cluster %s: password: %w", c.Name, err) } cc := config.Cluster{Name: c.Name, URL: config.NormalizeURL(c.URL), User: c.User, Password: string(pass), Timeout: c.Timeout, SSHHost: c.SSHHost, SSHPort: c.SSHPort, SSHHostKey: c.SSHHostKey} switch c.TLS { case TLSPin: cc.Fingerprint = c.Fingerprint case TLSCA: cc.CAPEM = c.CAPEM case TLSInsecure: cc.Insecure = true } return cc, nil } // ValidClusterRole tells whether r is a cluster role. func ValidClusterRole(r string) bool { return r == ClusterPrimary || r == ClusterTarget || r == ClusterDisabled } // SaveCluster creates or updates a cluster. An empty Password keeps the stored // one. oldName renames a cluster (empty: same name). A primary cluster makes // the previous primary a target. The pinned TLS certificate and SSH host key // are forgotten when the address they belong to changes. func (s *Store) SaveCluster(ctx context.Context, oldName string, c Cluster) error { c.Name = strings.TrimSpace(c.Name) c.URL = config.NormalizeURL(c.URL) if c.Name == "" || c.URL == "" { return errors.New("name and url are required") } if c.User == "" { c.User = "root@pam" } if c.Role == "" { c.Role = ClusterTarget } if !ValidClusterRole(c.Role) { return fmt.Errorf("role %q: primary, target or disabled expected", c.Role) } switch c.TLS { case "": c.TLS = TLSPin case TLSPin, TLSCA, TLSSystem, TLSInsecure: default: return fmt.Errorf("tls %q: fp, ca, sys or insecure expected", c.TLS) } if c.TLS == TLSCA && strings.TrimSpace(c.CAPEM) == "" { return errors.New("the CA certificate (PEM) is required") } if c.SSHPort == 0 { c.SSHPort = 22 } if oldName == "" { oldName = c.Name } mapping, _ := json.Marshal(nonNil(c.Mapping)) policies, _ := json.Marshal(nonNilP(c.Policies)) tx, err := s.db.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback() old, blob, err := scanCluster(tx.QueryRowContext(ctx, `SELECT `+clusterCols+` FROM clusters WHERE name = ?`, oldName).Scan) exists := err == nil if err != nil && !errors.Is(err, sql.ErrNoRows) { return err } if c.Password != "" { if blob, err = s.box.Seal([]byte(c.Password)); err != nil { return err } } if len(blob) == 0 { return errors.New("the root password is required") } hostKey := "" if exists { hostKey = old.SSHHostKey if old.URL != c.URL && c.Fingerprint == old.Fingerprint { c.Fingerprint = "" // another server: pinned again on the next test } if sshTarget(old) != sshTarget(c) { hostKey = "" } } if c.Role == ClusterPrimary { if _, err := tx.ExecContext(ctx, `UPDATE clusters SET role = 'target' WHERE role = 'primary'`); err != nil { return err } } now := time.Now().Unix() if exists { _, err = tx.ExecContext(ctx, `UPDATE clusters SET name=?, role=?, position=?, url=?, api_user=?, password=?, tls=?, fingerprint=?, ca_pem=?, timeout=?, ssh_host=?, ssh_port=?, ssh_hostkey=?, mapping=?, policies=?, updated_at=? WHERE name=?`, c.Name, c.Role, c.Position, c.URL, c.User, blob, c.TLS, c.Fingerprint, c.CAPEM, c.Timeout, c.SSHHost, c.SSHPort, hostKey, string(mapping), string(policies), now, oldName) } else { _, err = tx.ExecContext(ctx, `INSERT INTO clusters (name, role, position, url, api_user, password, tls, fingerprint, ca_pem, timeout, ssh_host, ssh_port, ssh_hostkey, mapping, policies, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, c.Name, c.Role, c.Position, c.URL, c.User, blob, c.TLS, c.Fingerprint, c.CAPEM, c.Timeout, c.SSHHost, c.SSHPort, "", string(mapping), string(policies), now) } if err != nil { if strings.Contains(err.Error(), "UNIQUE") { return fmt.Errorf("cluster %q already exists", c.Name) } return err } if exists && oldName != c.Name { if err := renameClusterNodes(ctx, tx, oldName, c.Name); err != nil { return err } } return tx.Commit() } // sshTarget is the SSH endpoint of a cluster, to detect a change of node. func sshTarget(c Cluster) string { host := c.SSHHost if host == "" { host = c.URL } return fmt.Sprintf("%s|%d", host, c.SSHPort) } // SetRole changes the role of a cluster (a new primary demotes the previous one). func (s *Store) SetRole(ctx context.Context, name, role string) error { if !ValidClusterRole(role) { return fmt.Errorf("role %q: primary, target or disabled expected", role) } tx, err := s.db.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback() if role == ClusterPrimary { if _, err := tx.ExecContext(ctx, `UPDATE clusters SET role = 'target' WHERE role = 'primary' AND name <> ?`, name); err != nil { return err } } res, err := tx.ExecContext(ctx, `UPDATE clusters SET role = ?, updated_at = ? WHERE name = ?`, role, time.Now().Unix(), name) if err != nil { return err } if n, _ := res.RowsAffected(); n == 0 { return ErrNotFound } return tx.Commit() } // Pin records the TLS fingerprint and/or the SSH host key seen on the first // connection (empty values are left unchanged), and the result of the test. func (s *Store) Pin(ctx context.Context, name, fingerprint, hostKey string, check *Check) error { var chk []byte if check != nil { chk, _ = json.Marshal(check) } _, err := s.db.ExecContext(ctx, `UPDATE clusters SET fingerprint = CASE WHEN ? <> '' THEN ? ELSE fingerprint END, ssh_hostkey = CASE WHEN ? <> '' THEN ? ELSE ssh_hostkey END, last_check = CASE WHEN ? <> '' THEN ? ELSE last_check END WHERE name = ?`, fingerprint, fingerprint, hostKey, hostKey, string(chk), string(chk), name) return err } // Forget clears the pinned TLS fingerprint and SSH host key (pinned again on the next test). func (s *Store) Forget(ctx context.Context, name string) error { res, err := s.db.ExecContext(ctx, `UPDATE clusters SET fingerprint = CASE WHEN tls = 'fp' THEN '' ELSE fingerprint END, ssh_hostkey = '' WHERE name = ?`, name) if err != nil { return err } if n, _ := res.RowsAffected(); n == 0 { return ErrNotFound } return nil } // DeleteCluster removes a cluster. func (s *Store) DeleteCluster(ctx context.Context, name string) error { res, err := s.db.ExecContext(ctx, `DELETE FROM clusters WHERE name = ?`, name) if err != nil { return err } if n, _ := res.RowsAffected(); n == 0 { return ErrNotFound } _, err = s.db.ExecContext(ctx, `DELETE FROM nodes WHERE cluster = ?`, name) return err } func nonNil(m map[model.Kind]map[string]map[string]any) any { if m == nil { return map[string]any{} } return m } func nonNilP(m map[model.Kind]map[string]adapter.Policy) any { if m == nil { return map[string]any{} } return m } // ---- settings ---- // Settings are the global synchronization settings (the rest of the YAML file). type Settings struct { Kinds []model.Kind `json:"kinds"` Objects map[model.Kind]config.Selection `json:"objects"` Policies map[model.Kind]map[string]adapter.Policy `json:"policies"` Prune bool `json:"prune"` } // Settings returns the global settings (all kinds by default). func (s *Store) Settings(ctx context.Context) (Settings, error) { var st Settings var raw string err := s.db.QueryRowContext(ctx, `SELECT value FROM settings WHERE key = 'sync'`).Scan(&raw) switch { case errors.Is(err, sql.ErrNoRows): case err != nil: return st, err default: if err := json.Unmarshal([]byte(raw), &st); err != nil { return st, err } } if len(st.Kinds) == 0 { st.Kinds = append(st.Kinds, model.Kinds...) } // empty maps, not null, for the web UI if st.Objects == nil { st.Objects = map[model.Kind]config.Selection{} } if st.Policies == nil { st.Policies = map[model.Kind]map[string]adapter.Policy{} } return st, nil } // SaveSettings replaces the global settings. func (s *Store) SaveSettings(ctx context.Context, st Settings) error { b, err := json.Marshal(st) if err != nil { return err } _, err = s.db.ExecContext(ctx, `INSERT INTO settings (key, value) VALUES ('sync', ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value`, string(b)) return err } // ---- configuration for the engine ---- // Config builds the engine configuration, passwords decrypted: the primary // cluster and the targets (disabled clusters are left aside). It fails when no // primary is defined. func (s *Store) Config(ctx context.Context) (*config.Config, error) { st, err := s.Settings(ctx) if err != nil { return nil, err } cfg := &config.Config{Kinds: st.Kinds, Objects: st.Objects, Policies: st.Policies, Prune: st.Prune} rows, err := s.db.QueryContext(ctx, `SELECT `+clusterCols+` FROM clusters ORDER BY position, name`) if err != nil { return nil, err } defer rows.Close() hasRef := false for rows.Next() { c, blob, err := scanCluster(rows.Scan) if err != nil { return nil, err } if c.Role == ClusterDisabled { continue } cc, err := s.connection(c, blob) if err != nil { return nil, err } if c.Role == ClusterPrimary { cfg.Reference, hasRef = cc, true continue } cfg.Clients = append(cfg.Clients, config.Client{Cluster: cc, Mapping: c.Mapping, Policies: c.Policies}) } if err := rows.Err(); err != nil { return nil, err } if !hasRef { return nil, ErrNoReference } return cfg, cfg.Validate() } // ErrNoReference: nothing can be compared before a primary cluster is chosen. var ErrNoReference = errors.New("no primary cluster defined")