mirror of
https://github.com/hanzoai/replicate.git
synced 2026-10-30 18:23:20 +00:00
* file: add file:// Leaser for filesystem-backed CAS Mirrors the S3 Leaser CAS semantics using POSIX O_EXCL for initial creation and content-hash ETag matching for renew/release. Atomic rename for the cutover. Same Leaser interface contract — callers swap backends without code changes. Use cases: per-process / single-node leases for dev + tests, container-local coordination where S3 is overkill, and cross-process tests of leader/standby patterns without spinning up MinIO. Tests cover: fresh acquire, expired-lease takeover, contended acquire (LeaseExistsError), renew bumps expiry + etag without touching generation, renew loses race to a stealer (ErrLeaseNotHeld), release roundtrip, release after steal (ErrLeaseNotHeld), and concurrent acquire with exactly one winner across 32 goroutines. * file: harden Leaser — MaxTTL clamp + foreign-owner renew guard Closes red findings #1 (HIGH) and #4 (MEDIUM) on the per-market shard layer's lease coordinator. #1 — Server-side MaxTTL clamp: - Add Leaser.MaxTTL (default 30s when zero). - Acquire + Renew reject TTL > MaxTTL with ErrTTLExceedsMax. Without this, one malicious or buggy caller pins the lease for an arbitrary duration regardless of liveness. - readLease rejects an on-disk ExpiresAt > now + MaxTTL with ErrLeaseCorrupt — defense in depth against a writer that bypassed the API. #4 — Owner check in RenewLease: - ETag CAS alone permits a caller who read another Leaser's ETag from disk to craft a renew call carrying any Owner string. We now read the on-disk record first and reject when existing.Owner != l.Owner with ErrLeaseNotHeld. ETag CAS remains primary; Owner check is belt-and-suspenders. Tests: - TestLeaser_RejectsTTLBeyondMax (acquire, renew, zero MaxTTL falls back to default) - TestLeaser_ReadRejectsCorruptedTTL (hand-write 1h ExpiresAt, Acquire rejects) - TestLeaser_RejectsRenewByForeignOwner (foreign Owner with stolen ETag → ErrLeaseNotHeld) All existing leaser tests continue to pass. * leaser: surface MaxLeaseTTL() on the interface Heartbeat schedulers depend on the leaser's TTL ceiling to validate that their renew cadence is at least 2x tighter than the server-side bound. Without surfacing it, callers either hardcode the bound or compute it from the leaser type — both brittle. The struct field stays MaxTTL (configuration knob); the interface method is MaxLeaseTTL() to avoid the Go name-collision rule ("field and method with the same name"). file:// and s3:// both default to DefaultLeaseMaxTTL (30s) when unset. * dblock: expose SQLite-compatible byte-range lock as a public package External callers that mirror Replica.applyLTXFile (e.g. shard-layer exclusive-lock primitive so their apply pipeline interlocks with SQLite's own locking and with the upstream applier. dblock.LockExclusive / dblock.Unlock target the SQLite pending-byte (0x40000000) and shared-region addresses, matching the existing internal.LockFileExclusive contract used by Replica.applyLTXFile. The internal/lock_*.go shims now forward to dblock so there is one canonical implementation across the module. Behavior identical to the previous internal/ implementation — file-level tests in file/ pass. * chore: bump go to 1.26.4 Picks up upstream security fixes for crypto/x509, mime, net/textproto, crypto/fips140. No dependency bumps — only the toolchain directive. go test -race ./... is green. * chore: fix gofmt + goimports drift across repo Pre-existing formatting drift on main exposed by go-imports-repo pre-commit hook. Pure whitespace alignment + import ordering — no logic changes. Touches 16 files unrelated to the file:// Leaser PR.
216 lines
6.4 KiB
Go
216 lines
6.4 KiB
Go
package replicate_test
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"io"
|
|
"testing"
|
|
|
|
"github.com/hanzoai/ltx"
|
|
"github.com/luxfi/age"
|
|
|
|
"github.com/hanzoai/replicate"
|
|
"github.com/hanzoai/replicate/mock"
|
|
)
|
|
|
|
// TestReplicaEncryptDecryptRoundTrip writes data through the Replica's encrypt
|
|
// path (WriteLTXFile), then reads it back through the decrypt path
|
|
// (OpenLTXFile) and verifies the bytes match.
|
|
func TestReplicaEncryptDecryptRoundTrip(t *testing.T) {
|
|
identity, err := age.GenerateX25519Identity()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
recipient := identity.Recipient()
|
|
|
|
plaintext := []byte("test WAL segment data — roundtrip verification")
|
|
|
|
// Capture encrypted bytes written by Replica.WriteLTXFile.
|
|
var stored bytes.Buffer
|
|
client := &mock.ReplicaClient{
|
|
WriteLTXFileFunc: func(_ context.Context, _ int, _, _ ltx.TXID, r io.Reader) (*ltx.FileInfo, error) {
|
|
if _, err := io.Copy(&stored, r); err != nil {
|
|
return nil, err
|
|
}
|
|
return <x.FileInfo{MinTXID: 1, MaxTXID: 1}, nil
|
|
},
|
|
OpenLTXFileFunc: func(_ context.Context, _ int, _, _ ltx.TXID, _, _ int64) (io.ReadCloser, error) {
|
|
return io.NopCloser(bytes.NewReader(stored.Bytes())), nil
|
|
},
|
|
}
|
|
|
|
r := replicate.NewReplica(nil)
|
|
r.Client = client
|
|
r.AgeRecipients = []age.Recipient{recipient}
|
|
r.AgeIdentities = []age.Identity{identity}
|
|
|
|
// Write through encrypt path.
|
|
if _, err := r.WriteLTXFile(context.Background(), 0, 1, 1, bytes.NewReader(plaintext)); err != nil {
|
|
t.Fatalf("WriteLTXFile: %v", err)
|
|
}
|
|
|
|
// Read through decrypt path.
|
|
rc, err := r.OpenLTXFile(context.Background(), 0, 1, 1, 0, 0)
|
|
if err != nil {
|
|
t.Fatalf("OpenLTXFile: %v", err)
|
|
}
|
|
defer rc.Close()
|
|
|
|
decrypted, err := io.ReadAll(rc)
|
|
if err != nil {
|
|
t.Fatalf("ReadAll: %v", err)
|
|
}
|
|
|
|
if !bytes.Equal(plaintext, decrypted) {
|
|
t.Fatalf("roundtrip mismatch:\n want: %q\n got: %q", plaintext, decrypted)
|
|
}
|
|
}
|
|
|
|
// TestReplicaEncryptedOutputDiffers verifies that the encrypted output stored
|
|
// by WriteLTXFile is not equal to the original plaintext.
|
|
func TestReplicaEncryptedOutputDiffers(t *testing.T) {
|
|
identity, err := age.GenerateX25519Identity()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
recipient := identity.Recipient()
|
|
|
|
plaintext := []byte("plaintext that must not appear in ciphertext")
|
|
|
|
var stored bytes.Buffer
|
|
client := &mock.ReplicaClient{
|
|
WriteLTXFileFunc: func(_ context.Context, _ int, _, _ ltx.TXID, r io.Reader) (*ltx.FileInfo, error) {
|
|
if _, err := io.Copy(&stored, r); err != nil {
|
|
return nil, err
|
|
}
|
|
return <x.FileInfo{MinTXID: 1, MaxTXID: 1}, nil
|
|
},
|
|
}
|
|
|
|
r := replicate.NewReplica(nil)
|
|
r.Client = client
|
|
r.AgeRecipients = []age.Recipient{recipient}
|
|
|
|
if _, err := r.WriteLTXFile(context.Background(), 0, 1, 1, bytes.NewReader(plaintext)); err != nil {
|
|
t.Fatalf("WriteLTXFile: %v", err)
|
|
}
|
|
|
|
if bytes.Equal(plaintext, stored.Bytes()) {
|
|
t.Fatal("encrypted output equals plaintext — encryption did not transform the data")
|
|
}
|
|
|
|
// Also verify plaintext does not appear as a substring of the ciphertext.
|
|
if bytes.Contains(stored.Bytes(), plaintext) {
|
|
t.Fatal("plaintext found as substring of ciphertext")
|
|
}
|
|
}
|
|
|
|
// TestReplicaDecryptWithWrongKey verifies that decrypting with a different
|
|
// identity than the one used for encryption fails.
|
|
func TestReplicaDecryptWithWrongKey(t *testing.T) {
|
|
encIdentity, err := age.GenerateX25519Identity()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
wrongIdentity, err := age.GenerateX25519Identity()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
plaintext := []byte("secret data encrypted for a specific recipient")
|
|
|
|
var stored bytes.Buffer
|
|
client := &mock.ReplicaClient{
|
|
WriteLTXFileFunc: func(_ context.Context, _ int, _, _ ltx.TXID, r io.Reader) (*ltx.FileInfo, error) {
|
|
if _, err := io.Copy(&stored, r); err != nil {
|
|
return nil, err
|
|
}
|
|
return <x.FileInfo{MinTXID: 1, MaxTXID: 1}, nil
|
|
},
|
|
OpenLTXFileFunc: func(_ context.Context, _ int, _, _ ltx.TXID, _, _ int64) (io.ReadCloser, error) {
|
|
return io.NopCloser(bytes.NewReader(stored.Bytes())), nil
|
|
},
|
|
}
|
|
|
|
// Write with the correct recipient.
|
|
rw := replicate.NewReplica(nil)
|
|
rw.Client = client
|
|
rw.AgeRecipients = []age.Recipient{encIdentity.Recipient()}
|
|
|
|
if _, err := rw.WriteLTXFile(context.Background(), 0, 1, 1, bytes.NewReader(plaintext)); err != nil {
|
|
t.Fatalf("WriteLTXFile: %v", err)
|
|
}
|
|
|
|
// Read with the wrong identity.
|
|
rr := replicate.NewReplica(nil)
|
|
rr.Client = client
|
|
rr.AgeIdentities = []age.Identity{wrongIdentity}
|
|
|
|
rc, err := rr.OpenLTXFile(context.Background(), 0, 1, 1, 0, 0)
|
|
if err == nil {
|
|
// If OpenLTXFile didn't fail, reading should fail.
|
|
_, readErr := io.ReadAll(rc)
|
|
rc.Close()
|
|
if readErr == nil {
|
|
t.Fatal("expected error decrypting with wrong key, got none")
|
|
}
|
|
}
|
|
// err != nil is the expected case — age.Decrypt returns an error.
|
|
}
|
|
|
|
// TestReplicaNoEncryption verifies that without age keys configured, data
|
|
// passes through WriteLTXFile and OpenLTXFile unmodified.
|
|
func TestReplicaNoEncryption(t *testing.T) {
|
|
plaintext := []byte("unencrypted WAL data should pass through verbatim")
|
|
|
|
var stored bytes.Buffer
|
|
client := &mock.ReplicaClient{
|
|
WriteLTXFileFunc: func(_ context.Context, _ int, _, _ ltx.TXID, r io.Reader) (*ltx.FileInfo, error) {
|
|
if _, err := io.Copy(&stored, r); err != nil {
|
|
return nil, err
|
|
}
|
|
return <x.FileInfo{MinTXID: 1, MaxTXID: 1}, nil
|
|
},
|
|
OpenLTXFileFunc: func(_ context.Context, _ int, _, _ ltx.TXID, _, _ int64) (io.ReadCloser, error) {
|
|
return io.NopCloser(bytes.NewReader(stored.Bytes())), nil
|
|
},
|
|
}
|
|
|
|
r := replicate.NewReplica(nil)
|
|
r.Client = client
|
|
// No AgeRecipients, no AgeIdentities — encryption disabled.
|
|
|
|
if r.EncryptionEnabled() {
|
|
t.Fatal("EncryptionEnabled should be false with no recipients")
|
|
}
|
|
if r.DecryptionEnabled() {
|
|
t.Fatal("DecryptionEnabled should be false with no identities")
|
|
}
|
|
|
|
// Write.
|
|
if _, err := r.WriteLTXFile(context.Background(), 0, 1, 1, bytes.NewReader(plaintext)); err != nil {
|
|
t.Fatalf("WriteLTXFile: %v", err)
|
|
}
|
|
|
|
// Stored data should be identical to plaintext (no encryption applied).
|
|
if !bytes.Equal(plaintext, stored.Bytes()) {
|
|
t.Fatalf("stored data differs from plaintext when encryption is disabled:\n want: %q\n got: %q", plaintext, stored.Bytes())
|
|
}
|
|
|
|
// Read.
|
|
rc, err := r.OpenLTXFile(context.Background(), 0, 1, 1, 0, 0)
|
|
if err != nil {
|
|
t.Fatalf("OpenLTXFile: %v", err)
|
|
}
|
|
defer rc.Close()
|
|
|
|
result, err := io.ReadAll(rc)
|
|
if err != nil {
|
|
t.Fatalf("ReadAll: %v", err)
|
|
}
|
|
|
|
if !bytes.Equal(plaintext, result) {
|
|
t.Fatalf("passthrough mismatch:\n want: %q\n got: %q", plaintext, result)
|
|
}
|
|
}
|