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 +}