From df99de8063c60b253a097ebf4bee66e0c3b7f4d5 Mon Sep 17 00:00:00 2001 From: Martin Hutchinson Date: Wed, 6 Aug 2025 15:03:12 +0000 Subject: [PATCH] [Vindex] Added Output Log to commit to map state The Output Log is a POSIX Tessera log in the provided demo, but can be changed for any other implementation that matches the OutputLog interface. The OutputLog is updated any time the vindex is updated. The value written to the new leaf is the concatenation of: 1. hex encoding of the map root 2. a couple of newline characters 3. the raw checkpoint from the InputLog at which the vindex was built This includes a refactoring to push the checkpoint parsing into the Input and Output Log definitions to improve encapsulation. This unlocks what we need to start generating fully comprehensive proofs for Lookup operations. --- vindex/README.md | 10 +++- vindex/cmd/logandmap/main.go | 107 ++++++++++++++++++++++++++--------- vindex/map.go | 94 +++++++++++++++++++++++++----- vindex/map_test.go | 21 +++++-- vindex/outputlog.go | 75 ++++++++++++++++++++++++ 5 files changed, 258 insertions(+), 49 deletions(-) create mode 100644 vindex/outputlog.go diff --git a/vindex/README.md b/vindex/README.md index cdaf7a4..8834841 100644 --- a/vindex/README.md +++ b/vindex/README.md @@ -191,7 +191,7 @@ In this repository, there is a demo of running a [tlog-tiles][] log using [Tesse Below are instructions for running this demo with sample key material: ```shell -LOG_PRIVATE_KEY=PRIVATE+KEY+logandmap+38581672+AXJ0FKWOcO2ch6WC8kP705Ed3Gxu7pVtZLhfHAQwp+FE; go run ./vindex/cmd/logandmap --input_log_dir ~/logandmap/inputlog/ --walPath ~/logandmap/map.wal +INPUT_LOG_PRIVATE_KEY=PRIVATE+KEY+example.com/inputlog+bd6268fb+ATPZW5UsUYHJo24lwgK1ykm9VafhyUtUxX5evV4ZIokY OUTPUT_LOG_PRIVATE_KEY=PRIVATE+KEY+example.com/outputlog+07392c46+ATPJ4crkyUbPeaRffN/4NUof3KV0pQznVIPGOQm3SDEJ go run ./vindex/cmd/logandmap --storage_dir ~/logandmap/ ``` Running the above will run a web server hosting the following URLs: @@ -203,7 +203,11 @@ The input log has entries for packages in the set {`foo`, `bar`, `baz`, `splat`} To inspect the log, you can use the woodpecker tool (using the corresponding public key to the private key used above): ```shell -go run github.com/mhutchinson/woodpecker@main --custom_log_type=tiles --custom_log_url=http://localhost:8088/inputlog --custom_log_vkey=logandmap+38581672+Ab/PCr1eCclRPRMBqw/r5An1xO71MCnImLiospEq6b4l +# To inspect the Input Log +go run github.com/mhutchinson/woodpecker@main --custom_log_type=tiles --custom_log_url=http://localhost:8088/inputlog/ --custom_log_vkey=example.com/inputlog+bd6268fb+AWdGkrHKBm+pOubTrcBTV8JMDLFlF1Y8WUH1nrtLNXDr + +# To inspect the Output Log +go run github.com/mhutchinson/woodpecker@main --custom_log_type=tiles --custom_log_url=http://localhost:8088/outputlog/ --custom_log_vkey=example.com/outputlog+07392c46+AWyS8y8ZsRmQnTr6Fr2knaa8+t6CPYFh5Ho3wJEr14B8 ``` Use left/right cursor to browse, and `q` to quit. @@ -222,7 +226,7 @@ go run ./vindex/cmd/client --base_url http://localhost:8088/vindex/ --lookup=foo | 2 | Implementation of in-memory Merkle Radix Tree | ✅ | | 3 | Incremental update | ✅ | | 4 | Verify that mapped data matches Input Log Checkpoint | ✅ | -| 5 | Output log | ❌ | +| 5 | Output log | ✅ | | 6 | Proofs served on Lookup | ❌ | | 7 | Storage backed verifiable-map | ❌ | | 8 | MapFn defined in WASM | ❌ | diff --git a/vindex/cmd/logandmap/main.go b/vindex/cmd/logandmap/main.go index 331e9a6..2e71340 100644 --- a/vindex/cmd/logandmap/main.go +++ b/vindex/cmd/logandmap/main.go @@ -50,9 +50,10 @@ import ( ) var ( - inputLogPrivKeyFile = flag.String("input_log_private_key", "", "Location of private key file. If unset, uses the contents of the INPUT_LOG_PRIVATE_KEY environment variable.") - storageDir = flag.String("storage_dir", "", "Root directory in which to store the data for the demo. This will create subdirectories for the Input Log, Output Log, and allocate space to store the verifiable map persistence.") - listen = flag.String("listen", ":8088", "Address to set up HTTP server listening on") + inputLogPrivKeyFile = flag.String("input_log_private_key", "", "Location of private key file. If unset, uses the contents of the INPUT_LOG_PRIVATE_KEY environment variable.") + outputLogPrivKeyFile = flag.String("output_log_private_key", "", "Location of private key file. If unset, uses the contents of the OUTPUT_LOG_PRIVATE_KEY environment variable.") + storageDir = flag.String("storage_dir", "", "Root directory in which to store the data for the demo. This will create subdirectories for the Input Log, Output Log, and allocate space to store the verifiable map persistence.") + listen = flag.String("listen", ":8088", "Address to set up HTTP server listening on") ) func main() { @@ -78,22 +79,48 @@ func run(ctx context.Context) error { return errors.New("storage_dir must be set") } inputLogDir := path.Join(*storageDir, "inputlog") + outputLogDir := path.Join(*storageDir, "outputlog") mapRoot := path.Join(*storageDir, "vindex") if err := os.MkdirAll(inputLogDir, 0o755); err != nil { return fmt.Errorf("failed to create input log directory: %v", err) } + if err := os.MkdirAll(outputLogDir, 0o755); err != nil { + return fmt.Errorf("failed to create output log directory: %v", err) + } if err := os.MkdirAll(mapRoot, 0o755); err != nil { return fmt.Errorf("failed to create vindex directory: %v", err) } + inputLog, inputCloser := inputLogOrDie(ctx, inputLogDir) + defer inputCloser() + + outputLog, outputCloser := outputLogOrDie(ctx, outputLogDir) + defer outputCloser() + + vi, err := vindex.NewVerifiableIndex(ctx, inputLog, mapFnFromFlags(), outputLog, mapRoot) + if err != nil { + return fmt.Errorf("failed to create vindex: %v", err) + } + + // Keeps the map synced with the latest published log state. + go maintainMap(ctx, vi) + + // Run a web server to handle queries over the verifiable index. + go runWebServer(vi, inputLogDir, outputLogDir) + <-ctx.Done() + return nil +} + +// inputLogOrDie returns an input log that is being updated periodically. +func inputLogOrDie(ctx context.Context, inputLogDir string) (log logReaderSource, closer func()) { // Gather the info needed for reading/writing checkpoints ils, ilv := getInputLogSignerVerifierOrDie() // Set up a Tessera POSIX log ild, err := posix.New(ctx, posix.Config{Path: inputLogDir}) if err != nil { - return fmt.Errorf("failed to create new log: %v", err) + klog.Exit(fmt.Errorf("failed to create input log: %v", err)) } inputAppender, inputShutdown, inputReader, err := tessera.NewAppender(ctx, ild, tessera.NewAppendOptions(). @@ -101,47 +128,37 @@ func run(ctx context.Context) error { WithCheckpointInterval(5*time.Second). WithBatching(256, time.Second)) if err != nil { - return fmt.Errorf("failed to get appender: %v", err) + klog.Exit(fmt.Errorf("failed to get appender: %v", err)) } - defer func() { - _ = inputShutdown(ctx) - }() - // Create the verifiable index connected to the LogReader. inputLog := logReaderSource{ r: inputReader, - } - inputLogCpParseFn := func(cpRaw []byte) (*log.Checkpoint, error) { - // No witnesses required yet - cp, _, _, err := log.ParseCheckpoint(cpRaw, ilv.Name(), ilv) - return cp, err - } - vi, err := vindex.NewVerifiableIndex(ctx, inputLog, inputLogCpParseFn, mapFnFromFlags(), mapRoot) - if err != nil { - return fmt.Errorf("failed to create vindex: %v", err) + v: ilv, } // Submits new entries to the log in the background. go submitEntries(ctx, inputAppender) - // Keeps the map synced with the latest published log state. - go maintainMap(ctx, vi) - - // Run a web server to handle queries over the verifiable index. - go runWebServer(vi, inputLogDir) - <-ctx.Done() - return nil + return inputLog, func() { + _ = inputShutdown(ctx) + } } // logReaderSource adapts a tessera.LogReader to a vindex.InputLog. type logReaderSource struct { r tessera.LogReader + v note.Verifier } func (s logReaderSource) Checkpoint(ctx context.Context) (checkpoint []byte, err error) { return s.r.ReadCheckpoint(ctx) } +func (s logReaderSource) Parse(cpRaw []byte) (*log.Checkpoint, error) { + cp, _, _, err := log.ParseCheckpoint(cpRaw, s.v.Name(), s.v) + return cp, err +} + func (s logReaderSource) Leaves(ctx context.Context, start, end uint64) iter.Seq2[[]byte, error] { tsf := func(ctx context.Context) (uint64, error) { return end, nil @@ -171,6 +188,17 @@ func (s logReaderSource) Leaves(ctx context.Context, start, end uint64) iter.Seq } } +// outputLogOrDie returns an output log using a POSIX log in the given directory. +func outputLogOrDie(ctx context.Context, outputLogDir string) (log vindex.OutputLog, closer func()) { + s, v := getOutputLogSignerVerifierOrDie() + + l, c, err := vindex.NewOutputLog(ctx, outputLogDir, s, v) + if err != nil { + klog.Exit(err) + } + return l, c +} + // maintainMap reads entries from the log and sync them to the vindex. func maintainMap(ctx context.Context, vi *vindex.VerifiableIndex) { ticker := time.NewTicker(10 * time.Second) @@ -226,7 +254,7 @@ func submitEntries(ctx context.Context, appender *tessera.Appender) { } } -func runWebServer(vi *vindex.VerifiableIndex, ild string) { +func runWebServer(vi *vindex.VerifiableIndex, inLogDir, outLogDir string) { web := NewServer(func(h [sha256.Size]byte) ([]uint64, error) { idxes, size := vi.Lookup(h) if size == 0 { @@ -235,9 +263,11 @@ func runWebServer(vi *vindex.VerifiableIndex, ild string) { return idxes, nil }) - ilfs := http.FileServer(http.Dir(ild)) + ilfs := http.FileServer(http.Dir(inLogDir)) + olfs := http.FileServer(http.Dir(outLogDir)) r := mux.NewRouter() r.PathPrefix("/inputlog/").Handler(http.StripPrefix("/inputlog/", ilfs)) + r.PathPrefix("/outputlog/").Handler(http.StripPrefix("/outputlog/", olfs)) web.registerHandlers(r) hServer := &http.Server{ Addr: *listen, @@ -272,6 +302,29 @@ func getInputLogSignerVerifierOrDie() (note.Signer, note.Verifier) { return s, v } +// Read output log private key from file or environment variable and generate the +// note Signer and Verifier pair for it. +func getOutputLogSignerVerifierOrDie() (note.Signer, note.Verifier) { + var privKey string + var err error + if len(*outputLogPrivKeyFile) > 0 { + privKey, err = getKeyFile(*outputLogPrivKeyFile) + if err != nil { + klog.Exitf("Unable to get private key: %v", err) + } + } else { + privKey = os.Getenv("OUTPUT_LOG_PRIVATE_KEY") + if len(privKey) == 0 { + klog.Exit("Supply private key file path using --output_log_private_key or set OUTPUT_LOG_PRIVATE_KEY environment variable") + } + } + s, v, err := fnote.NewEd25519SignerVerifier(privKey) + if err != nil { + klog.Exitf("Failed to get signer/verifier: %v", err) + } + return s, v +} + func getKeyFile(path string) (string, error) { k, err := os.ReadFile(path) if err != nil { diff --git a/vindex/map.go b/vindex/map.go index 4f7a02b..73a1e37 100644 --- a/vindex/map.go +++ b/vindex/map.go @@ -24,6 +24,7 @@ import ( "crypto/sha256" "encoding/base64" "encoding/binary" + "encoding/hex" "fmt" "io" "iter" @@ -62,11 +63,23 @@ type MapFn func([]byte) [][sha256.Size]byte type InputLog interface { // Checkpoint returns the latest checkpoint committing to the input log state. Checkpoint(ctx context.Context) (checkpoint []byte, err error) + // Parse unmarshals and verifies a checkpoint obtained from GetCheckpoint. + Parse(checkpoint []byte) (*log.Checkpoint, error) // Leaves returns all the leaves in the range [start, end), outputting them via // the returned iterator. Leaves(ctx context.Context, start, end uint64) iter.Seq2[[]byte, error] } +// OutputLog is where map roots are written as leaves. +type OutputLog interface { + // GetCheckpoint returns the latest checkpoint committing to the output log state. + Checkpoint(ctx context.Context) (checkpoint []byte, err error) + // Parse unmarshals and verifies a checkpoint obtained from GetCheckpoint. + Parse(checkpoint []byte) (*log.Checkpoint, error) + // Append adds a new leaf and returns the checkpoint that commits to it. + Append(ctx context.Context, data []byte) (idx uint64, checkpoint []byte, err error) +} + // OpenCheckpointFn is a function that parses a checkpoint, validating it, and returns a parsed // checkpoint. This is expected to be a thin wrapper around log.ParseCheckpoint with the // validators set up according to the index operator's policy on the number of witnesses @@ -78,7 +91,7 @@ type OpenCheckpointFn func(cpRaw []byte) (*log.Checkpoint, error) // path. // Note that only one IndexBuilder should exist for any given walPath at any time. The behaviour is unspecified, // but likely broken, if multiple processes are writing to the same file at any given time. -func NewVerifiableIndex(ctx context.Context, inputLog InputLog, inputLogParseFn OpenCheckpointFn, mapFn MapFn, rootDir string) (*VerifiableIndex, error) { +func NewVerifiableIndex(ctx context.Context, inputLog InputLog, mapFn MapFn, outputLog OutputLog, rootDir string) (*VerifiableIndex, error) { stateDir := path.Join(rootDir, "state") if err := os.MkdirAll(stateDir, 0o755); err != nil { return nil, err @@ -143,18 +156,19 @@ func NewVerifiableIndex(ctx context.Context, inputLog InputLog, inputLogParseFn return nil, fmt.Errorf("InitStorage: %s", err) } mapper := &inputLogMapper{ - inputLog: inputLog, - inputLogParseFn: inputLogParseFn, - mapFn: mapFn, - walWriter: wal, - db: db, - r: cr, + inputLog: inputLog, + mapFn: mapFn, + walWriter: wal, + db: db, + r: cr, } b := &VerifiableIndex{ mapper: mapper, walReader: reader, db: db, + outputLog: outputLog, vindex: *mpt.NewTree(sha256.Sum256, vtreeStorage), + vstore: vtreeStorage, data: map[[sha256.Size]byte][]uint64{}, } if err := b.buildMap(ctx); err != nil { @@ -166,11 +180,10 @@ func NewVerifiableIndex(ctx context.Context, inputLog InputLog, inputLogParseFn // inputLogMapper reads the Input Log, checking that the data matches the commitments, // and updates the WAL and DB with the resulting information. type inputLogMapper struct { - inputLog InputLog - inputLogParseFn OpenCheckpointFn - mapFn MapFn - walWriter *walWriter - db *pebble.DB + inputLog InputLog + mapFn MapFn + walWriter *walWriter + db *pebble.DB r *compact.Range } @@ -191,11 +204,15 @@ func (m *inputLogMapper) syncFromInputLog(ctx context.Context) error { if err != nil { return fmt.Errorf("failed to get latest checkpoint from DB: %s", err) } - cp, err := m.inputLogParseFn(rawCp) + cp, err := m.inputLog.Parse(rawCp) if err != nil { return fmt.Errorf("failed to parse checkpoint: %s", err) } + if cp.Size == 0 { + return nil + } + if m.r.End() < cp.Size { ctx, done := context.WithCancel(ctx) defer done() @@ -296,9 +313,11 @@ type VerifiableIndex struct { mapper *inputLogMapper walReader *walReader db *pebble.DB + outputLog OutputLog indexMu sync.RWMutex // covers vindex and data vindex mpt.Tree + vstore mpt.Storage data map[[sha256.Size]byte][]uint64 // servingSize is the size of the input log we are serving for. @@ -344,7 +363,54 @@ func (b *VerifiableIndex) Update(ctx context.Context) error { if err := b.mapper.syncFromInputLog(ctx); err != nil { return err } - return b.buildMap(ctx) + if err := b.buildMap(ctx); err != nil { + return err + } + return b.publish(ctx) +} + +func (b *VerifiableIndex) publish(ctx context.Context) error { + // Get the latest input log checkpoint the map was built from. + // TODO(mhutchinson): Possibly just pass this in? + inCp, inCloser, err := b.db.Get([]byte(dbLatestCheckpointKey)) + if err != nil { + if err == pebble.ErrNotFound { + // If the key isn't there then nothing to do. + return nil + } + return fmt.Errorf("failed to read latest checkpoint: %v", err) + } + if err := inCloser.Close(); err != nil { + return fmt.Errorf("failed to close: %v", err) + } + + // Construct the leaf for the output log + rootNode, err := b.vstore.Load(mpt.RootLabel) + if err != nil { + return fmt.Errorf("failed to load vindex root: %v", err) + } + leaf := append(hex.AppendEncode(nil, rootNode.Hash[:]), '\n', '\n') + leaf = append(leaf, inCp...) + + outIdx, rawCp, err := b.outputLog.Append(ctx, leaf) + if err != nil { + return fmt.Errorf("failed to append to output log: %v", err) + } + if klog.V(1).Enabled() { + _, inSize, _, err := checkpointUnsafe(inCp) + if err != nil { + klog.Error(err) + return nil + } + _, outSize, _, err := checkpointUnsafe(rawCp) + if err != nil { + klog.Error(err) + return nil + } + klog.V(1).Infof("Published checkpoint for input log size %d into output log at index %d, and got checkpoint for output log size %d", inSize, outIdx, outSize) + } + + return nil } // buildMap reads from the WAL until the file has been consumed and the map has been diff --git a/vindex/map_test.go b/vindex/map_test.go index 2ce0a10..e453ffb 100644 --- a/vindex/map_test.go +++ b/vindex/map_test.go @@ -23,6 +23,7 @@ import ( "encoding/hex" "iter" "os" + "path" "testing" "github.com/google/go-cmp/cmp" @@ -48,15 +49,12 @@ func TestVerifiableIndex(t *testing.T) { t: testonly.New(rfc6962.DefaultHasher), leaves: make([][]byte, 0), s: s, + v: v, } for _, str := range []string{"foo: 2", "bar: 5", "bar: 10", "foo: 8"} { inputLog.Append(str) } - inputLogCpParseFn := func(cpRaw []byte) (*log.Checkpoint, error) { - cp, _, _, err := log.ParseCheckpoint(cpRaw, v.Name(), v) - return cp, err - } mapFn := func(leaf []byte) [][sha256.Size]byte { key, _, found := bytes.Cut(leaf, []byte(":")) if !found { @@ -77,7 +75,14 @@ func TestVerifiableIndex(t *testing.T) { if err := os.MkdirAll(f.Name(), 0o755); err != nil { t.Fatal(err) } - vi, err := NewVerifiableIndex(ctx, inputLog, inputLogCpParseFn, mapFn, f.Name()) + + old := path.Join(f.Name(), "outputlog") + outputLog, closer, err := NewOutputLog(ctx, old, s, v) + if err != nil { + t.Fatal(err) + } + defer closer() + vi, err := NewVerifiableIndex(ctx, inputLog, mapFn, outputLog, f.Name()) if err != nil { t.Fatal(err) } @@ -115,6 +120,7 @@ type inMemoryTreeSource struct { t *testonly.Tree leaves [][]byte s note.Signer + v note.Verifier } func (s *inMemoryTreeSource) Checkpoint(ctx context.Context) (checkpoint []byte, err error) { @@ -130,6 +136,11 @@ func (s *inMemoryTreeSource) Checkpoint(ctx context.Context) (checkpoint []byte, return note.Sign(n, s.s) } +func (s *inMemoryTreeSource) Parse(cpRaw []byte) (*log.Checkpoint, error) { + cp, _, _, err := log.ParseCheckpoint(cpRaw, s.v.Name(), s.v) + return cp, err +} + func (s *inMemoryTreeSource) Leaves(ctx context.Context, start, end uint64) iter.Seq2[[]byte, error] { return func(yield func([]byte, error) bool) { for _, entry := range s.leaves { diff --git a/vindex/outputlog.go b/vindex/outputlog.go new file mode 100644 index 0000000..594bbcd --- /dev/null +++ b/vindex/outputlog.go @@ -0,0 +1,75 @@ +// Copyright 2025 Google LLC. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package vindex + +import ( + "context" + "fmt" + "time" + + "github.com/transparency-dev/formats/log" + "github.com/transparency-dev/tessera" + "github.com/transparency-dev/tessera/storage/posix" + "golang.org/x/mod/sumdb/note" +) + +// outputLogOrDie returns an output log using a POSIX log in the given directory. +func NewOutputLog(ctx context.Context, outputLogDir string, s note.Signer, v note.Verifier) (log OutputLog, closer func(), err error) { + driver, err := posix.New(ctx, posix.Config{Path: outputLogDir}) + if err != nil { + return nil, nil, fmt.Errorf("failed to create input log: %v", err) + } + + appender, shutdown, reader, err := tessera.NewAppender(ctx, driver, tessera.NewAppendOptions(). + WithCheckpointSigner(s). + WithCheckpointInterval(5*time.Second). + WithBatching(1, time.Second)) + if err != nil { + return nil, nil, fmt.Errorf("failed to get appender: %v", err) + } + awaiter := tessera.NewPublicationAwaiter(ctx, reader.ReadCheckpoint, 100*time.Millisecond) + + outputLog := posixOutputLog{ + a: appender, + w: awaiter, + r: reader, + v: v, + } + + return outputLog, func() { + _ = shutdown(ctx) + }, nil +} + +type posixOutputLog struct { + a *tessera.Appender + w *tessera.PublicationAwaiter + r tessera.LogReader + v note.Verifier +} + +func (l posixOutputLog) Checkpoint(ctx context.Context) (checkpoint []byte, err error) { + return l.r.ReadCheckpoint(ctx) +} + +func (l posixOutputLog) Parse(cpRaw []byte) (*log.Checkpoint, error) { + cp, _, _, err := log.ParseCheckpoint(cpRaw, l.v.Name(), l.v) + return cp, err +} + +func (l posixOutputLog) Append(ctx context.Context, data []byte) (idx uint64, checkpoint []byte, err error) { + index, cp, err := l.w.Await(ctx, l.a.Add(ctx, tessera.NewEntry(data))) + return index.Index, cp, err +}