mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
One derived key, no wrapping
tasks generated a DEK per shard, wrapped it under a KEK derived from the master, and stored the wrapped blob beside the database — so it also needed an upgrade path to rewrap, a principal vocabulary to derive against, and tests for all of it. hanzoai/cek derives the key from the master and the namespace instead. Nothing is stored, so there is nothing to wrap, rewrap, or upgrade. dek.go, principal.go and upgrade.go are gone with the tests that pinned them: they proved properties of a scheme that no longer exists, not properties the store still has to hold. That also unpins hanzoai/sqlite, which deleted the DEK API in v0.5.0. tasks now builds against it, which is what proves the migration is complete rather than merely compiling. Databases written under the old scheme will not open under the new derivation. There are no customers and no data to preserve. Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
Hanzo Dev
@@ -410,56 +410,72 @@ Persistence = per-`(principal, namespace)` SQLite shard
|
||||
`meta` tables. Survives restart (`TestStore_PersistsAcrossOpen`).
|
||||
|
||||
A `Principal` is the tenant that owns a shard — an org, optionally
|
||||
narrowed to a project and a user. It is one value used in the two places
|
||||
tenancy is decided, so they cannot disagree: the shard's directory
|
||||
(`<data>/<org>/<project>/<user>/<ns>.db`, each unset leg written `_`) and
|
||||
the key its DEK is wrapped under. Shards and namespaces are created on
|
||||
first use; nothing is declared up front.
|
||||
narrowed to a project. Shards and namespaces are created on first use;
|
||||
nothing is declared up front.
|
||||
|
||||
### Storage and encryption — `hanzoai/cek` + `hanzoai/namespace`
|
||||
|
||||
Tasks no longer has a storage layout or a key schedule of its own. It
|
||||
uses the estate's:
|
||||
|
||||
namespace names the tenant, and where its file lives
|
||||
cek turns the process master + that name into the file's key,
|
||||
and opens it
|
||||
sqlite opens a file under a raw key, knowing nothing about who owns it
|
||||
kms holds the master
|
||||
|
||||
<data>/orgs/<org>/<ns>.db
|
||||
<data>/orgs/<org>/projects/<project>/<ns>.db
|
||||
<data>/orgs/_platform/<ns>.db the root (unscoped) tenant
|
||||
|
||||
The key is DERIVED (HKDF of master + namespace + namespace-name), not
|
||||
generated, wrapped or stored. So there is no DEK, no `.dek` sidecar, no
|
||||
unwrap step, no rewrap step, no per-file key material to lose, and no
|
||||
migration path to maintain. A shard is born encrypted or it does not
|
||||
exist; losing the master loses the data, which is the property you want
|
||||
from encryption at rest and the reason the master lives in KMS.
|
||||
|
||||
`EmbedConfig.MasterKey` (32 bytes, `KMS_MASTER_KEY` base64 for `tasksd`)
|
||||
opens every shard encrypted at rest. Each file carries its own DEK,
|
||||
wrapped under a KEK derived master -> org -> project -> user and stored
|
||||
beside it as `<ns>.db.dek`, so rotating the master rewraps sidecars and
|
||||
never touches ciphertext. A keyed shard is ciphertext on every build:
|
||||
`hanzoai/sqlite` uses the live libsqlcipher codec when one is linked and
|
||||
its pure-Go SQLCipher envelope when one is not. On the envelope the file
|
||||
is sealed at checkpoint/close rather than per commit, which is what the
|
||||
manager's sweep bounds. Unset leaves shards plaintext (dev).
|
||||
is installed into cek at `store.New`. A **nil master no longer means
|
||||
plaintext** — there is no such posture. It means "I have no key to
|
||||
state": embedded in a process that already installed one (cloud resolves
|
||||
it through `credz` at boot) that key is used, so one deployment has one
|
||||
key; otherwise a random master is minted for this process and nothing it
|
||||
writes outlives the run, which is the honest shape of a keyless
|
||||
deployment and why a developer needs no configuration.
|
||||
|
||||
### Upgrading a store written before principals — `tasksd upgrade`
|
||||
A shard is ciphertext on every build: `hanzoai/sqlite` uses the live
|
||||
libsqlcipher codec when one is linked and its pure-Go SQLCipher envelope
|
||||
when one is not. On the envelope the file is written when it is SEALED
|
||||
(checkpoint/close), not per commit — which is what the manager's sweep
|
||||
bounds, and why `ListShards`/`ListPrincipals` read the resident set as
|
||||
well as the directory: a shard created moments ago has no file yet, and
|
||||
that is exactly the tenant the cron sweeper most needs to see.
|
||||
|
||||
An older binary addressed a shard by org alone (`<data>/<org>/<ns>.db`,
|
||||
`<data>/_/<ns>.db` for the unscoped tenant). This one reads three levels
|
||||
down, so on the first boot after the upgrade it finds nothing, creates
|
||||
empty shards beside the full ones, enumerates zero tenants, and no
|
||||
schedule fires again — silently. Nothing is destroyed; the bytes are one
|
||||
directory up.
|
||||
**The org is folded to its storage identity once, at the door**
|
||||
(`store.Org`, via the one injective slugger `namespace.Sanitize`), and
|
||||
never again when that identity is turned into a place. `Principal.Org`
|
||||
downstream is therefore the STORED name — which is what lets
|
||||
`ListPrincipals` read a slug back off disk and reach the same file.
|
||||
Folding twice would re-suffix it into a different, empty database, since
|
||||
Sanitize is injective but deliberately not idempotent. A name Sanitize
|
||||
refuses is kept raw so `Valid` rejects it, because folding it would yield
|
||||
the empty org — and an empty org leg is the ROOT principal, so a name too
|
||||
hostile to store would otherwise have become the deployment's own tenant.
|
||||
|
||||
tasksd upgrade --data <dir>
|
||||
### Known gap — no user-scoped shard
|
||||
|
||||
inserts the two sentinel levels uniformly (`<data>/X/*.db` →
|
||||
`<data>/X/_/_/`, including `_` → `_/_/_`). A rename within one
|
||||
filesystem: no page rewritten, no re-encryption. Journals move before
|
||||
their database, so a run that stops half way never leaves a database
|
||||
without the WAL holding its last commits. Idempotent — a moved shard
|
||||
leaves nothing behind — so a deploy can run it unconditionally ahead of
|
||||
the daemon.
|
||||
`Principal` still carries a `User` leg (it is the wire encoding, and
|
||||
`ParsePrincipal`/`String` round-trip it), but a principal that narrows to
|
||||
a user **cannot open a shard**: `Principal.Namespace` returns
|
||||
`store.ErrUserLeg`. `namespace.Key` refuses a user namespace on purpose —
|
||||
that layout has no place for one, "and inventing one silently is how a
|
||||
second convention starts". The old `<org>/<project>/<user>/` tree was
|
||||
tasks' own; sharing the estate's layout costs that depth until
|
||||
`hanzoai/namespace` grows a user layout. Scope to the org or its project.
|
||||
|
||||
Two refusals, both loud, both non-destructive:
|
||||
|
||||
- **A master key is set.** Pre-upgrade shards are plaintext and moving
|
||||
one does not encrypt it; the next boot would wrap a fresh DEK around a
|
||||
file it then failed to decrypt. Move first with `KMS_MASTER_KEY`
|
||||
unset, encrypt second.
|
||||
- **A destination is occupied** (`store.ErrOccupied`). The only way to
|
||||
get there is booting this binary before the move ran, which creates an
|
||||
EMPTY shard where the full one belongs — and anything started after
|
||||
that boot is in the empty one. Both files are kept, the other tenants
|
||||
still move, and the operator picks which is authoritative.
|
||||
|
||||
`upgrade` and `migrate` are different axes and stay separate:
|
||||
`upgrade` moves EVERY shard forward in time (layout), `migrate` moves ONE
|
||||
shard across space (nodes).
|
||||
`tasksd upgrade` is **gone** with the layout it moved shards into. The
|
||||
only subcommand is `migrate`, which moves ONE shard across NODES.
|
||||
|
||||
### What is genuinely missing (the real gap)
|
||||
|
||||
|
||||
+10
-40
@@ -70,17 +70,10 @@ func newCoordinator(mgr *store.Manager, rep replication.Replicator) func(ctx con
|
||||
}
|
||||
|
||||
func main() {
|
||||
// Two subcommands, two axes. `migrate` moves ONE shard between NODES;
|
||||
// `upgrade` moves EVERY shard forward into the layout this binary reads.
|
||||
if len(os.Args) > 1 {
|
||||
switch os.Args[1] {
|
||||
case "migrate":
|
||||
runMigrate(os.Args[2:])
|
||||
return
|
||||
case "upgrade":
|
||||
runUpgrade(os.Args[2:])
|
||||
return
|
||||
}
|
||||
// One subcommand: `migrate` moves ONE shard between NODES.
|
||||
if len(os.Args) > 1 && os.Args[1] == "migrate" {
|
||||
runMigrate(os.Args[2:])
|
||||
return
|
||||
}
|
||||
var (
|
||||
zapAddr = flag.String("zap", envStr("TASKS_ZAP_ADDR", ":9999"), "ZAP listen address: a host:port, or a path to bind a unix socket")
|
||||
@@ -230,32 +223,6 @@ func envStr(k, def string) string {
|
||||
return def
|
||||
}
|
||||
|
||||
// runUpgrade is the `tasksd upgrade --data <dir>` subcommand: the one-time
|
||||
// move of a store written before shards were principal-addressed into the
|
||||
// layout this binary reads (store.Upgrade). Idempotent, so a deploy can run
|
||||
// it unconditionally ahead of the daemon; it prints what it moved and exits
|
||||
// non-zero on anything it refused to do.
|
||||
func runUpgrade(args []string) {
|
||||
fs := flag.NewFlagSet("upgrade", flag.ExitOnError)
|
||||
dataDir := fs.String("data", envStr("TASKSD_DATA_DIR", "./tasks-data"), "Tasks persistence directory")
|
||||
_ = fs.Parse(args)
|
||||
master, err := masterKey()
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
moves, upgradeErr := store.Upgrade(*dataDir, master)
|
||||
if moves == nil {
|
||||
moves = []store.Move{}
|
||||
}
|
||||
out, _ := json.MarshalIndent(moves, "", " ")
|
||||
fmt.Println(string(out))
|
||||
if upgradeErr != nil {
|
||||
fmt.Fprintln(os.Stderr, "upgrade:", upgradeErr)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
// runMigrate is the `tasksd migrate ...` subcommand. Quiesces the
|
||||
// (tenant, namespace) shard via the consensus barrier, copies the
|
||||
// SQLite file under <dataDir>/_migrations/<id>/, and releases the
|
||||
@@ -265,16 +232,19 @@ func runMigrate(args []string) {
|
||||
dataDir := fs.String("data", envStr("TASKSD_DATA_DIR", "./tasks-data"), "Tasks persistence directory")
|
||||
org := fs.String("org", "", "Org id (empty = root tenant)")
|
||||
project := fs.String("project", "", "Project id (optional)")
|
||||
user := fs.String("user", "", "User id (optional)")
|
||||
ns := fs.String("namespace", "", "Namespace to migrate")
|
||||
to := fs.String("to", "", "Destination node id")
|
||||
from := fs.String("from", "", "Source node id (annotation only)")
|
||||
_ = fs.Parse(args)
|
||||
if *ns == "" || *to == "" {
|
||||
fmt.Fprintln(os.Stderr, "usage: tasksd migrate [--org <id> [--project <id> [--user <id>]]] --namespace <name> --to <node> [--from <node>]")
|
||||
fmt.Fprintln(os.Stderr, "usage: tasksd migrate [--org <id> [--project <id>]] --namespace <name> --to <node> [--from <node>]")
|
||||
os.Exit(2)
|
||||
}
|
||||
principal := store.Principal{Org: *org, Project: *project, User: *user}
|
||||
// Through the door: the operator writes the org as IAM spells it, and
|
||||
// store.Org folds it to the identity the shard is actually stored under.
|
||||
// A --user flag used to ride alongside these; the shared namespace layout
|
||||
// has no place for a user-scoped shard, so there is nothing to migrate.
|
||||
principal := store.OrgProject(*org, *project)
|
||||
if err := principal.Valid(); err != nil {
|
||||
fmt.Fprintln(os.Stderr, "migrate:", err)
|
||||
os.Exit(2)
|
||||
|
||||
@@ -11,7 +11,9 @@ retract (
|
||||
require (
|
||||
github.com/golang-jwt/jwt/v5 v5.3.1
|
||||
github.com/hanzoai/authz v1.10.29
|
||||
github.com/hanzoai/sqlite v0.4.0
|
||||
github.com/hanzoai/cek v0.2.1
|
||||
github.com/hanzoai/namespace v1.2.0
|
||||
github.com/hanzoai/sqlite v0.5.0
|
||||
github.com/luxfi/consensus v1.36.3
|
||||
github.com/luxfi/log v1.5.0
|
||||
github.com/luxfi/zap v1.2.7
|
||||
|
||||
@@ -41,12 +41,16 @@ github.com/grandcat/zeroconf v1.0.0 h1:uHhahLBKqwWBV6WZUDAT71044vwOTL+McW0mBJvo6
|
||||
github.com/grandcat/zeroconf v1.0.0/go.mod h1:lTKmG1zh86XyCoUeIHSA4FJMBwCJiQmGfcP2PdzytEs=
|
||||
github.com/hanzoai/authz v1.10.29 h1:b4vWtI9g4Mvay1zizW7cwly/hDk06r+oAdA/y+317Do=
|
||||
github.com/hanzoai/authz v1.10.29/go.mod h1:xkzFdiIFx4UQMlU0NkmSRz0dgRQYqSXpKgVjn8ijn3E=
|
||||
github.com/hanzoai/cek v0.2.1 h1:J4+zrfZrY3r9N8db+V1sVBmZInAnVJHybF4Xdtywtgg=
|
||||
github.com/hanzoai/cek v0.2.1/go.mod h1:+Aud0VRbBDw5oMxOL6ivIZW4cgTpDEdrMxFcMcBUytY=
|
||||
github.com/hanzoai/csqlite v0.1.0 h1:suwC3dh0INlfP/U0Es6cDf6JNQ+2+GVLLATPWCUux6k=
|
||||
github.com/hanzoai/csqlite v0.1.0/go.mod h1:H31a/O6VXuklR9UBkgY++bmAK5uzVfXPqU0F6P9Wsos=
|
||||
github.com/hanzoai/namespace v1.2.0 h1:FLEHzR/GyE/OltXozMvT2n2FfUUxTuVVbILKZBkEjLI=
|
||||
github.com/hanzoai/namespace v1.2.0/go.mod h1:vGuxMDHg5MdhH5q2bqqjZL/afINBWJubP4WD1YwZTm4=
|
||||
github.com/hanzoai/sqlcipher v0.1.1 h1:GARjSiUEa1lwhd1/f87XRaujZBG5s1ZwxrZW2Es/ADI=
|
||||
github.com/hanzoai/sqlcipher v0.1.1/go.mod h1:F0soUYM1i4sawOZUpRvVnWoUayPbeGVlGq01VXy9Aqg=
|
||||
github.com/hanzoai/sqlite v0.4.0 h1:xlF6w6xzn2VSeAVBKFrZfIq0ISqRSTHz2g3QN3eCcnE=
|
||||
github.com/hanzoai/sqlite v0.4.0/go.mod h1:sE5uvENQ+q+rI011AwFuAuJ7DwOoDnywyCQstZ0ekHQ=
|
||||
github.com/hanzoai/sqlite v0.5.0 h1:1YydiyNAvL+WcXC1lUqZsUDaR4/7YkVb+wZm9qq9DSc=
|
||||
github.com/hanzoai/sqlite v0.5.0/go.mod h1:7hlAtZspL0Ggx/j0cSo6npPFtUeikvIxnMDb7yTaJD0=
|
||||
github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
|
||||
github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
|
||||
github.com/holiman/uint256 v1.3.2 h1:a9EgMPSC1AAaj1SZL5zIQD3WbwTuHrMGOerLjGmM/TA=
|
||||
|
||||
@@ -34,7 +34,7 @@ func TestEngine_StartWorkflow(t *testing.T) {
|
||||
// A namespace is an addressing fact, not a resource to provision:
|
||||
// starting in one that has never been seen registers it.
|
||||
{name: "unseen namespace", ns: "ghost", wfID: "wf-c", typ: TypeRef{Name: "Demo"}},
|
||||
{name: "unusable namespace", ns: "../escape", wfID: "wf-d", typ: TypeRef{Name: "Demo"}, wantErr: "path separator"},
|
||||
{name: "unusable namespace", ns: "../escape", wfID: "wf-d", typ: TypeRef{Name: "Demo"}, wantErr: "cannot name a shard"},
|
||||
{name: "sentinel namespace", ns: "_", wfID: "wf-e", typ: TypeRef{Name: "Demo"}, wantErr: "sentinel"},
|
||||
}
|
||||
for _, c := range cases {
|
||||
|
||||
@@ -1,139 +0,0 @@
|
||||
// Copyright © 2026 Hanzo AI. MIT License.
|
||||
|
||||
package store
|
||||
|
||||
import (
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
||||
hanzosqlite "github.com/hanzoai/sqlite"
|
||||
)
|
||||
|
||||
// Every shard file is encrypted under its OWN data-encryption key (DEK),
|
||||
// generated once when the file is created and never changed. The DEK is
|
||||
// stored beside the file, wrapped under the key-encryption key its
|
||||
// Principal derives from the master key (Principal.KEK):
|
||||
//
|
||||
// <shard>.db SQLCipher ciphertext, encrypted with the DEK
|
||||
// <shard>.dek id(16) || AES-256-GCM(KEK, DEK)
|
||||
//
|
||||
// Rotating the master key rewraps the sidecar and leaves the ciphertext
|
||||
// pages untouched, so rotation is O(1) per shard and cannot brick a file.
|
||||
//
|
||||
// The wrapped blob is bound to (Principal, id) as GCM additional data, and
|
||||
// the id is random rather than derived from the path — so a shard survives
|
||||
// moving the data directory, and a sidecar lifted into another tenant's
|
||||
// directory fails the tag instead of unwrapping.
|
||||
|
||||
// dekSuffix names the sidecar beside a shard's .db file.
|
||||
const dekSuffix = ".dek"
|
||||
|
||||
// idLen is the length of a shard's random identity, the value the wrapped
|
||||
// DEK is bound to.
|
||||
const idLen = 16
|
||||
|
||||
// principalShard tags the binding context of a wrapped shard DEK. The KEK
|
||||
// chain already separates tenants; this separates one tenant's shards from
|
||||
// each other.
|
||||
const principalShard = hanzosqlite.PrincipalType("shard")
|
||||
|
||||
// dekPath returns the sidecar path for a shard file.
|
||||
func dekPath(shard string) string { return shard + dekSuffix }
|
||||
|
||||
// shardDEK returns the data-encryption key for the shard file at path,
|
||||
// creating it on first use. A nil master returns a nil key, which opens the
|
||||
// shard unencrypted — the zero-config dev posture.
|
||||
func shardDEK(path string, p Principal, master []byte) ([]byte, error) {
|
||||
if master == nil {
|
||||
return nil, nil
|
||||
}
|
||||
kek, err := p.KEK(master)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
side := dekPath(path)
|
||||
blob, err := os.ReadFile(side)
|
||||
switch {
|
||||
case err == nil:
|
||||
return unwrap(kek, p, blob)
|
||||
case !errors.Is(err, os.ErrNotExist):
|
||||
return nil, fmt.Errorf("store: read %s: %w", side, err)
|
||||
}
|
||||
|
||||
dek, err := hanzosqlite.NewDEK()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
id := make([]byte, idLen)
|
||||
if _, err := rand.Read(id); err != nil {
|
||||
return nil, fmt.Errorf("store: shard id: %w", err)
|
||||
}
|
||||
wrapped, err := hanzosqlite.WrapDEK(kek, dek, shardAAD(p, id))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
blob = append(append(make([]byte, 0, idLen+len(wrapped)), id...), wrapped...)
|
||||
if err := writeNew(side, blob); err != nil {
|
||||
if !errors.Is(err, os.ErrExist) {
|
||||
return nil, err
|
||||
}
|
||||
// Another opener created it first; theirs is authoritative.
|
||||
if blob, err = os.ReadFile(side); err != nil {
|
||||
return nil, fmt.Errorf("store: read %s: %w", side, err)
|
||||
}
|
||||
return unwrap(kek, p, blob)
|
||||
}
|
||||
return dek, nil
|
||||
}
|
||||
|
||||
// unwrap opens a sidecar blob under p's KEK.
|
||||
func unwrap(kek []byte, p Principal, blob []byte) ([]byte, error) {
|
||||
if len(blob) <= idLen {
|
||||
return nil, fmt.Errorf("store: wrapped DEK is %d bytes, too short", len(blob))
|
||||
}
|
||||
return hanzosqlite.UnwrapDEK(kek, blob[idLen:], shardAAD(p, blob[:idLen]))
|
||||
}
|
||||
|
||||
// shardAAD is the binding context of a wrapped shard DEK: its tenant and
|
||||
// its identity, in the injective length-prefixed encoding hanzoai/sqlite
|
||||
// uses for every principal binding.
|
||||
func shardAAD(p Principal, id []byte) []byte {
|
||||
return hanzosqlite.PrincipalAAD(principalShard, p.String()+"/"+hex.EncodeToString(id))
|
||||
}
|
||||
|
||||
// writeNew writes b to path and fails with os.ErrExist if another writer
|
||||
// got there first. The temp file lands in the destination directory so the
|
||||
// rename is atomic.
|
||||
func writeNew(path string, b []byte) error {
|
||||
if _, err := os.Stat(path); err == nil {
|
||||
return os.ErrExist
|
||||
}
|
||||
tmp, err := os.CreateTemp(filepath.Dir(path), filepath.Base(path)+".*")
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: create %s: %w", path, err)
|
||||
}
|
||||
defer os.Remove(tmp.Name())
|
||||
if _, err := tmp.Write(b); err != nil {
|
||||
tmp.Close()
|
||||
return fmt.Errorf("store: write %s: %w", path, err)
|
||||
}
|
||||
if err := tmp.Chmod(0o600); err != nil {
|
||||
tmp.Close()
|
||||
return fmt.Errorf("store: chmod %s: %w", path, err)
|
||||
}
|
||||
if err := tmp.Sync(); err != nil {
|
||||
tmp.Close()
|
||||
return fmt.Errorf("store: sync %s: %w", path, err)
|
||||
}
|
||||
if err := tmp.Close(); err != nil {
|
||||
return fmt.Errorf("store: close %s: %w", path, err)
|
||||
}
|
||||
if _, err := os.Stat(path); err == nil {
|
||||
return os.ErrExist
|
||||
}
|
||||
return os.Rename(tmp.Name(), path)
|
||||
}
|
||||
@@ -1,204 +0,0 @@
|
||||
// Copyright © 2026 Hanzo AI. MIT License.
|
||||
|
||||
package store_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/hanzoai/tasks/pkg/tasks/store"
|
||||
)
|
||||
|
||||
// secret is the value the encryption tests look for on disk. It is long and
|
||||
// distinctive so finding it is proof the page was written in the clear.
|
||||
var secret = []byte("workflow-input-that-must-not-appear-on-disk-4a1f9c")
|
||||
|
||||
const sqliteHeader = "SQLite format 3\x00"
|
||||
|
||||
func master(b byte) []byte { return bytes.Repeat([]byte{b}, store.MasterKeyLen) }
|
||||
|
||||
// write puts secret into a tenant's shard and closes the manager so the file
|
||||
// is sealed. It returns the shard's path.
|
||||
func write(t *testing.T, dir string, key []byte, p store.Principal) string {
|
||||
t.Helper()
|
||||
mgr, err := store.New(dir, key)
|
||||
if err != nil {
|
||||
t.Fatalf("store.New: %v", err)
|
||||
}
|
||||
ctx := context.Background()
|
||||
sh, err := mgr.Get(ctx, p, "default")
|
||||
if err != nil {
|
||||
t.Fatalf("Get: %v", err)
|
||||
}
|
||||
if err := sh.Put(ctx, "wf/default/a/1", secret); err != nil {
|
||||
t.Fatalf("Put: %v", err)
|
||||
}
|
||||
path := mgr.ShardPath(p, "default")
|
||||
if err := mgr.Close(); err != nil {
|
||||
t.Fatalf("Close: %v", err)
|
||||
}
|
||||
return path
|
||||
}
|
||||
|
||||
// Without a master key the shard is plaintext — which is what makes the
|
||||
// encrypted case below a real measurement rather than a tautology.
|
||||
func TestShard_PlaintextWithoutMaster(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
path := write(t, dir, nil, store.Org("acme"))
|
||||
raw, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
t.Fatalf("read %s: %v", path, err)
|
||||
}
|
||||
if !bytes.Contains(raw, secret) {
|
||||
t.Fatal("an unencrypted shard should hold the value verbatim; the probe is not measuring the file it thinks it is")
|
||||
}
|
||||
if !bytes.HasPrefix(raw, []byte(sqliteHeader)) {
|
||||
t.Fatal("an unencrypted shard should carry the SQLite header")
|
||||
}
|
||||
}
|
||||
|
||||
// With a master key the file on disk is ciphertext: no SQLite header, and
|
||||
// the value written through the shard appears nowhere in it.
|
||||
func TestShard_EncryptedAtRest(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
path := write(t, dir, master(0x11), store.Org("acme"))
|
||||
|
||||
raw, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
t.Fatalf("read %s: %v", path, err)
|
||||
}
|
||||
if len(raw) == 0 {
|
||||
t.Fatal("shard file is empty")
|
||||
}
|
||||
if bytes.Contains(raw, secret) {
|
||||
t.Fatalf("%s holds the value in the clear", path)
|
||||
}
|
||||
if bytes.HasPrefix(raw, []byte(sqliteHeader)) {
|
||||
t.Fatalf("%s is a plaintext SQLite database", path)
|
||||
}
|
||||
if _, err := os.Stat(path + ".dek"); err != nil {
|
||||
t.Fatalf("wrapped DEK sidecar missing: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// The same master key reads it back; a different one cannot.
|
||||
func TestShard_RoundTripsUnderItsMaster(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
p := store.Org("acme")
|
||||
write(t, dir, master(0x11), p)
|
||||
|
||||
mgr, err := store.New(dir, master(0x11))
|
||||
if err != nil {
|
||||
t.Fatalf("store.New: %v", err)
|
||||
}
|
||||
ctx := context.Background()
|
||||
sh, err := mgr.Get(ctx, p, "default")
|
||||
if err != nil {
|
||||
t.Fatalf("reopen under its own master: %v", err)
|
||||
}
|
||||
got, ok, err := sh.Get(ctx, "wf/default/a/1")
|
||||
if err != nil || !ok {
|
||||
t.Fatalf("Get: ok=%v err=%v", ok, err)
|
||||
}
|
||||
if !bytes.Equal(got, secret) {
|
||||
t.Fatalf("read %q, want %q", got, secret)
|
||||
}
|
||||
_ = mgr.Close()
|
||||
|
||||
wrong, err := store.New(dir, master(0x22))
|
||||
if err != nil {
|
||||
t.Fatalf("store.New: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = wrong.Close() })
|
||||
if _, err := wrong.Get(ctx, p, "default"); err == nil {
|
||||
t.Fatal("a shard opened under the wrong master must fail, not return an empty database")
|
||||
}
|
||||
}
|
||||
|
||||
// A sidecar is bound to its tenant: lifting one into another tenant's
|
||||
// directory must fail the tag rather than unwrap.
|
||||
func TestShard_SidecarBoundToItsTenant(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
key := master(0x11)
|
||||
victim := store.Org("acme")
|
||||
thief := store.Org("evil")
|
||||
victimPath := write(t, dir, key, victim)
|
||||
thiefPath := write(t, dir, key, thief)
|
||||
|
||||
blob, err := os.ReadFile(victimPath + ".dek")
|
||||
if err != nil {
|
||||
t.Fatalf("read sidecar: %v", err)
|
||||
}
|
||||
// Give the thief the victim's shard AND its sidecar.
|
||||
body, err := os.ReadFile(victimPath)
|
||||
if err != nil {
|
||||
t.Fatalf("read shard: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(thiefPath, body, 0o600); err != nil {
|
||||
t.Fatalf("plant shard: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(thiefPath+".dek", blob, 0o600); err != nil {
|
||||
t.Fatalf("plant sidecar: %v", err)
|
||||
}
|
||||
|
||||
mgr, err := store.New(dir, key)
|
||||
if err != nil {
|
||||
t.Fatalf("store.New: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = mgr.Close() })
|
||||
if _, err := mgr.Get(context.Background(), thief, "default"); err == nil {
|
||||
t.Fatal("a sidecar lifted into another tenant's directory must not unwrap")
|
||||
}
|
||||
}
|
||||
|
||||
// Every tenant's shard is encrypted, including the root one, and each under
|
||||
// a key of its own.
|
||||
func TestShard_EveryTenantEncrypted(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
key := master(0x33)
|
||||
tenants := []store.Principal{
|
||||
{},
|
||||
{Org: "acme"},
|
||||
{Org: "acme", Project: "web"},
|
||||
{Org: "acme", Project: "web", User: "z"},
|
||||
}
|
||||
seen := map[string]store.Principal{}
|
||||
for _, p := range tenants {
|
||||
path := write(t, dir, key, p)
|
||||
raw, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
t.Fatalf("read %s: %v", path, err)
|
||||
}
|
||||
if bytes.Contains(raw, secret) || bytes.HasPrefix(raw, []byte(sqliteHeader)) {
|
||||
t.Fatalf("%+v: %s is not encrypted", p, path)
|
||||
}
|
||||
blob, err := os.ReadFile(path + ".dek")
|
||||
if err != nil {
|
||||
t.Fatalf("%+v: sidecar missing: %v", p, err)
|
||||
}
|
||||
if prev, dup := seen[string(blob)]; dup {
|
||||
t.Fatalf("%+v and %+v share a wrapped DEK", prev, p)
|
||||
}
|
||||
seen[string(blob)] = p
|
||||
}
|
||||
}
|
||||
|
||||
// A master key of the wrong size is refused where it enters, not at the
|
||||
// first shard open.
|
||||
func TestManager_MasterKeyLength(t *testing.T) {
|
||||
for _, n := range []int{1, 16, 31, 33, 64} {
|
||||
if _, err := store.New(t.TempDir(), bytes.Repeat([]byte{1}, n)); err == nil {
|
||||
t.Fatalf("a %d-byte master key should be refused", n)
|
||||
}
|
||||
}
|
||||
mgr, err := store.New(t.TempDir(), master(0x44))
|
||||
if err != nil {
|
||||
t.Fatalf("a %d-byte master key should be accepted: %v", store.MasterKeyLen, err)
|
||||
}
|
||||
if !mgr.Encrypted() {
|
||||
t.Fatal("Encrypted() should report a keyed manager")
|
||||
}
|
||||
_ = mgr.Close()
|
||||
}
|
||||
+105
-50
@@ -3,11 +3,11 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
hanzosqlite "github.com/hanzoai/sqlite"
|
||||
"github.com/hanzoai/namespace"
|
||||
)
|
||||
|
||||
// Sentinel is the path segment standing for an unset leg of a Principal.
|
||||
@@ -41,7 +41,44 @@ type Principal struct {
|
||||
}
|
||||
|
||||
// Org returns the principal naming an org.
|
||||
func Org(org string) Principal { return Principal{Org: org} }
|
||||
//
|
||||
// It is THE DOOR: the raw IAM org a caller holds is folded here, once, through
|
||||
// namespace.Sanitize — the one injective slugger — so everything downstream
|
||||
// carries the STORAGE identity. That matters because a Principal is both the
|
||||
// routing key on a replication frame and the name of the directory its shard
|
||||
// lives in: folding again at the open would re-suffix a slug that already
|
||||
// carries a disambiguation suffix, and land the tenant in a different, empty
|
||||
// database than the one ListPrincipals just read off disk.
|
||||
//
|
||||
// A name Sanitize REFUSES — it carries whitespace, a control or a format rune,
|
||||
// the class no injective fold survives — is kept RAW rather than folded, so
|
||||
// that Valid rejects it. Folding it would produce the empty string, and an
|
||||
// empty org leg is the ROOT principal: a name too hostile to store would have
|
||||
// become the deployment's own platform tenant. Fail-closed, loudly.
|
||||
func Org(org string) Principal { return Principal{Org: fold(org)} }
|
||||
|
||||
// OrgProject is the door for a principal narrowed to one of an org's
|
||||
// projects. Both legs are folded, by the same rule, in the same place.
|
||||
func OrgProject(org, project string) Principal {
|
||||
return Principal{Org: fold(org), Project: fold(project)}
|
||||
}
|
||||
|
||||
// fold reduces one externally-chosen name to its storage identity.
|
||||
//
|
||||
// A name Sanitize REFUSES is returned RAW rather than folded, so that Valid
|
||||
// rejects it. Sanitize's refusal is the empty string, and an empty leg means
|
||||
// "unset" — an unset org leg is the ROOT principal — so folding a hostile
|
||||
// name would silently hand it the deployment's own platform tenant instead
|
||||
// of turning it away. An empty input is genuinely unset and stays so.
|
||||
func fold(name string) string {
|
||||
if name == "" {
|
||||
return ""
|
||||
}
|
||||
if slug := namespace.Sanitize(name); slug != "" {
|
||||
return slug
|
||||
}
|
||||
return name
|
||||
}
|
||||
|
||||
// legs returns the three legs in narrowing order.
|
||||
func (p Principal) legs() [Depth]string { return [Depth]string{p.Org, p.Project, p.User} }
|
||||
@@ -67,9 +104,6 @@ func (p Principal) String() string {
|
||||
return strings.Join(out, "/")
|
||||
}
|
||||
|
||||
// dir is the shard directory for p relative to the store root.
|
||||
func (p Principal) dir() string { return filepath.FromSlash(p.String()) }
|
||||
|
||||
// ParsePrincipal is the inverse of String.
|
||||
func ParsePrincipal(s string) (Principal, error) {
|
||||
legs := strings.Split(s, "/")
|
||||
@@ -110,67 +144,88 @@ func (p Principal) Valid() error {
|
||||
// or file, or collide with the unset-leg sentinel. Every name that becomes
|
||||
// a path segment — a principal's legs and a namespace alike — passes
|
||||
// through here, so there is one rule for all of them.
|
||||
//
|
||||
// The rule is hanzoai/namespace's own segment rule, asked rather than
|
||||
// restated: a name must already BE a legal namespace segment, exactly as
|
||||
// written. That settles what a name cannot be by construction instead of by
|
||||
// denylist — no separator so no path, no dot so no ".." and no hidden name,
|
||||
// no trailing space, no encoded form of any of those — and it holds for the
|
||||
// namespace leg too, which cek renders straight into a filename.
|
||||
//
|
||||
// "Exactly as written" is the strict part, and it is deliberate. The
|
||||
// namespace constructor case-folds, so it would ACCEPT "Acme" and quietly
|
||||
// store it as "acme"; a leg that only becomes legal after folding has two
|
||||
// spellings, and the second one names the same file under a different key.
|
||||
// Legs arrive already folded (see Org), so anything that would change here
|
||||
// did not come through the door.
|
||||
func ValidName(name string) error {
|
||||
switch name {
|
||||
case "":
|
||||
return fmt.Errorf("store: name required")
|
||||
case Sentinel:
|
||||
return fmt.Errorf("store: %q is the unset-leg sentinel and cannot name a tenant", name)
|
||||
case ".", "..":
|
||||
return fmt.Errorf("store: %q is not a name", name)
|
||||
}
|
||||
if strings.ContainsAny(name, `/\`) || strings.ContainsRune(name, 0) {
|
||||
return fmt.Errorf("store: name %q contains a path separator", name)
|
||||
if strings.ContainsRune(name, 0) {
|
||||
return fmt.Errorf("store: name %q contains a NUL", name)
|
||||
}
|
||||
ns, err := namespace.Org(name)
|
||||
if err != nil {
|
||||
return fmt.Errorf("store: %q cannot name a shard: %w", name, err)
|
||||
}
|
||||
if ns.ID() != name {
|
||||
return fmt.Errorf("store: name %q is not in its stored form (%q); it did not come through the door", name, ns.ID())
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// principalProject is the domain-separation tag for the project leg.
|
||||
// hanzoai/sqlite ships global/org/user; project is the third leg IAM
|
||||
// already validates and mints (auth.Claims.Project), so it needs its own
|
||||
// tag for the HKDF info to stay injective across legs.
|
||||
const principalProject = hanzosqlite.PrincipalType("project")
|
||||
|
||||
// rootPrincipalID is the id the root (zero) Principal derives under. The
|
||||
// root tenant is the service itself, so it derives as the global
|
||||
// principal — distinct from any org, project or user that could be named
|
||||
// "tasks".
|
||||
const rootPrincipalID = "tasks"
|
||||
|
||||
// KEK derives the key-encryption key for p from the master key by walking
|
||||
// the principal chain: master → org → project → user, skipping any leg p
|
||||
// leaves unset. Each narrowing step derives from the key it narrows, so a
|
||||
// user's key is cryptographically contained in its org's — reaching a
|
||||
// user's data requires the whole chain, not just the master. Each step
|
||||
// carries its own domain-separation tag, so narrowing to a project named
|
||||
// "z" and to a user named "z" derive different keys.
|
||||
// ErrUserLeg reports that a Principal narrows to a USER, which the shared
|
||||
// namespace layout has no place for.
|
||||
//
|
||||
// The KEK never encrypts a page. It wraps that shard's own DEK (see
|
||||
// dek.go), which is what makes master-key rotation O(1) per shard.
|
||||
func (p Principal) KEK(master []byte) ([]byte, error) {
|
||||
// hanzoai/namespace names an org, an org's project, and the deployment
|
||||
// itself. namespace.Key returns an error for a user namespace on purpose —
|
||||
// "this layout has no place for them, and inventing one silently is how a
|
||||
// second convention starts". Tasks used to write <root>/<org>/<project>/<user>/,
|
||||
// a layout of its own; it now shares the estate's, and the user leg is the
|
||||
// one part of its tenancy that does not survive the move.
|
||||
//
|
||||
// It is refused at the door rather than only when the file is keyed, so a
|
||||
// user-scoped shard fails the same way in dev as in production instead of
|
||||
// working plaintext and dying the day a master key is set.
|
||||
var ErrUserLeg = errors.New("store: the shared namespace layout has no place for a user-scoped shard; scope to the org or its project")
|
||||
|
||||
// Namespace is the entity hanzoai/namespace names this shard's tenant, and
|
||||
// therefore both where its file lives and what its key is derived from —
|
||||
// one name, two renderings, so the two cannot drift apart:
|
||||
//
|
||||
// root (zero) → system → orgs/_platform/<ns>.db
|
||||
// {Org} → org/<org> → orgs/<org>/<ns>.db
|
||||
// {Org, Project} → org/<org>/<proj> → orgs/<org>/projects/<proj>/<ns>.db
|
||||
// anything with User → ErrUserLeg
|
||||
//
|
||||
// The legs are already the folded storage identity (see Org), so this
|
||||
// VALIDATES them and does not fold again — namespace.Sanitize is injective
|
||||
// but not idempotent, and a second fold would rename a tenant whose first
|
||||
// fold carried a disambiguation suffix.
|
||||
func (p Principal) Namespace() (namespace.Namespace, error) {
|
||||
if err := p.Valid(); err != nil {
|
||||
return nil, err
|
||||
return namespace.Namespace{}, err
|
||||
}
|
||||
if p.User != "" {
|
||||
return namespace.Namespace{}, fmt.Errorf("%w: %s", ErrUserLeg, p)
|
||||
}
|
||||
if p.Root() {
|
||||
return hanzosqlite.DeriveKey(master, hanzosqlite.PrincipalGlobal, rootPrincipalID)
|
||||
return namespace.System(), nil
|
||||
}
|
||||
key, err := hanzosqlite.DeriveKey(master, hanzosqlite.PrincipalOrg, p.Org)
|
||||
ns, err := namespace.Org(p.Org)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return namespace.Namespace{}, err
|
||||
}
|
||||
for _, leg := range []struct {
|
||||
typ hanzosqlite.PrincipalType
|
||||
id string
|
||||
}{
|
||||
{principalProject, p.Project},
|
||||
{hanzosqlite.PrincipalUser, p.User},
|
||||
} {
|
||||
if leg.id == "" {
|
||||
continue
|
||||
}
|
||||
if key, err = hanzosqlite.DeriveChildKey(key, leg.typ, leg.id); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if p.Project == "" {
|
||||
return ns, nil
|
||||
}
|
||||
return key, nil
|
||||
g, err := namespace.NewGroup(p.Project)
|
||||
if err != nil {
|
||||
return namespace.Namespace{}, fmt.Errorf("store: project %q: %w", p.Project, err)
|
||||
}
|
||||
return ns.WithGroup(g), nil
|
||||
}
|
||||
|
||||
@@ -5,8 +5,10 @@ package store_test
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
|
||||
"github.com/hanzoai/cek"
|
||||
"github.com/hanzoai/tasks/pkg/tasks/store"
|
||||
)
|
||||
|
||||
@@ -72,26 +74,31 @@ func TestPrincipal_Rejected(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// The KEK must separate every tenant, and narrowing must derive a child of
|
||||
// the leg to its left rather than a fresh key off the master.
|
||||
func TestPrincipal_KEKSeparatesTenants(t *testing.T) {
|
||||
// Every tenant must derive its own key, and a different master must move
|
||||
// all of them. The derivation is cek's now, but the property is still
|
||||
// tasks' to hold: it depends on Principal.Namespace being injective, which
|
||||
// is the part this package owns.
|
||||
func TestPrincipal_NamespaceSeparatesTenants(t *testing.T) {
|
||||
master := bytes.Repeat([]byte{0x5a}, store.MasterKeyLen)
|
||||
tenants := []store.Principal{
|
||||
{},
|
||||
{Org: "acme"},
|
||||
{Org: "acme", Project: "web"},
|
||||
{Org: "acme", User: "z"},
|
||||
{Org: "acme", Project: "web", User: "z"},
|
||||
{Org: "acme", Project: "z"},
|
||||
{Org: "other"},
|
||||
}
|
||||
seen := map[string]store.Principal{}
|
||||
for _, p := range tenants {
|
||||
k, err := p.KEK(master)
|
||||
ns, err := p.Namespace()
|
||||
if err != nil {
|
||||
t.Fatalf("KEK(%+v): %v", p, err)
|
||||
t.Fatalf("Namespace(%+v): %v", p, err)
|
||||
}
|
||||
k, err := cek.DeriveKey(master, ns, "default")
|
||||
if err != nil {
|
||||
t.Fatalf("DeriveKey(%+v): %v", p, err)
|
||||
}
|
||||
if len(k) != store.MasterKeyLen {
|
||||
t.Fatalf("KEK(%+v) is %d bytes", p, len(k))
|
||||
t.Fatalf("key for %+v is %d bytes", p, len(k))
|
||||
}
|
||||
if prev, dup := seen[string(k)]; dup {
|
||||
t.Fatalf("%+v and %+v derive the same key", prev, p)
|
||||
@@ -101,11 +108,73 @@ func TestPrincipal_KEKSeparatesTenants(t *testing.T) {
|
||||
// A different master must move every key.
|
||||
other := bytes.Repeat([]byte{0x5b}, store.MasterKeyLen)
|
||||
for _, p := range tenants {
|
||||
k, _ := p.KEK(other)
|
||||
ns, _ := p.Namespace()
|
||||
k, _ := cek.DeriveKey(other, ns, "default")
|
||||
if _, collides := seen[string(k)]; collides {
|
||||
t.Fatalf("%+v derives the same key under a different master", p)
|
||||
}
|
||||
}
|
||||
// Two namespaces of the SAME tenant must not share a key either.
|
||||
acme, _ := store.Org("acme").Namespace()
|
||||
a, _ := cek.DeriveKey(master, acme, "default")
|
||||
b, _ := cek.DeriveKey(master, acme, "staging")
|
||||
if bytes.Equal(a, b) {
|
||||
t.Fatal("two namespaces of one tenant derive the same key")
|
||||
}
|
||||
}
|
||||
|
||||
// A user-scoped principal has no place in the shared namespace layout, and
|
||||
// is refused rather than silently written somewhere. It is refused at the
|
||||
// door — not only once a key is in play — so it fails identically in dev
|
||||
// and in production.
|
||||
func TestPrincipal_UserLegRefused(t *testing.T) {
|
||||
for _, p := range []store.Principal{
|
||||
{Org: "acme", User: "z"},
|
||||
{Org: "acme", Project: "web", User: "z"},
|
||||
} {
|
||||
if _, err := p.Namespace(); !errors.Is(err, store.ErrUserLeg) {
|
||||
t.Fatalf("Namespace(%+v) = %v, want ErrUserLeg", p, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The org is folded to its storage identity ONCE, at the door (Org), and
|
||||
// never again when that identity is turned into a place (Namespace).
|
||||
//
|
||||
// This is the invariant ListPrincipals rests on. It reads a slug back off
|
||||
// disk and builds a Principal from it directly; if Namespace folded a
|
||||
// second time the tenant would render into a DIFFERENT, empty database,
|
||||
// because namespace.Sanitize is injective but deliberately not idempotent —
|
||||
// it re-suffixes anything already shaped like its own output, so that an
|
||||
// entity cannot squat on the rendered form of another's name.
|
||||
func TestOrg_FoldsOnceAtTheDoor(t *testing.T) {
|
||||
// "Acme" is not a clean slug, so the door folds it to a disambiguated one.
|
||||
folded := store.Org("Acme")
|
||||
if folded.Org == "Acme" {
|
||||
t.Fatal("Org did not fold a non-slug name")
|
||||
}
|
||||
// Distinct raw orgs must stay distinct after folding.
|
||||
if store.Org("Acme").Org == store.Org("acme").Org {
|
||||
t.Fatal("two distinct orgs folded onto one identity")
|
||||
}
|
||||
// A name no injective fold survives is refused, not mangled.
|
||||
if err := store.Org("ac me").Valid(); err == nil {
|
||||
t.Fatal("an org carrying whitespace should be refused")
|
||||
}
|
||||
// THE INVARIANT: a Principal rebuilt from the stored identity — which is
|
||||
// exactly what ListPrincipals does — names the same place as the one the
|
||||
// door produced.
|
||||
want, err := folded.Namespace()
|
||||
if err != nil {
|
||||
t.Fatalf("Namespace: %v", err)
|
||||
}
|
||||
got, err := store.Principal{Org: folded.Org}.Namespace()
|
||||
if err != nil {
|
||||
t.Fatalf("Namespace (rebuilt): %v", err)
|
||||
}
|
||||
if got != want {
|
||||
t.Fatalf("rebuilt tenant names %s, door named %s — the fold ran twice", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// Two tenants that name the same namespace must not share a shard.
|
||||
@@ -121,8 +190,7 @@ func TestManager_TenantsAreIsolated(t *testing.T) {
|
||||
{},
|
||||
{Org: "acme"},
|
||||
{Org: "acme", Project: "web"},
|
||||
{Org: "acme", User: "z"},
|
||||
{Org: "acme", Project: "web", User: "z"},
|
||||
{Org: "acme", Project: "z"},
|
||||
{Org: "other"},
|
||||
}
|
||||
for _, p := range tenants {
|
||||
@@ -134,6 +202,10 @@ func TestManager_TenantsAreIsolated(t *testing.T) {
|
||||
t.Fatalf("Put(%+v): %v", p, err)
|
||||
}
|
||||
}
|
||||
// A user-scoped tenant opens no shard at all, so it can share none.
|
||||
if _, err := mgr.Get(ctx, store.Principal{Org: "acme", User: "z"}, "default"); !errors.Is(err, store.ErrUserLeg) {
|
||||
t.Fatalf("Get(user-scoped) = %v, want ErrUserLeg", err)
|
||||
}
|
||||
for _, p := range tenants {
|
||||
sh, _ := mgr.Get(ctx, p, "default")
|
||||
got, ok, err := sh.Get(ctx, "wf/default/a/1")
|
||||
|
||||
+197
-70
@@ -3,16 +3,18 @@
|
||||
// Package store implements per-(org, namespace) SQLite shards with
|
||||
// optional consensus replication.
|
||||
//
|
||||
// Layout on disk:
|
||||
// Layout on disk — hanzoai/namespace's, shared with every other store in
|
||||
// the estate rather than one of tasks' own:
|
||||
//
|
||||
// <rootDir>/<org>/<project>/<user>/<namespace>.db
|
||||
//
|
||||
// The three directory levels are the Principal that owns the shard, each
|
||||
// unset leg written as the sentinel "_" (see principal.go). The root
|
||||
// tenant — the unscoped embedded / dev path — is therefore <rootDir>/_/_/_.
|
||||
// <rootDir>/orgs/<org>/<namespace>.db
|
||||
// <rootDir>/orgs/<org>/projects/<project>/<namespace>.db
|
||||
// <rootDir>/orgs/_platform/<namespace>.db the root (unscoped) tenant
|
||||
//
|
||||
// A shard is created on first use. Namespaces are not declared up front.
|
||||
//
|
||||
// A Principal that narrows to a USER has no place in this layout and is
|
||||
// refused — see ErrUserLeg.
|
||||
//
|
||||
// Schema (single table; key/value blob):
|
||||
//
|
||||
// CREATE TABLE kv(
|
||||
@@ -23,9 +25,11 @@
|
||||
//
|
||||
// Encryption:
|
||||
//
|
||||
// A non-nil master key opens every shard at rest encrypted, under a DEK
|
||||
// of its own wrapped for its Principal (see dek.go). A nil master key
|
||||
// opens plaintext — the zero-config dev posture.
|
||||
// Every shard is encrypted at rest, keyed by hanzoai/cek from the process
|
||||
// master and the namespace that owns it. There is no plaintext posture and
|
||||
// no key material beside the file: a shard is born encrypted or it does not
|
||||
// exist. A deployment with no master mints a random one that dies with the
|
||||
// process, which is the honest shape of a keyless run.
|
||||
//
|
||||
// Replication:
|
||||
//
|
||||
@@ -42,6 +46,7 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strings"
|
||||
@@ -49,13 +54,43 @@ import (
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/hanzoai/cek"
|
||||
"github.com/hanzoai/namespace"
|
||||
"github.com/hanzoai/tasks/pkg/tasks/replication"
|
||||
)
|
||||
|
||||
// shardSuffix names a shard file. One namespace, one file, one suffix —
|
||||
// the manager builds paths with it and the upgrade recognises them by it.
|
||||
// shardSuffix names a shard file. One namespace, one file, one suffix.
|
||||
const shardSuffix = ".db"
|
||||
|
||||
// The segments hanzoai/namespace lays its tree out with, READ OUT OF
|
||||
// namespace ITSELF rather than spelled again here.
|
||||
//
|
||||
// These strings are directory names on live volumes. A second spelling of
|
||||
// one does not fail loudly — it opens an empty database beside the real
|
||||
// one — so the only safe way to walk a layout you do not own is to ask the
|
||||
// package that writes it where it put things.
|
||||
var (
|
||||
orgsRoot, platformSlug = splitTail(mustKey(namespace.System()))
|
||||
projectsDir = midSegment(mustKey(namespace.MustOrg("o").WithGroup(namespace.MustGroup("g"))))
|
||||
)
|
||||
|
||||
// mustKey renders a namespace's directory. The inputs are fixed in the
|
||||
// source, so a failure is a broken dependency, not a runtime condition.
|
||||
func mustKey(ns namespace.Namespace) []string {
|
||||
key, err := namespace.Key(ns, "probe")
|
||||
if err != nil {
|
||||
panic("store: hanzoai/namespace cannot render its own layout: " + err.Error())
|
||||
}
|
||||
return strings.Split(path.Dir(key), "/")
|
||||
}
|
||||
|
||||
// splitTail returns the first and last segments: "orgs", "_platform".
|
||||
func splitTail(segs []string) (string, string) { return segs[0], segs[len(segs)-1] }
|
||||
|
||||
// midSegment returns the segment naming the project container in
|
||||
// orgs/<org>/projects/<group>.
|
||||
func midSegment(segs []string) string { return segs[len(segs)-2] }
|
||||
|
||||
// IdleEvictAfter sets how long an open shard may sit unused before the
|
||||
// manager closes it. Mutable for tests.
|
||||
var IdleEvictAfter = 10 * time.Minute
|
||||
@@ -64,9 +99,12 @@ var IdleEvictAfter = 10 * time.Minute
|
||||
var ErrClosed = errors.New("store: manager closed")
|
||||
|
||||
// Manager owns the on-disk shard layout and the open-shard cache.
|
||||
//
|
||||
// It holds no key. The master lives in hanzoai/cek — one key per process,
|
||||
// for every database the process opens — because threading it through every
|
||||
// caller would not make it less global, only harder to see.
|
||||
type Manager struct {
|
||||
rootDir string
|
||||
master []byte
|
||||
replicator replication.Replicator
|
||||
mu sync.Mutex
|
||||
shards map[shardKey]*Shard
|
||||
@@ -82,22 +120,43 @@ type shardKey struct {
|
||||
// New opens the manager rooted at rootDir. The directory is created on
|
||||
// demand; errors are returned only for unrecoverable IO problems.
|
||||
//
|
||||
// master is the 32-byte root key every shard's DEK is wrapped under. A nil
|
||||
// master opens shards plaintext — the zero-config dev posture; any other
|
||||
// length is rejected here rather than at the first shard open.
|
||||
// master is the 32-byte root key every shard is derived from, resolved from
|
||||
// KMS by whoever runs this. It is installed into hanzoai/cek, which is where
|
||||
// the process keeps it.
|
||||
//
|
||||
// A nil master does NOT mean plaintext — there is no such posture any more.
|
||||
// It means "I have no key to state":
|
||||
//
|
||||
// - Embedded in a process that already installed one (cloud resolves it
|
||||
// through credz at boot), that key is used. One deployment, one key —
|
||||
// an embedded engine minting a second would write files its host could
|
||||
// not read.
|
||||
// - Otherwise a random master is minted for this process. Nothing it
|
||||
// writes outlives the run, by construction, which is the honest shape
|
||||
// of a deployment that was given no key and the reason a developer
|
||||
// needs no configuration at all.
|
||||
func New(rootDir string, master []byte) (*Manager, error) {
|
||||
if rootDir == "" {
|
||||
return nil, fmt.Errorf("store.New: rootDir required")
|
||||
}
|
||||
if master != nil && len(master) != MasterKeyLen {
|
||||
return nil, fmt.Errorf("store.New: master key must be %d bytes, got %d", MasterKeyLen, len(master))
|
||||
switch {
|
||||
case master != nil:
|
||||
if len(master) != MasterKeyLen {
|
||||
return nil, fmt.Errorf("store.New: master key must be %d bytes, got %d", MasterKeyLen, len(master))
|
||||
}
|
||||
if err := cek.SetMaster(master); err != nil {
|
||||
return nil, fmt.Errorf("store.New: %w", err)
|
||||
}
|
||||
case !cek.HasMaster():
|
||||
if _, err := cek.SetDevMaster(); err != nil {
|
||||
return nil, fmt.Errorf("store.New: %w", err)
|
||||
}
|
||||
}
|
||||
if err := os.MkdirAll(rootDir, 0o755); err != nil {
|
||||
return nil, fmt.Errorf("store.New: mkdir: %w", err)
|
||||
}
|
||||
m := &Manager{
|
||||
rootDir: rootDir,
|
||||
master: master,
|
||||
shards: make(map[shardKey]*Shard),
|
||||
stopGC: make(chan struct{}),
|
||||
}
|
||||
@@ -106,10 +165,11 @@ func New(rootDir string, master []byte) (*Manager, error) {
|
||||
}
|
||||
|
||||
// MasterKeyLen is the required length of the store's root key.
|
||||
const MasterKeyLen = 32
|
||||
const MasterKeyLen = cek.KeyLen
|
||||
|
||||
// Encrypted reports whether shards are opened at rest encrypted.
|
||||
func (m *Manager) Encrypted() bool { return m.master != nil }
|
||||
// Encrypted reports whether shards are opened at rest encrypted. They always
|
||||
// are; it answers whether this process holds a key to open any at all.
|
||||
func (m *Manager) Encrypted() bool { return cek.HasMaster() }
|
||||
|
||||
// WithReplicator installs r as the consensus driver for every shard
|
||||
// opened from now on, and re-installs it on already-open shards.
|
||||
@@ -148,6 +208,10 @@ func (m *Manager) Get(ctx context.Context, p Principal, ns string) (*Shard, erro
|
||||
if err := p.Valid(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
tenant, err := p.Namespace()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
k := shardKey{principal: p, ns: ns}
|
||||
m.mu.Lock()
|
||||
if s, ok := m.shards[k]; ok {
|
||||
@@ -157,16 +221,7 @@ func (m *Manager) Get(ctx context.Context, p Principal, ns string) (*Shard, erro
|
||||
}
|
||||
m.mu.Unlock()
|
||||
|
||||
dir := filepath.Join(m.rootDir, p.dir())
|
||||
if err := os.MkdirAll(dir, 0o755); err != nil {
|
||||
return nil, fmt.Errorf("store.Get: mkdir %s: %w", dir, err)
|
||||
}
|
||||
path := filepath.Join(dir, ns+shardSuffix)
|
||||
dek, err := shardDEK(path, p, m.master)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
s, err := openShard(path, p, ns, dek)
|
||||
s, err := openShard(tenant, m.rootDir, p, ns)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -187,50 +242,90 @@ func (m *Manager) Get(ctx context.Context, p Principal, ns string) (*Shard, erro
|
||||
// ListPrincipals enumerates every tenant that owns at least one shard.
|
||||
// Used by the cron sweeper, which must see EVERY tenant's schedules from
|
||||
// the root engine.
|
||||
//
|
||||
// It walks the namespace layout: orgs/_platform is the root tenant, every
|
||||
// other orgs/<slug> is an org, and each of those may hold projects/<slug>.
|
||||
// The slug it reads back is ALREADY the folded storage identity, so the
|
||||
// Principal it builds renders straight back to the directory it came from
|
||||
// (see Org for why folding twice would not).
|
||||
func (m *Manager) ListPrincipals(ctx context.Context) ([]Principal, error) {
|
||||
var out []Principal
|
||||
err := m.walkTenants(m.rootDir, nil, &out)
|
||||
orgs, err := subdirs(filepath.Join(m.rootDir, orgsRoot))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, org := range orgs {
|
||||
p := Principal{Org: org}
|
||||
if org == platformSlug {
|
||||
p = Principal{}
|
||||
}
|
||||
if m.holdsShard(p) {
|
||||
out = append(out, p)
|
||||
}
|
||||
if org == platformSlug {
|
||||
continue // the deployment itself has no projects
|
||||
}
|
||||
projects, err := subdirs(filepath.Join(m.rootDir, orgsRoot, org, projectsDir))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, proj := range projects {
|
||||
if q := (Principal{Org: org, Project: proj}); m.holdsShard(q) {
|
||||
out = append(out, q)
|
||||
}
|
||||
}
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].String() < out[j].String() })
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// walkTenants descends exactly Depth directory levels, collecting the
|
||||
// principal each leaf directory encodes.
|
||||
func (m *Manager) walkTenants(dir string, legs []string, out *[]Principal) error {
|
||||
if len(legs) == Depth {
|
||||
p, err := ParsePrincipal(strings.Join(legs, "/"))
|
||||
if err != nil {
|
||||
// A directory that does not encode a principal is not ours.
|
||||
return nil
|
||||
}
|
||||
*out = append(*out, p)
|
||||
return nil
|
||||
}
|
||||
entries, err := os.ReadDir(dir)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
for _, e := range entries {
|
||||
if !e.IsDir() {
|
||||
continue
|
||||
}
|
||||
if err := m.walkTenants(filepath.Join(dir, e.Name()), append(legs, e.Name()), out); err != nil {
|
||||
return err
|
||||
// holdsShard reports whether p owns at least one shard. A directory that
|
||||
// holds only projects/ is a container, not a tenant, and listing it would
|
||||
// hand the sweeper a tenant with nothing in it.
|
||||
func (m *Manager) holdsShard(p Principal) bool { return len(m.shardNames(p)) > 0 }
|
||||
|
||||
// shardNames is every namespace p owns, sorted: the shard files on disk,
|
||||
// UNION the ones this process holds open.
|
||||
//
|
||||
// The union is not belt-and-braces, it is the answer. On the pure-Go
|
||||
// SQLCipher envelope the ciphertext is written when the file is SEALED —
|
||||
// at checkpoint or close — so a shard created moments ago has no file yet
|
||||
// and a disk-only walk reports its tenant as having nothing. That is
|
||||
// precisely the tenant the cron sweeper most needs to see. Reading the
|
||||
// resident set as well makes the answer independent of which backend is
|
||||
// linked and of when the last seal happened.
|
||||
func (m *Manager) shardNames(p Principal) []string {
|
||||
seen := map[string]bool{}
|
||||
if dir, err := m.tenantDir(p); err == nil {
|
||||
entries, err := os.ReadDir(dir)
|
||||
if err == nil {
|
||||
for _, e := range entries {
|
||||
name := e.Name()
|
||||
if e.IsDir() || !strings.HasSuffix(name, shardSuffix) || strings.HasPrefix(name, "-") {
|
||||
continue
|
||||
}
|
||||
seen[strings.TrimSuffix(name, shardSuffix)] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
m.mu.Lock()
|
||||
for k := range m.shards {
|
||||
if k.principal == p {
|
||||
seen[k.ns] = true
|
||||
}
|
||||
}
|
||||
m.mu.Unlock()
|
||||
out := make([]string, 0, len(seen))
|
||||
for ns := range seen {
|
||||
out = append(out, ns)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// ListShards enumerates every namespace shard the principal owns. Used by
|
||||
// cross-namespace operations like ListNamespaces().
|
||||
func (m *Manager) ListShards(ctx context.Context, p Principal) ([]*Shard, error) {
|
||||
dir := filepath.Join(m.rootDir, p.dir())
|
||||
// subdirs lists the directory names directly under dir. A missing dir is
|
||||
// empty, not an error: nothing has been written yet.
|
||||
func subdirs(dir string) ([]string, error) {
|
||||
entries, err := os.ReadDir(dir)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
@@ -238,20 +333,36 @@ func (m *Manager) ListShards(ctx context.Context, p Principal) ([]*Shard, error)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
out := make([]*Shard, 0, len(entries))
|
||||
var out []string
|
||||
for _, e := range entries {
|
||||
name := e.Name()
|
||||
if !strings.HasSuffix(name, shardSuffix) || strings.HasPrefix(name, "-") {
|
||||
continue
|
||||
if e.IsDir() {
|
||||
out = append(out, e.Name())
|
||||
}
|
||||
ns := strings.TrimSuffix(name, shardSuffix)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// tenantDir is the directory holding every shard p owns.
|
||||
func (m *Manager) tenantDir(p Principal) (string, error) {
|
||||
path, err := m.shardPath(p, "any")
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return filepath.Dir(path), nil
|
||||
}
|
||||
|
||||
// ListShards enumerates every namespace shard the principal owns. Used by
|
||||
// cross-namespace operations like ListNamespaces().
|
||||
func (m *Manager) ListShards(ctx context.Context, p Principal) ([]*Shard, error) {
|
||||
names := m.shardNames(p)
|
||||
out := make([]*Shard, 0, len(names))
|
||||
for _, ns := range names {
|
||||
s, err := m.Get(ctx, p, ns)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, s)
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].ns < out[j].ns })
|
||||
return out, nil
|
||||
}
|
||||
|
||||
@@ -345,9 +456,25 @@ func CopyFile(dst, src string) (int64, error) {
|
||||
return io.Copy(out, in)
|
||||
}
|
||||
|
||||
// ShardPath returns the on-disk file for (principal, ns).
|
||||
// ShardPath returns the on-disk file for (principal, ns), or "" for a
|
||||
// principal the layout cannot name.
|
||||
func (m *Manager) ShardPath(p Principal, ns string) string {
|
||||
return filepath.Join(m.rootDir, p.dir(), ns+shardSuffix)
|
||||
path, err := m.shardPath(p, ns)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
return path
|
||||
}
|
||||
|
||||
// shardPath renders (principal, ns) through namespace — the SAME rendering
|
||||
// cek.Open uses to place the file, so the path a caller is handed is the
|
||||
// path the shard was opened at and not a second guess at it.
|
||||
func (m *Manager) shardPath(p Principal, ns string) (string, error) {
|
||||
tenant, err := p.Namespace()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return namespace.Path(m.rootDir, tenant, ns)
|
||||
}
|
||||
|
||||
// Replicator returns the currently-installed driver, or nil.
|
||||
|
||||
@@ -17,6 +17,8 @@ import (
|
||||
// opens through it (never modernc directly) so a store embedded in the unified
|
||||
// cloud binary shares that single registration instead of double-registering
|
||||
// "sqlite" and panicking at init.
|
||||
"github.com/hanzoai/cek"
|
||||
"github.com/hanzoai/namespace"
|
||||
hanzosqlite "github.com/hanzoai/sqlite"
|
||||
"github.com/hanzoai/tasks/pkg/tasks/replication"
|
||||
)
|
||||
@@ -69,18 +71,27 @@ CREATE TABLE IF NOT EXISTS meta (
|
||||
);
|
||||
`
|
||||
|
||||
// openShard opens the shard file at path. A non-nil dek opens it at rest
|
||||
// encrypted; hanzoai/sqlite routes that through the live libsqlcipher codec
|
||||
// when one is linked and through its pure-Go SQLCipher codec envelope when
|
||||
// one is not, so a keyed shard is ciphertext on every build and there is
|
||||
// nothing here to branch on.
|
||||
func openShard(path string, p Principal, ns string, dek []byte) (*Shard, error) {
|
||||
// openShard opens the shard holding namespace ns for tenant, under dir.
|
||||
//
|
||||
// cek derives the file's key from the process master and the tenant that
|
||||
// owns it, picks the path, creates the directory and opens it encrypted —
|
||||
// so the file's location and its key are two renderings of one name and
|
||||
// cannot be paired wrongly. hanzoai/sqlite routes the keyed open through
|
||||
// the live libsqlcipher codec when one is linked and through its pure-Go
|
||||
// SQLCipher codec envelope when one is not, so a shard is ciphertext on
|
||||
// every build and there is nothing here to branch on.
|
||||
func openShard(tenant namespace.Namespace, dir string, p Principal, ns string) (*Shard, error) {
|
||||
// WAL + busy_timeout + synchronous=NORMAL + foreign_keys are set by schemaSQL
|
||||
// below (explicit PRAGMA statements), so they apply on both backends rather
|
||||
// than relying on driver-specific DSN params.
|
||||
conn, err := hanzosqlite.OpenDB(path, dek)
|
||||
conn, err := cek.Open(tenant, ns, dir)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("store.openShard: open %s: %w", path, err)
|
||||
return nil, fmt.Errorf("store.openShard: open %s/%s: %w", tenant, ns, err)
|
||||
}
|
||||
path, err := namespace.Path(dir, tenant, ns)
|
||||
if err != nil {
|
||||
_ = conn.Close()
|
||||
return nil, err
|
||||
}
|
||||
conn.SetMaxOpenConns(1) // single writer
|
||||
conn.SetMaxIdleConns(1)
|
||||
|
||||
@@ -1,143 +0,0 @@
|
||||
// Copyright © 2026 Hanzo AI. MIT License.
|
||||
|
||||
package store
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// The one-time move a store written by an older binary needs before this
|
||||
// one can read it. That binary addressed a shard by org alone:
|
||||
//
|
||||
// <root>/<org>/<ns>.db
|
||||
// <root>/_/<ns>.db the unscoped embedded / dev tenant
|
||||
//
|
||||
// This one addresses it by the whole Principal, three levels deep, every
|
||||
// unset leg written as the sentinel (see principal.go):
|
||||
//
|
||||
// <root>/<org>/_/_/<ns>.db
|
||||
// <root>/_/_/_/<ns>.db
|
||||
//
|
||||
// So the move inserts two sentinel levels, uniformly. Nothing is read,
|
||||
// re-encoded or re-keyed — a rename within one filesystem, not a page
|
||||
// rewritten.
|
||||
//
|
||||
// Without it the manager finds nothing where it looks and creates empty
|
||||
// shards beside the full ones: every mid-flight execution, every schedule
|
||||
// and every namespace ceases to exist while its bytes sit one directory
|
||||
// up, and the sweeper enumerates zero tenants, so no schedule fires again.
|
||||
// Nothing is destroyed and the move brings all of it back.
|
||||
|
||||
// Move is one file's relocation.
|
||||
type Move struct {
|
||||
From string `json:"from"`
|
||||
To string `json:"to"`
|
||||
// Skipped is set when the destination was already occupied. Both files
|
||||
// survive untouched and Upgrade returns ErrOccupied, because only an
|
||||
// operator can say which one is authoritative: the destination can only
|
||||
// have got there by this binary booting before the move ran, which
|
||||
// creates an EMPTY shard where the full one belongs — and anything
|
||||
// started after that boot is in the empty one.
|
||||
Skipped bool `json:"skipped,omitempty"`
|
||||
}
|
||||
|
||||
// ErrOccupied reports that at least one shard had a file at its
|
||||
// destination. Nothing was overwritten and nothing was deleted; the run
|
||||
// carried on past it, so the other tenants still moved.
|
||||
var ErrOccupied = errors.New("store: a shard already had a file at its destination")
|
||||
|
||||
// Upgrade moves every shard under root into the principal-addressed layout
|
||||
// and reports what it moved. Idempotent: a moved shard leaves nothing at
|
||||
// its old path, and a directory already in this layout holds no shard file
|
||||
// at its top level, so a re-run costs a scan and nothing else.
|
||||
//
|
||||
// master must be nil. A store written by the older binary is PLAINTEXT —
|
||||
// encryption arrived with this layout — and moving a file does not encrypt
|
||||
// it. Boot once with the key unset and the shards read; boot with the key
|
||||
// set and each one gets a fresh DEK sidecar and is then opened as
|
||||
// SQLCipher ciphertext, which it is not. Move first, encrypt second.
|
||||
func Upgrade(root string, master []byte) ([]Move, error) {
|
||||
if master != nil {
|
||||
return nil, fmt.Errorf("store.Upgrade: refusing to move %s while a master key is set: "+
|
||||
"shards written before this layout are plaintext and moving them does not encrypt them, "+
|
||||
"so the next boot would wrap a fresh DEK around a file it then failed to decrypt — "+
|
||||
"run the move with the key unset", root)
|
||||
}
|
||||
entries, err := os.ReadDir(root)
|
||||
if err != nil {
|
||||
if os.IsNotExist(err) {
|
||||
return nil, nil
|
||||
}
|
||||
return nil, fmt.Errorf("store.Upgrade: read %s: %w", root, err)
|
||||
}
|
||||
var moves []Move
|
||||
occupied := 0
|
||||
for _, e := range entries {
|
||||
if !e.IsDir() {
|
||||
continue
|
||||
}
|
||||
dir := filepath.Join(root, e.Name())
|
||||
files, err := os.ReadDir(dir)
|
||||
if err != nil {
|
||||
return moves, fmt.Errorf("store.Upgrade: read %s: %w", dir, err)
|
||||
}
|
||||
names := shardFiles(files)
|
||||
if names == nil {
|
||||
continue
|
||||
}
|
||||
dst := filepath.Join(dir, Sentinel, Sentinel)
|
||||
if err := os.MkdirAll(dst, 0o755); err != nil {
|
||||
return moves, fmt.Errorf("store.Upgrade: mkdir %s: %w", dst, err)
|
||||
}
|
||||
for _, name := range names {
|
||||
m := Move{From: filepath.Join(dir, name), To: filepath.Join(dst, name)}
|
||||
if _, err := os.Stat(m.To); err == nil {
|
||||
// Isolation, like the cron sweep's: one collision must not
|
||||
// cost every other tenant its move. Reported at the end.
|
||||
m.Skipped, occupied = true, occupied+1
|
||||
} else if err := os.Rename(m.From, m.To); err != nil {
|
||||
return moves, fmt.Errorf("store.Upgrade: move %s: %w", m.From, err)
|
||||
}
|
||||
moves = append(moves, m)
|
||||
}
|
||||
}
|
||||
if occupied > 0 {
|
||||
return moves, fmt.Errorf("%w: %d of %d; this binary booted before the move ran and created "+
|
||||
"empty shards where the full ones belong. Nothing was overwritten or deleted. Stop the "+
|
||||
"service, decide which file of each pair is authoritative, remove the other, and re-run",
|
||||
ErrOccupied, occupied, len(moves))
|
||||
}
|
||||
return moves, nil
|
||||
}
|
||||
|
||||
// shardFiles returns what to move out of a pre-upgrade tenant directory,
|
||||
// journals before their database, or nil when the directory holds no shard
|
||||
// at all — which is how a directory already in this layout, and
|
||||
// _migrations/, are left alone without naming either of them.
|
||||
//
|
||||
// Journals first because a run that stops half way must never leave a
|
||||
// database in its new home without the WAL holding its last commits. The
|
||||
// engine reads only the new layout, so a database left behind is merely
|
||||
// unmoved and a re-run finishes the job; a database moved without its WAL
|
||||
// has lost writes.
|
||||
func shardFiles(files []os.DirEntry) []string {
|
||||
var dbs, journals []string
|
||||
for _, f := range files {
|
||||
if f.IsDir() {
|
||||
continue
|
||||
}
|
||||
if strings.HasSuffix(f.Name(), shardSuffix) {
|
||||
dbs = append(dbs, f.Name())
|
||||
continue
|
||||
}
|
||||
journals = append(journals, f.Name())
|
||||
}
|
||||
if len(dbs) == 0 {
|
||||
return nil
|
||||
}
|
||||
return append(journals, dbs...)
|
||||
}
|
||||
@@ -1,275 +0,0 @@
|
||||
// Copyright © 2026 Hanzo AI. MIT License.
|
||||
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// These tests drive the move against a REAL store: data is written through
|
||||
// a manager, the tree is pushed back into the pre-upgrade shape, and the
|
||||
// same data has to come back out afterwards. Asserting on paths alone would
|
||||
// prove the files landed somewhere, not that the engine can read them.
|
||||
|
||||
// oldLayout writes value at key for each principal/namespace pair, closes
|
||||
// the manager, then lifts every shard file two levels up — which is exactly
|
||||
// the tree the older binary left behind. Returns the root.
|
||||
func oldLayout(t *testing.T, shards map[Principal]string, key, value string) string {
|
||||
t.Helper()
|
||||
root := t.TempDir()
|
||||
m, err := New(root, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
for p, ns := range shards {
|
||||
s, err := m.Get(context.Background(), p, ns)
|
||||
if err != nil {
|
||||
t.Fatalf("Get(%s/%s): %v", p, ns, err)
|
||||
}
|
||||
if err := s.Put(context.Background(), key, []byte(value)); err != nil {
|
||||
t.Fatalf("Put: %v", err)
|
||||
}
|
||||
}
|
||||
if err := m.Close(); err != nil {
|
||||
t.Fatalf("Close: %v", err)
|
||||
}
|
||||
for p := range shards {
|
||||
deep := filepath.Join(root, p.dir()) // <root>/<org>/_/_
|
||||
flat := filepath.Dir(filepath.Dir(deep)) // <root>/<org>
|
||||
entries, err := os.ReadDir(deep)
|
||||
if err != nil {
|
||||
t.Fatalf("ReadDir(%s): %v", deep, err)
|
||||
}
|
||||
for _, e := range entries {
|
||||
if err := os.Rename(filepath.Join(deep, e.Name()), filepath.Join(flat, e.Name())); err != nil {
|
||||
t.Fatalf("flatten %s: %v", e.Name(), err)
|
||||
}
|
||||
}
|
||||
if err := os.RemoveAll(filepath.Join(flat, Sentinel)); err != nil {
|
||||
t.Fatalf("RemoveAll: %v", err)
|
||||
}
|
||||
}
|
||||
return root
|
||||
}
|
||||
|
||||
// readBack opens a fresh manager on root and returns the value at key.
|
||||
func readBack(t *testing.T, root string, p Principal, ns, key string) string {
|
||||
t.Helper()
|
||||
m, err := New(root, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
defer m.Close()
|
||||
s, err := m.Get(context.Background(), p, ns)
|
||||
if err != nil {
|
||||
t.Fatalf("Get(%s/%s): %v", p, ns, err)
|
||||
}
|
||||
v, ok, err := s.Get(context.Background(), key)
|
||||
if err != nil {
|
||||
t.Fatalf("Get(%s): %v", key, err)
|
||||
}
|
||||
if !ok {
|
||||
return ""
|
||||
}
|
||||
return string(v)
|
||||
}
|
||||
|
||||
// TestUpgradeRestoresPreUpgradeShards is the whole point: a store the older
|
||||
// binary wrote is invisible to this one — the manager creates empty shards
|
||||
// beside the full ones and the sweeper enumerates nothing — until the move
|
||||
// runs. Both tenant shapes are covered, an org and the root, because the
|
||||
// root's directory is itself the sentinel and is the easy one to get wrong.
|
||||
func TestUpgradeRestoresPreUpgradeShards(t *testing.T) {
|
||||
shards := map[Principal]string{
|
||||
Org("acme"): "default",
|
||||
{}: "default",
|
||||
}
|
||||
root := oldLayout(t, shards, "sc/default/nightly", "backup")
|
||||
|
||||
// The premise: before the move this store has no tenants at all, which
|
||||
// is what stops the sweeper. Asserted before anything opens a shard,
|
||||
// because opening one CREATES it — that is the whole failure mode.
|
||||
m, err := New(root, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
ps, err := m.ListPrincipals(context.Background())
|
||||
m.Close()
|
||||
if err != nil {
|
||||
t.Fatalf("ListPrincipals: %v", err)
|
||||
}
|
||||
if len(ps) != 0 {
|
||||
t.Fatalf("pre-move principals = %v, want none — the sweeper sees nothing", ps)
|
||||
}
|
||||
|
||||
moves, err := Upgrade(root, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("Upgrade: %v", err)
|
||||
}
|
||||
if len(moves) != 2 {
|
||||
t.Fatalf("moves = %+v, want one per shard", moves)
|
||||
}
|
||||
for _, mv := range moves {
|
||||
if mv.Skipped {
|
||||
t.Fatalf("move %+v was skipped on a clean tree", mv)
|
||||
}
|
||||
}
|
||||
|
||||
for p, ns := range shards {
|
||||
if got := readBack(t, root, p, ns, "sc/default/nightly"); got != "backup" {
|
||||
t.Fatalf("post-move read for %s = %q, want backup", p, got)
|
||||
}
|
||||
}
|
||||
m2, err := New(root, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("New: %v", err)
|
||||
}
|
||||
defer m2.Close()
|
||||
ps, err = m2.ListPrincipals(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("ListPrincipals: %v", err)
|
||||
}
|
||||
if len(ps) != 2 {
|
||||
t.Fatalf("post-move principals = %v, want acme and root", ps)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUpgradeCarriesTheJournal proves a shard's WAL travels with it. A
|
||||
// database that lands in its new home without the journal holding its last
|
||||
// commits has silently lost writes, which is worse than not moving at all.
|
||||
func TestUpgradeCarriesTheJournal(t *testing.T) {
|
||||
root := oldLayout(t, map[Principal]string{Org("acme"): "default"}, "k", "v")
|
||||
wal := filepath.Join(root, "acme", "default.db-wal")
|
||||
if err := os.WriteFile(wal, []byte("journal"), 0o600); err != nil {
|
||||
t.Fatalf("WriteFile: %v", err)
|
||||
}
|
||||
if _, err := Upgrade(root, nil); err != nil {
|
||||
t.Fatalf("Upgrade: %v", err)
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(root, "acme", Sentinel, Sentinel, "default.db-wal")); err != nil {
|
||||
t.Fatalf("journal did not travel with its database: %v", err)
|
||||
}
|
||||
if _, err := os.Stat(wal); !os.IsNotExist(err) {
|
||||
t.Fatalf("journal left behind at %s (err=%v)", wal, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUpgradeIsIdempotent: running it again on a tree already in this
|
||||
// layout is a scan and nothing else, and the data is untouched.
|
||||
func TestUpgradeIsIdempotent(t *testing.T) {
|
||||
root := oldLayout(t, map[Principal]string{Org("acme"): "default"}, "k", "v")
|
||||
if _, err := Upgrade(root, nil); err != nil {
|
||||
t.Fatalf("first Upgrade: %v", err)
|
||||
}
|
||||
moves, err := Upgrade(root, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("second Upgrade: %v", err)
|
||||
}
|
||||
if len(moves) != 0 {
|
||||
t.Fatalf("second run moved %+v, want nothing", moves)
|
||||
}
|
||||
if got := readBack(t, root, Org("acme"), "default", "k"); got != "v" {
|
||||
t.Fatalf("read after two runs = %q, want v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUpgradeRefusesAnOccupiedTarget covers the collision an operator will
|
||||
// actually hit: the new binary BOOTED before the move ran, so it created an
|
||||
// empty shard exactly where the full one belongs. Overwriting would discard
|
||||
// whatever was started after that boot; skipping quietly would leave the
|
||||
// store looking upgraded and still empty. So neither: both files survive,
|
||||
// the run continues for every other tenant, and the error says what to do.
|
||||
func TestUpgradeRefusesAnOccupiedTarget(t *testing.T) {
|
||||
shards := map[Principal]string{Org("acme"): "default", Org("globex"): "default"}
|
||||
root := oldLayout(t, shards, "k", "real")
|
||||
|
||||
// The boot. Reading through a manager creates <root>/acme/_/_/default.db.
|
||||
if got := readBack(t, root, Org("acme"), "default", "k"); got != "" {
|
||||
t.Fatalf("the empty shard read %q, want empty", got)
|
||||
}
|
||||
|
||||
moves, err := Upgrade(root, nil)
|
||||
if !errors.Is(err, ErrOccupied) {
|
||||
t.Fatalf("Upgrade err = %v, want ErrOccupied", err)
|
||||
}
|
||||
if !strings.Contains(err.Error(), "re-run") {
|
||||
t.Fatalf("error %q does not say what to do", err)
|
||||
}
|
||||
|
||||
var skipped, moved int
|
||||
for _, m := range moves {
|
||||
if m.Skipped {
|
||||
skipped++
|
||||
continue
|
||||
}
|
||||
moved++
|
||||
}
|
||||
if skipped != 1 || moved != 1 {
|
||||
t.Fatalf("moves = %+v, want acme skipped and globex moved", moves)
|
||||
}
|
||||
// Neither file was consumed.
|
||||
if _, err := os.Stat(filepath.Join(root, "acme", "default.db")); err != nil {
|
||||
t.Fatalf("source was consumed: %v", err)
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(root, "acme", Sentinel, Sentinel, "default.db")); err != nil {
|
||||
t.Fatalf("destination was consumed: %v", err)
|
||||
}
|
||||
// Isolation: the collision did not cost the other tenant its move.
|
||||
if got := readBack(t, root, Org("globex"), "default", "k"); got != "real" {
|
||||
t.Fatalf("globex read = %q, want real", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUpgradeRefusesAMasterKey: the move does not encrypt, so running it
|
||||
// with a key set would produce the one state that cannot be recovered by
|
||||
// running it again — a plaintext file wrapped in a fresh DEK sidecar and
|
||||
// then opened as ciphertext. It refuses, says why, and touches nothing.
|
||||
func TestUpgradeRefusesAMasterKey(t *testing.T) {
|
||||
root := oldLayout(t, map[Principal]string{Org("acme"): "default"}, "k", "v")
|
||||
_, err := Upgrade(root, make([]byte, MasterKeyLen))
|
||||
if err == nil {
|
||||
t.Fatal("Upgrade with a master key returned nil")
|
||||
}
|
||||
for _, want := range []string{"master key", "plaintext"} {
|
||||
if !strings.Contains(err.Error(), want) {
|
||||
t.Fatalf("error %q does not say %q", err, want)
|
||||
}
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(root, "acme", "default.db")); err != nil {
|
||||
t.Fatalf("refusal moved something anyway: %v", err)
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(root, "acme", Sentinel)); !os.IsNotExist(err) {
|
||||
t.Fatalf("refusal created %s (err=%v)", Sentinel, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUpgradeLeavesNonTenantDirsAlone: the rule is "a directory holding a
|
||||
// shard file directly is a pre-upgrade tenant". _migrations/<id>/<ns>.db
|
||||
// keeps its shards one level further down, so it is not one, and the rule
|
||||
// leaves it alone without knowing its name.
|
||||
func TestUpgradeLeavesNonTenantDirsAlone(t *testing.T) {
|
||||
root := oldLayout(t, map[Principal]string{Org("acme"): "default"}, "k", "v")
|
||||
job := filepath.Join(root, "_migrations", "mig-1")
|
||||
if err := os.MkdirAll(job, 0o755); err != nil {
|
||||
t.Fatalf("MkdirAll: %v", err)
|
||||
}
|
||||
copied := filepath.Join(job, "default.db")
|
||||
if err := os.WriteFile(copied, []byte("copy"), 0o600); err != nil {
|
||||
t.Fatalf("WriteFile: %v", err)
|
||||
}
|
||||
|
||||
if _, err := Upgrade(root, nil); err != nil {
|
||||
t.Fatalf("Upgrade: %v", err)
|
||||
}
|
||||
if _, err := os.Stat(copied); err != nil {
|
||||
t.Fatalf("migration copy moved: %v", err)
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(root, "_migrations", Sentinel)); !os.IsNotExist(err) {
|
||||
t.Fatalf("_migrations was treated as a tenant (err=%v)", err)
|
||||
}
|
||||
}
|
||||
@@ -30,8 +30,6 @@ func TestTenancy_NamespaceOnDemand(t *testing.T) {
|
||||
{},
|
||||
Org("acme"),
|
||||
{Org: "acme", Project: "web"},
|
||||
{Org: "acme", User: "z"},
|
||||
{Org: "acme", Project: "web", User: "z"},
|
||||
} {
|
||||
v := emb.View(p)
|
||||
wf, err := v.en.StartWorkflow("never-declared", tenancySecret, "", TypeRef{Name: "Demo"}, "default", nil)
|
||||
@@ -48,6 +46,23 @@ func TestTenancy_NamespaceOnDemand(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// A user-scoped tenant is refused outright. The shared namespace layout
|
||||
// names an org and an org's project and nothing narrower, so the depth
|
||||
// tasks used to address by hand no longer has a place to be written — and
|
||||
// it fails here, at the start, rather than by quietly landing in the org's
|
||||
// own shard.
|
||||
func TestTenancy_UserScopedIsRefused(t *testing.T) {
|
||||
emb := embedFixture(t, t.TempDir(), nil)
|
||||
for _, p := range []Principal{
|
||||
{Org: "acme", User: "z"},
|
||||
{Org: "acme", Project: "web", User: "z"},
|
||||
} {
|
||||
if _, err := emb.View(p).en.StartWorkflow("default", "wf", "", TypeRef{Name: "Demo"}, "default", nil); err == nil {
|
||||
t.Fatalf("StartWorkflow(%+v) should be refused", p)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Each tenant sees only its own workflows, even in the same namespace.
|
||||
func TestTenancy_ViewsAreIsolated(t *testing.T) {
|
||||
emb := embedFixture(t, t.TempDir(), nil)
|
||||
@@ -55,7 +70,7 @@ func TestTenancy_ViewsAreIsolated(t *testing.T) {
|
||||
{},
|
||||
Org("acme"),
|
||||
{Org: "acme", Project: "web"},
|
||||
{Org: "acme", User: "z"},
|
||||
{Org: "acme", Project: "api"},
|
||||
Org("other"),
|
||||
}
|
||||
for _, p := range tenants {
|
||||
@@ -82,7 +97,7 @@ func TestTenancy_ViewsAreIsolated(t *testing.T) {
|
||||
func TestTenancy_EncryptedAtRest(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
key := bytes.Repeat([]byte{0x7e}, 32)
|
||||
tenants := []Principal{{}, Org("acme"), {Org: "acme", Project: "web", User: "z"}}
|
||||
tenants := []Principal{{}, Org("acme"), {Org: "acme", Project: "web"}}
|
||||
|
||||
emb := embedFixture(t, dir, key)
|
||||
for _, p := range tenants {
|
||||
|
||||
Reference in New Issue
Block a user