more wip collect

2 files changed, 49 insertions, 19 deletions

diff --git a/collect.go b/collect.go
index c1e2c6d..739400f 100644
--- a/collect.go
+++ b/collect.go
@@ -15,12 +15,12 @@ import (
 	"time"
 
 	"git.cs.kau.se/rasmoste/ct-sans/internal/chunk"
+	"git.cs.kau.se/rasmoste/ct-sans/internal/merkle"
 	"git.cs.kau.se/rasmoste/ct-sans/internal/utils"
 	ct "github.com/google/certificate-transparency-go"
 	"github.com/google/certificate-transparency-go/client"
 	"github.com/google/certificate-transparency-go/jsonclient"
 	"github.com/google/certificate-transparency-go/scanner"
-	"gitlab.torproject.org/rgdd/ct/pkg/merkle"
 	"gitlab.torproject.org/rgdd/ct/pkg/metadata"
 )
 
@@ -77,7 +77,6 @@ func collect(opts options) error {
 				cancel()
 				return
 			}
-
 			cli, err := client.New(string(log.URL), &http.Client{}, jsonclient.Options{UserAgent: "wip2"})
 			if err != nil {
 				fmt.Fprintf(os.Stderr, "ERROR: %s: %v\n", *log.Description, err)
@@ -89,7 +88,6 @@ func collect(opts options) error {
 				StartIndex:    th.TreeSize,
 				EndIndex:      int64(sth.TreeSize),
 				ParallelFetch: int(opts.workersPerLog),
-				Continuous:    false,
 			})
 
 			//
@@ -101,7 +99,10 @@ func collect(opts options) error {
 				for i := 0; i < len(eb.Entries); i++ {
 					leafHashes = append(leafHashes, merkle.HashLeafNode(eb.Entries[i].LeafInput))
 				}
-				sans := []string{"example.com"} // TODO: fixme
+				sans, errs := utils.SANsFromLeafEntries(eb.Start, eb.Entries)
+				for _, err := range errs {
+					fmt.Fprintf(os.Stderr, "WARNING: %s: %v", *log.Description, err)
+				}
 				chunks <- &chunk.Chunk{eb.Start, leafHashes, sans}
 			}
 
