amadan

carlos / platform Public

console: O1/O2-bracketed listings and signed exact-version handles under CARLOS_HOST_CLAIM (claim Plan 7 Task 2)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude 4519638e801203ec4e634ab8b0d390a92d537daa parent bc75435
5 files changed, +1019 −4
  • internal/console/apidb.go +47 −0
  • internal/console/apidbclaim.go +297 −0
  • internal/console/apidbclaim_http_test.go +275 −0
  • internal/console/apidbclaim_test.go +389 −0
  • internal/console/server.go +11 −4
diff --git a/internal/console/apidb.go b/internal/console/apidb.go
index bf1a783d0..d474a3b88 100644
--- a/internal/console/apidb.go
+++ b/internal/console/apidb.go
@@ -31,6 +31,7 @@ import (
"github.com/carlosframework/platform/internal/activator"
"github.com/carlosframework/platform/internal/dbdoor"
+ "github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
@@ -91,6 +92,7 @@ const dbAuditAction = "db-read"
// dbRequest is what dbGate resolves for both handlers.
type dbRequest struct {
+ claim hostclaim.Observed
acct *Account
actor string
app string
@@ -162,6 +164,20 @@ func (s *Server) dbGate(w http.ResponseWriter, r *http.Request) (dbRequest, bool
jsonError(w, http.StatusBadGateway, "could not read the replica. Try again.")
return dbRequest{}, false
}
+ if s.cfg.HostClaim {
+ g := dbRequest{acct: acct, actor: actor, app: app, host: host}
+ if s.cfg.DBVersions == nil {
+ jsonError(w, http.StatusServiceUnavailable, "this console has no versioned database reader")
+ return dbRequest{}, false
+ }
+ g.claim, err = s.dbClaim(ctx, g, *rec)
+ if err != nil {
+ s.dbClaimFailure(w, r, g, err)
+ return dbRequest{}, false
+ }
+ g.prefix = g.claim.Claim.Holder.Location
+ return g, true
+ }
rd, err := s.dbReplica(ctx, prefix)
if err != nil {
log.Printf("console: api db %s/%s/%s: replica client: %v", acct.Sqid, app, host, err)
@@ -244,6 +260,16 @@ func (s *Server) handleAPIDBList(w http.ResponseWriter, r *http.Request) {
if !ok {
return
}
+ if s.cfg.HostClaim {
+ out, err := s.listDBVersions(r.Context(), g, level)
+ if err != nil {
+ s.dbClaimFailure(w, r, g, err)
+ return
+ }
+ w.Header().Set("Cache-Control", "no-store")
+ writeJSON(w, http.StatusOK, out)
+ return
+ }
itr, err := g.rd.LTXFiles(r.Context(), level, 0, false)
if err != nil {
s.dbStoreFailure(w, g, fmt.Sprintf("list level %d", level), err)
@@ -288,12 +314,27 @@ func (s *Server) handleAPIDBGet(w http.ResponseWriter, r *http.Request) {
if !ok {
return
}
+ var pinned *dbVersionReader
+ if s.cfg.HostClaim {
+ pinned, err = s.pinnedDBReader(r, g, level, minTXID, maxTXID)
+ if err != nil {
+ s.dbClaimFailure(w, r, g, err)
+ return
+ }
+ g.rd = pinned
+ }
rangeHdr := r.Header.Get("Range")
offset, size, err := dbdoor.ParseRange(rangeHdr)
if err != nil {
jsonError(w, http.StatusRequestedRangeNotSatisfiable, "unsupported range")
return
}
+ if pinned != nil {
+ if _, err := dbVersionRange(pinned.handle.Size, offset, size); err != nil {
+ jsonError(w, http.StatusRequestedRangeNotSatisfiable, "unsupported range")
+ return
+ }
+ }
if !s.dbStreams.acquire(g.actor) {
w.Header().Set("Retry-After", dbStreamRetryAfter)
jsonError(w, http.StatusTooManyRequests, "too many database downloads at once. Wait a few seconds and try again.")
@@ -330,6 +371,12 @@ func (s *Server) handleAPIDBGet(w http.ResponseWriter, r *http.Request) {
s.auditActivity(g.acct, g.app, g.actor, Activity{
Action: dbAuditAction, TargetType: "instance", Target: g.host, Detail: detail,
})
+ if pinned != nil {
+ if err := dbVersionHeaders(w, r, pinned.handle.Size, offset, size); err != nil {
+ jsonError(w, http.StatusRequestedRangeNotSatisfiable, "unsupported range")
+ return
+ }
+ }
w.Header().Set("Content-Type", "application/octet-stream")
w.Header().Set("Cache-Control", "no-store")
status := http.StatusOK
diff --git a/internal/console/apidbclaim.go b/internal/console/apidbclaim.go
new file mode 100644
index 000000000..9283268a2
--- /dev/null
+++ b/internal/console/apidbclaim.go
@@ -0,0 +1,297 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "crypto/ecdsa"
+ "crypto/hmac"
+ "crypto/sha256"
+ "encoding/base64"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "os"
+ "reflect"
+ "sort"
+ "strconv"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/dbdoor"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/superfly/ltx"
+)
+
+const dbHandleTTL = 15 * time.Minute
+
+var (
+ errDBNotAdopted = errors.New("host not adopted")
+ errDBClaimChanged = errors.New("host claim changed; retry the database pull")
+ errDBClaimClosed = errors.New("host is not accepting database reads")
+ errDBGeneration = errors.New("database generation requires a generation-aware reader")
+ errDBHandle = errors.New("invalid database file handle")
+ errDBHandleExpired = errors.New("database file handle expired; retry the database pull")
+)
+
+// dbFileHandle is encoded as this ordered JSON struct, never as a map.
+// Account and App bind the capability to the authenticated route as well.
+type dbFileHandle struct {
+ V int `json:"v"`
+ Host string `json:"host"`
+ Location string `json:"location"`
+ Key string `json:"key"`
+ VersionID string `json:"version_id"`
+ Incarnation string `json:"incarnation"`
+ Expiry int64 `json:"expiry"`
+ Account string `json:"account"`
+ App string `json:"app"`
+ Size int64 `json:"size"`
+}
+
+func dbHandleSecret(key *ecdsa.PrivateKey) []byte {
+ mac := hmac.New(sha256.New, key.D.FillBytes(make([]byte, 32)))
+ mac.Write([]byte("carlos-db-handle-key-v1\x00"))
+ return mac.Sum(nil)
+}
+
+func dbHandleMAC(secret, body []byte) []byte {
+ mac := hmac.New(sha256.New, secret)
+ mac.Write([]byte("carlos-db-file-v1\x00"))
+ mac.Write(body)
+ return mac.Sum(nil)
+}
+
+func signDBHandle(secret []byte, h dbFileHandle) (string, error) {
+ if len(secret) != sha256.Size {
+ return "", errDBHandle
+ }
+ body, err := json.Marshal(h)
+ if err != nil {
+ return "", err
+ }
+ return base64.RawURLEncoding.EncodeToString(body) + "." + base64.RawURLEncoding.EncodeToString(dbHandleMAC(secret, body)), nil
+}
+
+func verifyDBHandle(secret []byte, token string, now time.Time) (dbFileHandle, error) {
+ var h dbFileHandle
+ if len(secret) != sha256.Size || len(token) > 8192 {
+ return h, errDBHandle
+ }
+ a, b, ok := strings.Cut(token, ".")
+ if !ok {
+ return h, errDBHandle
+ }
+ body, err := base64.RawURLEncoding.DecodeString(a)
+ if err != nil {
+ return h, errDBHandle
+ }
+ sig, err := base64.RawURLEncoding.DecodeString(b)
+ if err != nil || !hmac.Equal(sig, dbHandleMAC(secret, body)) {
+ return h, errDBHandle
+ }
+ dec := json.NewDecoder(bytes.NewReader(body))
+ dec.DisallowUnknownFields()
+ if err := dec.Decode(&h); err != nil {
+ return dbFileHandle{}, errDBHandle
+ }
+ canonical, err := json.Marshal(h)
+ if err != nil || !bytes.Equal(body, canonical) || h.V != 1 || h.VersionID == "" || h.VersionID == "null" || h.Size < 0 {
+ return dbFileHandle{}, errDBHandle
+ }
+ if now.Unix() >= h.Expiry {
+ return dbFileHandle{}, errDBHandleExpired
+ }
+ return h, nil
+}
+
+func dbLevelPrefix(location string, level int) string {
+ // litestream/s3 uses hex level directories, unlike its file replica.
+ return fmt.Sprintf("%s/%04x/", location, level)
+}
+
+func dbVersionKey(location string, level int, lo, hi uint64) string {
+ return dbLevelPrefix(location, level) + ltx.FormatFilename(ltx.TXID(lo), ltx.TXID(hi))
+}
+
+func (s *Server) dbClaim(ctx context.Context, g dbRequest, rec provision.Record) (hostclaim.Observed, error) {
+ obs, err := hostclaim.Read(ctx, s.cfg.Releases, g.host)
+ if err != nil {
+ return hostclaim.Observed{}, err
+ }
+ if obs.Absent {
+ return obs, errDBNotAdopted
+ }
+ h := obs.Claim.Holder
+ if h == nil {
+ return obs, errDBClaimClosed
+ }
+ inc, err := hostclaim.Incarnation(g.acct.Sqid, g.app, g.host, rec.CreatedAt)
+ if err != nil {
+ return obs, err
+ }
+ if h.Family != "instance" || h.Account != g.acct.Sqid || h.App != g.app || h.Key != provision.RecordKey(g.acct.Sqid, g.app, g.host) || h.Incarnation != inc || !h.CreatedAt.Equal(rec.CreatedAt) {
+ return obs, os.ErrNotExist
+ }
+ switch obs.Claim.Op.Kind {
+ case hostclaim.KindDeleting, hostclaim.KindParking, hostclaim.KindPurging:
+ return obs, errDBClaimClosed
+ }
+ if h.Location == "" {
+ return obs, errDBClaimClosed
+ }
+ // The moves generation restore contract is not provided by Plans 1/2.
+ // Never present root files as the installed generation's database.
+ if h.Generation != "" {
+ return obs, errDBGeneration
+ }
+ return obs, nil
+}
+
+func sameDBAdmission(a, b hostclaim.Observed) bool {
+ if a.Absent || b.Absent || a.Claim == nil || b.Claim == nil {
+ return false
+ }
+ // §6.5 permits unrelated chore revisions; compare all holder/op content,
+ // including stamp, session, admissions, journal and execution ownership.
+ return a.Host == b.Host && reflect.DeepEqual(a.Claim.Holder, b.Claim.Holder) && reflect.DeepEqual(a.Claim.Op, b.Claim.Op)
+}
+
+func (s *Server) dbClaimFailure(w http.ResponseWriter, r *http.Request, g dbRequest, err error) {
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ http.NotFound(w, r)
+ case errors.Is(err, errDBNotAdopted), errors.Is(err, errDBClaimChanged), errors.Is(err, errDBClaimClosed), errors.Is(err, errDBGeneration):
+ w.Header().Set("Retry-After", "1")
+ jsonError(w, http.StatusConflict, err.Error())
+ case errors.Is(err, errDBHandleExpired):
+ jsonError(w, http.StatusGone, err.Error())
+ case errors.Is(err, errDBHandle):
+ jsonError(w, http.StatusBadRequest, err.Error())
+ default:
+ s.dbStoreFailure(w, g, "claim/version read", err)
+ }
+}
+
+func (s *Server) listDBVersions(ctx context.Context, g dbRequest, level int) (dbdoor.ListResponse, error) {
+ out := dbdoor.ListResponse{Protocol: 2, Files: []dbdoor.LTXFile{}}
+ h := g.claim.Claim.Holder
+ prefix := dbLevelPrefix(h.Location, level)
+ versions, err := s.cfg.DBVersions.ListVersions(ctx, s.cfg.HibernateBucket, prefix)
+ if err != nil {
+ return dbdoor.ListResponse{}, err
+ }
+ seen := map[string]bool{}
+ for _, v := range versions {
+ if !strings.HasPrefix(v.Key, prefix) {
+ return dbdoor.ListResponse{}, errors.New("db: malformed version listing")
+ }
+ if !v.IsLatest || v.DeleteMarker {
+ continue
+ }
+ name := strings.TrimPrefix(v.Key, prefix)
+ lo, hi, err := ltx.ParseFilename(name)
+ if err != nil || lo == 0 || hi < lo || v.Key != dbVersionKey(h.Location, level, uint64(lo), uint64(hi)) {
+ continue
+ }
+ if v.VersionID == "" || v.VersionID == "null" || v.Size < 0 || seen[v.Key] {
+ return dbdoor.ListResponse{}, errors.New("db: malformed version listing")
+ }
+ seen[v.Key] = true
+ token, err := signDBHandle(s.dbHandleKey, dbFileHandle{V: 1, Host: g.host, Location: h.Location, Key: v.Key, VersionID: v.VersionID, Incarnation: h.Incarnation, Expiry: time.Now().Add(dbHandleTTL).Unix(), Account: g.acct.Sqid, App: g.app, Size: v.Size})
+ if err != nil {
+ return dbdoor.ListResponse{}, err
+ }
+ out.Files = append(out.Files, dbdoor.LTXFile{Level: level, MinTXID: uint64(lo), MaxTXID: uint64(hi), Size: v.Size, CreatedAt: v.LastModified, Handle: token})
+ }
+ o2, err := hostclaim.Read(ctx, s.cfg.Releases, g.host)
+ if err != nil {
+ return dbdoor.ListResponse{}, err
+ }
+ if !sameDBAdmission(g.claim, o2) {
+ return dbdoor.ListResponse{}, errDBClaimChanged
+ }
+ sort.Slice(out.Files, func(i, j int) bool {
+ if out.Files[i].MinTXID == out.Files[j].MinTXID {
+ return out.Files[i].MaxTXID < out.Files[j].MaxTXID
+ }
+ return out.Files[i].MinTXID < out.Files[j].MinTXID
+ })
+ return out, nil
+}
+
+func (s *Server) pinnedDBReader(r *http.Request, g dbRequest, level int, lo, hi uint64) (*dbVersionReader, error) {
+ h, err := verifyDBHandle(s.dbHandleKey, r.Header.Get(dbdoor.HandleHeader), time.Now())
+ if err != nil {
+ return nil, err
+ }
+ holder := g.claim.Claim.Holder
+ if h.Host != g.host || h.Account != g.acct.Sqid || h.App != g.app || h.Location != holder.Location || h.Incarnation != holder.Incarnation || h.Key != dbVersionKey(h.Location, level, lo, hi) {
+ return nil, errDBHandle
+ }
+ return &dbVersionReader{store: s.cfg.DBVersions, bucket: s.cfg.HibernateBucket, handle: h}, nil
+}
+
+// This adapter keeps the existing stream budget, watchdog and audit path.
+// Its only object read names the signed version, including on every resume.
+type dbVersionReader struct {
+ store lease.VersionedStore
+ bucket string
+ handle dbFileHandle
+}
+
+func (v *dbVersionReader) LTXFiles(context.Context, int, ltx.TXID, bool) (ltx.FileIterator, error) {
+ return nil, errDBHandle
+}
+func (v *dbVersionReader) OpenLTXFile(ctx context.Context, level int, lo, hi ltx.TXID, offset, size int64) (io.ReadCloser, error) {
+ if v.handle.Key != dbVersionKey(v.handle.Location, level, uint64(lo), uint64(hi)) {
+ return nil, errDBHandle
+ }
+ n, err := dbVersionRange(v.handle.Size, offset, size)
+ if err != nil {
+ return nil, err
+ }
+ rc, err := v.store.OpenVersion(ctx, v.bucket, v.handle.Key, v.handle.VersionID)
+ if errors.Is(err, lease.ErrMirrorNotFound) {
+ return nil, os.ErrNotExist
+ }
+ if err != nil {
+ return nil, err
+ }
+ if _, err = io.CopyN(io.Discard, rc, offset); err != nil {
+ rc.Close()
+ return nil, err
+ }
+ return &dbRangeReader{Reader: io.LimitReader(rc, n), Closer: rc}, nil
+}
+
+type dbRangeReader struct {
+ io.Reader
+ io.Closer
+}
+
+func dbVersionRange(total, offset, size int64) (int64, error) {
+ if total < 0 || offset < 0 || size < 0 || offset > total || (offset == total && total != 0) {
+ return 0, errDBHandle
+ }
+ n := total - offset
+ if size != 0 && size < n {
+ n = size
+ }
+ return n, nil
+}
+func dbVersionHeaders(w http.ResponseWriter, r *http.Request, total, offset, size int64) error {
+ n, err := dbVersionRange(total, offset, size)
+ if err != nil {
+ return err
+ }
+ w.Header().Set("Content-Length", strconv.FormatInt(n, 10))
+ w.Header().Set(dbdoor.HandleHeader, r.Header.Get(dbdoor.HandleHeader))
+ if r.Header.Get("Range") != "" && n > 0 {
+ w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, offset+n-1, total))
+ }
+ return nil
+}
diff --git a/internal/console/apidbclaim_http_test.go b/internal/console/apidbclaim_http_test.go
new file mode 100644
index 000000000..bde54e1b0
--- /dev/null
+++ b/internal/console/apidbclaim_http_test.go
@@ -0,0 +1,275 @@
+package console
+
+import (
+ "context"
+ "database/sql"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "io/fs"
+ "net/http"
+ "os"
+ "path/filepath"
+ "strconv"
+ "strings"
+ "sync/atomic"
+ "testing"
+
+ lsfile "github.com/benbjohnson/litestream/file"
+
+ "github.com/carlosframework/platform/internal/dbdoor"
+ "github.com/carlosframework/platform/internal/dbdoor/dbdoortest"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/svcrole"
+)
+
+func enableDoorPlan(t *testing.T, fx *dbFx) *leasetest.VersionedBucket {
+ t.Helper()
+ rec, err := provision.GetRecord(t.Context(), fx.rel, "bab", "titogo", dbTestHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ inc, err := hostclaim.Incarnation("bab", "titogo", dbTestHost, rec.CreatedAt)
+ if err != nil {
+ t.Fatal(err)
+ }
+ c := hostclaim.Claim{V: 2, Host: dbTestHost, Gen: 1, Holder: &hostclaim.Holder{Family: "instance", Key: provision.RecordKey("bab", "titogo", dbTestHost), Account: "bab", App: "titogo", CreatedAt: rec.CreatedAt, Incarnation: inc, Location: "instances/" + dbTestHost}, Op: hostclaim.Op{Kind: hostclaim.KindServing}}
+ raw, _ := json.Marshal(c)
+ key, _ := hostclaim.Key(dbTestHost)
+ if _, err := fx.rel.Put(t.Context(), key, raw, ""); err != nil {
+ t.Fatal(err)
+ }
+ m := leasetest.NewVersionedBucket()
+ m.AddBucket("data-test")
+ fx.srv.cfg.HostClaim = true
+ fx.srv.cfg.DBVersions = m
+ secret, err := fx.st.SigningKey()
+ if err != nil {
+ t.Fatal(err)
+ }
+ fx.srv.dbHandleKey = dbHandleSecret(secret)
+ return m
+}
+
+func TestDoorPlanHTTPCompatibilityAndRestore(t *testing.T) {
+ for _, enabled := range []bool{false, true} {
+ t.Run(fmt.Sprint(enabled), func(t *testing.T) {
+ fx := dbFixture(t)
+ fx.replicate(t, dbTestHost, 3)
+ if enabled {
+ model := enableDoorPlan(t, fx)
+ root := filepath.Join(fx.root, "instances", dbTestHost, "ltx")
+ err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
+ if err != nil {
+ return err
+ }
+ if d.IsDir() {
+ return nil
+ }
+ rel, err := filepath.Rel(root, p)
+ if err != nil {
+ return err
+ }
+ parts := strings.Split(filepath.ToSlash(rel), "/")
+ if len(parts) != 2 {
+ return fmt.Errorf("unexpected file: %s", rel)
+ }
+ level, err := strconv.Atoi(parts[0])
+ if err != nil {
+ return err
+ }
+ b, err := os.ReadFile(p)
+ if err != nil {
+ return err
+ }
+ model.Put("data-test", dbLevelPrefix("instances/"+dbTestHost, level)+parts[1], b)
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+ do := func(ctx context.Context, tail string, h http.Header) (*http.Response, error) {
+ r, err := http.NewRequestWithContext(ctx, "GET", fx.url(dbdoor.Base("bab", "titogo", dbTestHost)+tail), nil)
+ if err != nil {
+ return nil, err
+ }
+ r.Header = h.Clone()
+ r.Header.Set("Authorization", "Bearer "+fx.member)
+ return http.DefaultClient.Do(r)
+ }
+ got, err := dbdoor.RestoreTo(t.Context(), dbdoor.NewClient(do), t.TempDir())
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer os.RemoveAll(got.Dir)
+ db, err := sql.Open("sqlite", "file:"+got.Path+"?mode=ro")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer db.Close()
+ var rows int
+ if err := db.QueryRow("SELECT count(*) FROM t").Scan(&rows); err != nil {
+ t.Fatal(err)
+ }
+ if rows != 3*dbdoortest.Rows {
+ t.Fatal(rows)
+ }
+ // The old CLI ignores additive fields and sends no handle.
+ res, err := do(t.Context(), dbdoor.ListTail(0), http.Header{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ var list dbdoor.ListResponse
+ if err := json.NewDecoder(res.Body).Decode(&list); err != nil {
+ t.Fatal(err)
+ }
+ res.Body.Close()
+ if len(list.Files) == 0 {
+ t.Fatal("empty L0")
+ }
+ f := list.Files[0]
+ res, err = do(t.Context(), dbdoor.GetTail(f.Level, f.MinTXID, f.MaxTXID), http.Header{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ io.Copy(io.Discard, res.Body)
+ res.Body.Close()
+ want := http.StatusOK
+ if enabled {
+ want = http.StatusBadRequest
+ }
+ if res.StatusCode != want {
+ t.Fatalf("old CLI: %d want %d", res.StatusCode, want)
+ }
+ if !enabled && list.Protocol != 0 {
+ t.Fatal("flag-off wire changed")
+ }
+ if enabled && fx.opens.Load() != 0 {
+ t.Fatal("pinned door opened unversioned replica")
+ }
+ })
+ }
+}
+
+func TestDoorPlanHTTPAuthorizationAndMissingClaim(t *testing.T) {
+ fx := dbFixture(t)
+ model := enableDoorPlan(t, fx)
+ model.FailOn = func(op, key string) error { t.Fatalf("unauthorized data access %s %s", op, key); return nil }
+ stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "door-plan-stranger")
+ status, h, body := rawReq(t, "GET", fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), stranger, "")
+ assertNotFound(t, "stranger", canonicalNotFound(t), status, h, body)
+ admin := serviceToken(t, fx.st, OpsAccountID, "door-plan-admin", "titogo", svcrole.Admin)
+ status, _, _ = rawReq(t, "GET", fx.url(dbdoor.ListPath("bab", "titogo", dbTestHost, 0)), admin, "")
+ if status != http.StatusForbidden {
+ t.Fatal(status)
+ }
+ // A second existing record with no claim must not auto-adopt here.
+ other := "other.go.tito.io"
+ putDBRecord(t, fx.consoleFixture, provision.Record{Host: other})
+ status, h, body = rawReq(t, "GET", fx.url(dbdoor.ListPath("bab", "titogo", other, 0)), fx.member, "")
+ if status != 409 || h.Get("Retry-After") != "1" || !strings.Contains(body, "host not adopted") {
+ t.Fatal(status, h, body)
+ }
+}
+
+func TestDoorPlanHTTPTruncationResumeAndFlagOff(t *testing.T) {
+ fx := dbFixture(t)
+ model := enableDoorPlan(t, fx)
+ key := dbVersionKey("instances/"+dbTestHost, 0, 1, 1)
+ model.Put("data-test", key, []byte("original"))
+ var opens atomic.Int32
+ fx.srv.cfg.DBVersions = doorOpenHook{VersionedStore: model, open: func(ctx context.Context, b, k, id string) (io.ReadCloser, error) {
+ rc, err := model.OpenVersion(ctx, b, k, id)
+ if err != nil {
+ return nil, err
+ }
+ if opens.Add(1) == 1 {
+ return &dbRangeReader{Reader: io.LimitReader(rc, 3), Closer: rc}, nil
+ }
+ return rc, nil
+ }}
+ do := func(tail string, h http.Header) *http.Response {
+ t.Helper()
+ r, err := http.NewRequest("GET", fx.url(dbdoor.Base("bab", "titogo", dbTestHost)+tail), nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r.Header = h.Clone()
+ r.Header.Set("Authorization", "Bearer "+fx.member)
+ res, err := http.DefaultClient.Do(r)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return res
+ }
+ res := do(dbdoor.ListTail(0), http.Header{})
+ var list dbdoor.ListResponse
+ if err := json.NewDecoder(res.Body).Decode(&list); err != nil {
+ t.Fatal(err)
+ }
+ res.Body.Close()
+ if len(list.Files) != 1 {
+ t.Fatal(list)
+ }
+ f := list.Files[0]
+ h := http.Header{}
+ h.Set(dbdoor.HandleHeader, f.Handle)
+ res = do(dbdoor.GetTail(0, 1, 1), h)
+ first, err := io.ReadAll(res.Body)
+ res.Body.Close()
+ if !errors.Is(err, io.ErrUnexpectedEOF) || string(first) != "ori" || res.Header.Get(dbdoor.HandleHeader) != f.Handle {
+ t.Fatal(string(first), err, res.Header)
+ }
+ model.Put("data-test", key, []byte("replacement"))
+ h.Set("Range", "bytes=3-")
+ res = do(dbdoor.GetTail(0, 1, 1), h)
+ rest, err := io.ReadAll(res.Body)
+ res.Body.Close()
+ if err != nil || res.StatusCode != 206 || string(append(first, rest...)) != "original" {
+ t.Fatal(string(rest), err, res.StatusCode)
+ }
+ events, err := fx.st.Events(OpsAccountID, "titogo", 10)
+ if err != nil {
+ t.Fatal(err)
+ }
+ ranged := false
+ for _, e := range events {
+ if e.Action == dbAuditAction && e.Target == dbTestHost && strings.Contains(e.Detail, "byte 3") {
+ ranged = true
+ }
+ }
+ if !ranged {
+ t.Fatal("missing pinned range audit", events)
+ }
+ // A flag-off console ignores the handle header exactly as today (R1):
+ // it serves the coordinate and never echoes the handle, so a v2 client
+ // rejects the response (Task 1's acknowledgement check).
+ // Keep startup configuration immutable while either server is running.
+ legacy := dbFixture(t)
+ legacy.replicate(t, dbTestHost, 1)
+ lc := lsfile.NewReplicaClient(filepath.Join(legacy.root, "instances", dbTestHost))
+ lf := listFiles(t, lc, 0)[0]
+ req, err := http.NewRequest("GET", legacy.url(dbdoor.GetPath("bab", "titogo", dbTestHost, 0, uint64(lf.MinTXID), uint64(lf.MaxTXID))), nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ req.Header = h.Clone()
+ req.Header.Del("Range") // a whole-file GET, so success is 200, not 206
+ req.Header.Set("Authorization", "Bearer "+legacy.member)
+ res, err = http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ io.Copy(io.Discard, res.Body)
+ res.Body.Close()
+ if res.StatusCode != 200 {
+ t.Fatal("flag-off changed its answer to a request carrying a handle header", res.StatusCode)
+ }
+ if res.Header.Get(dbdoor.HandleHeader) != "" {
+ t.Fatal("flag-off console echoed a handle it cannot enforce")
+ }
+}
diff --git a/internal/console/apidbclaim_test.go b/internal/console/apidbclaim_test.go
new file mode 100644
index 000000000..6c5639012
--- /dev/null
+++ b/internal/console/apidbclaim_test.go
@@ -0,0 +1,389 @@
+package console
+
+import (
+ "context"
+ "crypto/ecdsa"
+ "crypto/elliptic"
+ "crypto/rand"
+ "encoding/json"
+ "errors"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/dbdoor"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+type doorVersionHook struct {
+ lease.VersionedStore
+ after func()
+}
+
+func (v doorVersionHook) ListVersions(ctx context.Context, b, p string) ([]lease.ObjectVersion, error) {
+ out, err := v.VersionedStore.ListVersions(ctx, b, p)
+ if v.after != nil {
+ v.after()
+ }
+ return out, err
+}
+
+type doorReadFault struct {
+ release.Store
+ err error
+}
+
+func (s *doorReadFault) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if s.err != nil {
+ return nil, "", s.err
+ }
+ return s.Store.Get(ctx, key)
+}
+
+func doorPlanFixture(t *testing.T, legacy bool) (*Server, dbRequest, provision.Record, *leasetest.VersionedBucket) {
+ t.Helper()
+ rec := provision.Record{Host: "shop.example", Account: "bab", App: "jam"}
+ inc, err := hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
+ if err != nil {
+ t.Fatal(err)
+ }
+ loc := "instances/" + rec.Host
+ if !legacy {
+ loc += "/" + inc
+ }
+ claim := hostclaim.Claim{V: 2, Host: rec.Host, Gen: 1, Holder: &hostclaim.Holder{Family: "instance", Key: provision.RecordKey(rec.Account, rec.App, rec.Host), Account: rec.Account, App: rec.App, Incarnation: inc, Location: loc}, Op: hostclaim.Op{Kind: hostclaim.KindServing}}
+ store := release.NewDirStore(t.TempDir())
+ data, _ := json.Marshal(claim)
+ key, _ := hostclaim.Key(rec.Host)
+ if _, err := store.Put(t.Context(), key, data, ""); err != nil {
+ t.Fatal(err)
+ }
+ model := leasetest.NewVersionedBucket()
+ model.AddBucket("data")
+ s := &Server{cfg: Config{HostClaim: true, Releases: store, HibernateBucket: "data", DBVersions: model}, dbHandleKey: []byte(strings.Repeat("s", 32))}
+ g := dbRequest{acct: &Account{Sqid: "bab"}, app: "jam", host: rec.Host}
+ g.claim, err = s.dbClaim(t.Context(), g, rec)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return s, g, rec, model
+}
+
+func writeDoorClaim(t *testing.T, s *Server, g dbRequest, c *hostclaim.Claim) {
+ t.Helper()
+ key, _ := hostclaim.Key(g.host)
+ _, etag, err := s.cfg.Releases.Get(t.Context(), key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ b, err := json.Marshal(c)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err = s.cfg.Releases.Put(t.Context(), key, b, etag); err != nil {
+ t.Fatal(err)
+ }
+}
+func cloneDoorClaim(c *hostclaim.Claim) *hostclaim.Claim {
+ b, _ := json.Marshal(c)
+ var out hostclaim.Claim
+ _ = json.Unmarshal(b, &out)
+ return &out
+}
+
+func TestDoorPlanHandles(t *testing.T) {
+ key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+ if err != nil {
+ t.Fatal(err)
+ }
+ secret := dbHandleSecret(key)
+ h := dbFileHandle{V: 1, Host: "shop.example", Location: "instances/shop.example/inc", Key: "instances/shop.example/inc/0000/a", VersionID: "v+/?", Incarnation: "inc", Expiry: 200, Account: "bab", App: "jam", Size: 6}
+ token, err := signDBHandle(secret, h)
+ if err != nil {
+ t.Fatal(err)
+ }
+ got, err := verifyDBHandle(dbHandleSecret(key), token, time.Unix(199, 0))
+ if err != nil || got != h {
+ t.Fatal(got, err)
+ }
+ for _, bad := range []string{"", token + "x", "x" + token, strings.Repeat("x", 8193)} {
+ if _, err := verifyDBHandle(secret, bad, time.Unix(199, 0)); !errors.Is(err, errDBHandle) {
+ t.Fatal(bad, err)
+ }
+ }
+ if _, err := verifyDBHandle([]byte(strings.Repeat("x", 32)), token, time.Unix(199, 0)); !errors.Is(err, errDBHandle) {
+ t.Fatal(err)
+ }
+ if _, err := verifyDBHandle(secret, token, time.Unix(200, 0)); !errors.Is(err, errDBHandleExpired) {
+ t.Fatal(err)
+ }
+}
+
+func TestDoorPlanListingAndPinnedOpen(t *testing.T) {
+ for _, legacy := range []bool{false, true} {
+ t.Run(map[bool]string{true: "legacy", false: "scoped"}[legacy], func(t *testing.T) {
+ s, g, _, m := doorPlanFixture(t, legacy)
+ loc := g.claim.Claim.Holder.Location
+ key := dbVersionKey(loc, 0, 1, 1)
+ m.Put("data", key, []byte("older"))
+ id := m.Put("data", key, []byte("original"))
+ deleted := dbVersionKey(loc, 0, 2, 2)
+ m.Put("data", deleted, []byte("hidden"))
+ m.BareDelete("data", deleted)
+ m.Put("data", loc+"/lock.json", []byte("secret"))
+ m.Put("data", dbVersionKey(loc+"/other-incarnation", 0, 3, 3), []byte("other"))
+ m.Put("data", dbVersionKey(loc+"-sibling", 0, 4, 4), []byte("sibling"))
+ m.Put("data", dbVersionKey(loc+"/g-old", 0, 5, 5), []byte("generation"))
+ m.PageSize = 1
+ out, err := s.listDBVersions(t.Context(), g, 0)
+ if err != nil || out.Protocol != 2 || len(out.Files) != 1 {
+ t.Fatalf("%+v %v", out, err)
+ }
+ token := out.Files[0].Handle
+ h, err := verifyDBHandle(s.dbHandleKey, token, time.Now())
+ if err != nil || h.VersionID != id || h.Location != loc {
+ t.Fatal(h, err)
+ }
+ req := httptest.NewRequest(http.MethodGet, "/", nil)
+ req.Header.Set(dbdoor.HandleHeader, token)
+ rd, err := s.pinnedDBReader(req, g, 0, 1, 1)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // Stall after admission, then overwrite the identical legacy key and
+ // advance the holder to a new incarnation before the physical open.
+ m.Put("data", key, []byte("replacement"))
+ next := cloneDoorClaim(g.claim.Claim)
+ next.Gen++
+ next.Holder.CreatedAt = time.Unix(1, 0)
+ next.Holder.Incarnation, _ = hostclaim.Incarnation("bab", "jam", g.host, next.Holder.CreatedAt)
+ next.Holder.Location = "instances/" + g.host + "/" + next.Holder.Incarnation
+ writeDoorClaim(t, s, g, next)
+ rc, err := rd.OpenLTXFile(t.Context(), 0, 1, 1, 2, 3)
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, err := io.ReadAll(rc)
+ rc.Close()
+ if err != nil || string(body) != "igi" {
+ t.Fatalf("%q %v", body, err)
+ }
+ g2 := g
+ g2.claim.Claim = next
+ if _, err := s.pinnedDBReader(req, g2, 0, 1, 1); err == nil {
+ t.Fatal("new incarnation accepted old handle")
+ }
+ bad := g
+ bad.host = "other.example"
+ if _, err := s.pinnedDBReader(req, bad, 0, 1, 1); err == nil {
+ t.Fatal("cross host")
+ }
+ if _, err := s.pinnedDBReader(req, g, 0, 2, 2); err == nil {
+ t.Fatal("cross coordinate")
+ }
+ if err := m.DeleteVersion(t.Context(), "data", key, id); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := rd.OpenLTXFile(t.Context(), 0, 1, 1, 0, 0); !errors.Is(err, os.ErrNotExist) {
+ t.Fatal(err)
+ }
+ })
+ }
+}
+
+func TestDoorPlanBracket(t *testing.T) {
+ for _, change := range []string{"holder", "op", "same-gen-op", "chore", "absent", "malformed", "read-error"} {
+ t.Run(change, func(t *testing.T) {
+ s, g, _, m := doorPlanFixture(t, false)
+ m.Put("data", dbVersionKey(g.claim.Claim.Holder.Location, 0, 1, 1), []byte("old"))
+ fault := &doorReadFault{Store: s.cfg.Releases}
+ s.cfg.Releases = fault
+ s.cfg.DBVersions = doorVersionHook{VersionedStore: m, after: func() {
+ next := cloneDoorClaim(g.claim.Claim)
+ next.Gen++
+ switch change {
+ case "holder":
+ next.Holder.Stamp = next.Gen
+ case "op":
+ next.Op = hostclaim.Op{Kind: hostclaim.KindDeleting, ID: "delete"}
+ case "same-gen-op":
+ next.Gen--
+ next.Op.Exec = "another-box"
+ case "chore":
+ next.Touched.Complete = true
+ case "absent":
+ fault.err = os.ErrNotExist
+ return
+ case "read-error":
+ fault.err = errors.New("offline")
+ return
+ case "malformed":
+ next.V = 999
+ }
+ writeDoorClaim(t, s, g, next)
+ }}
+ out, err := s.listDBVersions(t.Context(), g, 0)
+ if change == "chore" {
+ if err != nil || len(out.Files) != 1 {
+ t.Fatal(out, err)
+ }
+ } else if err == nil || len(out.Files) != 0 {
+ t.Fatalf("partial/success: %+v %v", out, err)
+ }
+ })
+ }
+}
+
+func TestDoorPlanClaimRefusals(t *testing.T) {
+ for _, change := range []string{"absent", "free", "deleting", "parking", "purging", "owner", "incarnation", "location", "generation", "malformed", "read-error"} {
+ t.Run(change, func(t *testing.T) {
+ s, g, rec, m := doorPlanFixture(t, true)
+ m.FailOn = func(op, key string) error { t.Fatalf("refusal reached data bucket: %s %s", op, key); return nil }
+ c := cloneDoorClaim(g.claim.Claim)
+ want := errDBClaimClosed
+ switch change {
+ case "absent":
+ s.cfg.Releases = release.NewDirStore(t.TempDir())
+ want = errDBNotAdopted
+ case "free":
+ c.Holder = nil
+ c.Op = hostclaim.Op{}
+ case "deleting", "parking", "purging":
+ c.Op = hostclaim.Op{Kind: change, ID: "x"}
+ case "owner":
+ g.app = "other"
+ want = os.ErrNotExist
+ case "incarnation":
+ rec.CreatedAt = time.Unix(1, 0)
+ want = os.ErrNotExist
+ case "location":
+ c.Holder.Location = ""
+ want = hostclaim.ErrMalformed
+ case "generation":
+ c.Holder.Generation = "g-current"
+ want = errDBGeneration
+ case "malformed":
+ c.V = 99
+ want = hostclaim.ErrMalformed
+ case "read-error":
+ want = errors.New("offline")
+ s.cfg.Releases = &doorReadFault{Store: s.cfg.Releases, err: want}
+ }
+ if change != "absent" && change != "read-error" {
+ writeDoorClaim(t, s, dbRequest{host: rec.Host}, c)
+ }
+ if _, err := s.dbClaim(t.Context(), g, rec); !errors.Is(err, want) {
+ t.Fatalf("%v want %v", err, want)
+ }
+ })
+ }
+}
+
+func TestDoorPlanListingFailureAndRanges(t *testing.T) {
+ s, g, _, m := doorPlanFixture(t, true)
+ for _, lo := range []uint64{1, 2, 3} {
+ m.Put("data", dbVersionKey(g.claim.Claim.Holder.Location, 0, lo, lo), []byte("sixsix"))
+ }
+ m.PageSize = 1
+ m.FailListAfterPages = 1
+ if out, err := s.listDBVersions(t.Context(), g, 0); err == nil || len(out.Files) != 0 {
+ t.Fatal(out, err)
+ }
+ for _, tc := range []struct {
+ offset, size, want int64
+ ok bool
+ }{{0, 0, 6, true}, {2, 0, 4, true}, {2, 2, 2, true}, {2, 100, 4, true}, {6, 0, 0, false}, {-1, 0, 0, false}, {0, -1, 0, false}} {
+ n, err := dbVersionRange(6, tc.offset, tc.size)
+ if (err == nil) != tc.ok || n != tc.want {
+ t.Fatal(tc, n, err)
+ }
+ }
+ w := httptest.NewRecorder()
+ r := httptest.NewRequest("GET", "/", nil)
+ r.Header.Set("Range", "bytes=2-3")
+ r.Header.Set(dbdoor.HandleHeader, "signed")
+ if err := dbVersionHeaders(w, r, 6, 2, 2); err != nil {
+ t.Fatal(err)
+ }
+ if w.Header().Get("Content-Length") != "2" || w.Header().Get("Content-Range") != "bytes 2-3/6" || w.Header().Get(dbdoor.HandleHeader) != "signed" {
+ t.Fatal(w.Header())
+ }
+}
+
+type doorOpenHook struct {
+ lease.VersionedStore
+ open func(context.Context, string, string, string) (io.ReadCloser, error)
+}
+
+func (v doorOpenHook) OpenVersion(ctx context.Context, b, k, id string) (io.ReadCloser, error) {
+ return v.open(ctx, b, k, id)
+}
+
+type doorTrackedReader struct {
+ io.Reader
+ closed bool
+}
+
+func (r *doorTrackedReader) Close() error { r.closed = true; return nil }
+
+type doorCancelledReader struct{ ctx context.Context }
+
+func (r doorCancelledReader) Read([]byte) (int, error) { return 0, r.ctx.Err() }
+
+func TestDoorPlanRangeReaderCloses(t *testing.T) {
+ for _, mode := range []string{"complete", "short-skip", "cancelled-skip"} {
+ t.Run(mode, func(t *testing.T) {
+ s, g, _, m := doorPlanFixture(t, true)
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ tracked := &doorTrackedReader{Reader: strings.NewReader("abcdef")}
+ if mode == "short-skip" {
+ tracked.Reader = strings.NewReader("a")
+ }
+ if mode == "cancelled-skip" {
+ cancel()
+ tracked.Reader = doorCancelledReader{ctx: ctx}
+ }
+ key := dbVersionKey(g.claim.Claim.Holder.Location, 0, 1, 1)
+ s.cfg.DBVersions = doorOpenHook{VersionedStore: m, open: func(ctx context.Context, b, k, id string) (io.ReadCloser, error) {
+ if b != "data" || k != key || id != "selected" {
+ t.Fatal(b, k, id)
+ }
+ return tracked, nil
+ }}
+ rd := &dbVersionReader{store: s.cfg.DBVersions, bucket: "data", handle: dbFileHandle{Location: g.claim.Claim.Holder.Location, Key: key, VersionID: "selected", Size: 6}}
+ rc, err := rd.OpenLTXFile(ctx, 0, 1, 1, 2, 2)
+ if mode != "complete" {
+ if err == nil || !tracked.closed {
+ t.Fatal(err, tracked.closed)
+ }
+ if mode == "cancelled-skip" && !errors.Is(err, context.Canceled) {
+ t.Fatal(err)
+ }
+ return
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ b, err := io.ReadAll(rc)
+ if err != nil || string(b) != "cd" {
+ t.Fatal(string(b), err)
+ }
+ if tracked.closed {
+ t.Fatal("closed before caller finished")
+ }
+ rc.Close()
+ if !tracked.closed {
+ t.Fatal("reader leaked")
+ }
+ })
+ }
+}
diff --git a/internal/console/server.go b/internal/console/server.go
index 01c315e6e..44994910b 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -13,6 +13,7 @@ import (
"encoding/json"
"errors"
"fmt"
+ "github.com/carlosframework/platform/internal/lease"
"html/template"
"log"
"math"
@@ -455,8 +456,10 @@ const (
// Config is everything the server reads its world through.
type Config struct {
- Origin string // the console's public origin — the OAuth client_id
- Keymail string // keymail base URL
+ HostClaim bool
+ DBVersions lease.VersionedStore
+ Origin string // the console's public origin — the OAuth client_id
+ Keymail string // keymail base URL
// Registry is the path to the platform registry DB. Read-mostly (routes,
// domains, alerts), but fleets.go's fleet/box/placement doors (customer
// fleets plan, Task 7) genuinely write through it via openRegistry
@@ -741,8 +744,9 @@ type Config struct {
// Server is the console's http.Handler.
type Server struct {
- cfg Config
- store *Store
+ dbHandleKey []byte
+ cfg Config
+ store *Store
// appRegionOf and beforeClaimInsert are TEST SEAMS (nil in
// production): a store failure on the app-region read (C20), and a
// competing insert landing after the region was resolved.
@@ -1191,6 +1195,9 @@ func NewServer(cfg Config, st *Store) (*Server, error) {
// CLI identity: the device-code login `carlos auth login` drives, the
// approval page, and the tokens an owner can see and kill (authcli.go).
// The db door's production replica reader (apidb.go); tests repoint it.
+ if cfg.HostClaim {
+ s.dbHandleKey = dbHandleSecret(key)
+ }
s.dbReplica = s.cachedReadReplica
s.dbStreams = newDBStreamLimits(dbMaxStreams, dbMaxStreamsPerPrincipal)
s.dbStreamIdle = dbStreamIdleDefault