@@ -210,30 +211,39 @@ func readSnapshot(opts options, logID []byte) (ct.SignedTreeHead, error) {
 }
 
 func persistChunk(cli *client.LogClient, opts options, logID []byte, minSequence int64, c *chunk.Chunk) (bool, error) {
-	if len(c.LeafHashes) == 0 {
+	chunkSize := int64(len(c.LeafHashes))
+	if chunkSize == 0 {
 		return false, nil // nothing to persist
 	}
-	if int64(len(c.LeafHashes)) < minSequence {
+	if chunkSize < minSequence {
 		return true, nil // wait for more leaves
 	}
 
 	// Read persisted tree state from disk
-	th, err := readState(opts, logID)
+	oldTH, err := readState(opts, logID)
 	if err != nil {
 		return false, err
 	}
-	if th.TreeSize != c.Start {
-		return false, fmt.Errorf("disk state says next index is %d, in-memory says %d", th.TreeSize, c.Start)
+	if oldTH.TreeSize != c.Start {
+		return false, fmt.Errorf("disk state says next index is %d, in-memory says %d", oldTH.TreeSize, c.Start)
 	}
-
-	// Derive next intermediate tree state to persist
+	// Read signed tree head from disk
+	sth, err := readSnapshot(opts, logID)
+	if err != nil {
+		return false, err
+	}
+	// Derive next intermediate tree state from a compact range
+	//
+	// Santity checks: expected indces/sizes and consistent root hashes.
+	// This is redundant, but could, e.g., catch bugs with our storage.
 	//
 	// Independent context because we need to run inclusion and consistency
 	// queries after the parent context is cancelled to persist on shutdown
 	//
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	p, err := cli.GetProofByHash(ctx, c.LeafHashes[0][:], uint64(c.Start+int64(len(c.LeafHashes))))
+	newTH := treeHead{TreeSize: c.Start + chunkSize}
+	p, err := cli.GetProofByHash(ctx, c.LeafHashes[0][:], uint64(newTH.TreeSize))
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "WARNING: %x: %v\n", logID, err)
 		return true, nil // try again later
@@ -241,7 +251,24 @@ func persistChunk(cli *client.LogClient, opts options, logID []byte, minSequence
 	if p.LeafIndex != c.Start {
 		return false, fmt.Errorf("log says proof for entry %d is at index %d", c.Start, p.LeafIndex)
 	}
-	// TODO: ranged inclusion verify + consistency proof
+	if newTH.RootHash, err = merkle.TreeHeadFromRangeProof(c.LeafHashes, uint64(c.Start), utils.Proof(p.AuditPath)); err != nil {
+		return false, err
+	}
+	hashes, err := cli.GetSTHConsistency(ctx, uint64(oldTH.TreeSize), uint64(newTH.TreeSize))
+	if err != nil {
+		return true, nil // try again later
+	}
+	if err := merkle.VerifyConsistency(uint64(oldTH.TreeSize), uint64(newTH.TreeSize), oldTH.RootHash, newTH.RootHash, utils.Proof(hashes)); err != nil {
+		return false, fmt.Errorf("%d %x is inconsistent with on-disk state: %v", newTH.TreeSize, newTH.RootHash, err)
+	}
+
+	// Check that new tree state is consistent with the signed tree head
+	if hashes, err = cli.GetSTHConsistency(ctx, uint64(newTH.TreeSize), sth.TreeSize); err != nil {
+		return true, nil // try again later
+	}
+	if err := merkle.VerifyConsistency(uint64(newTH.TreeSize), sth.TreeSize, newTH.RootHash, sth.SHA256RootHash, utils.Proof(hashes)); err != nil {
+		return false, fmt.Errorf("%d %x is inconsistent with signed tree head: %v", newTH.TreeSize, newTH.RootHash, err)
+	}
 
 	// Persist SANs to disk
 	fp, err := os.OpenFile(fmt.Sprintf("%s/%x/%s", opts.logDirectory, logID, opts.sansFile), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644)
@@ -256,8 +283,8 @@ func persistChunk(cli *client.LogClient, opts options, logID []byte, minSequence
 		return false, err
 	}
 
-	// Persist intermediate log state to disk
-	b, err := json.Marshal(&treeHead{c.Start + int64(len(c.LeafHashes)), [sha256.Size]byte{}})
+	// Persist new tree state to disk
+	b, err := json.Marshal(&newTH)
 	if err != nil {
 		return false, err
 	}
@@ -265,6 +292,6 @@ func persistChunk(cli *client.LogClient, opts options, logID []byte, minSequence
 		return false, err
 	}
 
-	fmt.Fprintf(os.Stderr, "DEBUG: %x: persist: start=%d end=%d\n", logID, c.Start, c.Start+int64(len(c.LeafHashes)))
+	fmt.Fprintf(os.Stderr, "DEBUG: %x: persist: start=%d next=%d\n", logID, oldTH.TreeSize, newTH.TreeSize)
 	return false, nil
 }
diff --git a/main.go b/main.go
index 02ed413..dea0df7 100644
--- a/main.go
+++ b/main.go
@@ -86,9 +86,12 @@ func main() {
 	opts.stateFile = "th.json"
 	opts.sansFile = "sans.lst"
 
-	opts.workersPerLog = 10
-	opts.batchSize = 16384
-	opts.persistSize = 256
+	//opts.workersPerLog = 100
+	//opts.batchSize = 100
+	//opts.persistSize = 10000
+	opts.workersPerLog = 1
+	opts.batchSize = 10
+	opts.persistSize = 1
 
 	// Hand-over to the respective subcommands
 	var err error