serverlessclone.go

// Copyright 2021 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.

// serverlessclone is a one-shot tool for downloading entries from an
// HTTP(s) exposed transparency log generated by the serverless tooling.
package main

import (
	"context"
	"flag"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"os"
	"time"

	"github.com/golang/glog"
	"github.com/google/trillian-examples/clone/internal/cloner"
	"github.com/google/trillian-examples/clone/logdb"
	"github.com/transparency-dev/serverless-log/client"
	"golang.org/x/mod/sumdb/note"

	_ "github.com/go-sql-driver/mysql"
)

var (
	logURL   = flag.String("url", "", "The base URL for the log HTTP API, should end with a trailing slash")
	vkey     = flag.String("vkey", "", "The verification key for the log checkpoints")
	origin   = flag.String("origin", "", "The origin string for the log checkpoints")
	mysqlURI = flag.String("mysql_uri", "", "URL of a MySQL database to clone the log into. The DB should contain only one log.")

	writeBatchSize = flag.Uint("write_batch_size", 100, "The number of leaves to write in each DB transaction.")
	workers        = flag.Uint("workers", 50, "The number of worker threads to run in parallel to fetch entries.")
	timeout        = flag.Duration("timeout", 10*time.Second, "Maximum time to wait for http connections to complete.")
)

func main() {
	flag.Parse()

	if *logURL == "" {
		glog.Exit("Missing required parameter 'url'")
	}
	if *vkey == "" {
		glog.Exit("Missing required parameter 'vkey'")
	}
	if *origin == "" {
		glog.Exit("Missing required parameter 'origin'")
	}
	if *mysqlURI == "" {
		glog.Exit("Missing required parameter 'mysql_uri'")
	}

	ctx := context.Background()
	db, err := logdb.NewDatabase(*mysqlURI)
	if err != nil {
		glog.Exitf("Failed to connect to database: %q", err)
	}

	v, err := note.NewVerifier(*vkey)
	if err != nil {
		glog.Exitf("Failed to create verifier: %v", err)
	}
	u, err := url.Parse(*logURL)
	if err != nil {
		glog.Exitf("Invalid log URL %q: %v", *logURL, err)
	}
	f := newFetcher(u)

	targetCp, rawCp, _, err := client.FetchCheckpoint(ctx, f, v, *origin)
	if err != nil {
		glog.Exitf("Failed to get latest checkpoint from log: %v", err)
	}
	glog.Infof("Target checkpoint is for tree size %d", targetCp.Size)

	cp := cloner.UnwrappedCheckpoint{
		Size: targetCp.Size,
		Hash: targetCp.Hash,
		Raw:  rawCp,
	}
	if err := clone(ctx, db, f, cp); err != nil {
		glog.Exitf("Failed to clone: %v", err)
	}
}

func clone(ctx context.Context, db *logdb.Database, f client.Fetcher, targetCp cloner.UnwrappedCheckpoint) error {
	cl := cloner.New(*workers, 1, *writeBatchSize, db)

	next, err := cl.Next()
	if err != nil {
		return fmt.Errorf("couldn't determine first leaf to fetch: %v", err)
	}
	// TODO(mhutchinson): other implementations don't have this check. Is this redundant,
	// OR can it be moved deeper into the call stack?
	if next >= uint64(targetCp.Size) {
		glog.Infof("No work to do. Local tree size = %d, latest log tree size = %d", next, targetCp.Size)
		return nil
	}

	batchFetch := func(start uint64, leaves [][]byte) (uint64, error) {
		if len(leaves) != 1 {
			return 0, fmt.Errorf("true batch fetching not supported")
		}
		leaf, err := client.GetLeaf(ctx, f, start)
		leaves[0] = leaf
		return 1, err
	}

	if err := cl.CloneAndVerify(ctx, batchFetch, targetCp); err != nil {
		return fmt.Errorf("failed to clone and verify log: %v", err)
	}
	return nil
}

// newFetcher creates a Fetcher for the log at the given root location.
func newFetcher(root *url.URL) client.Fetcher {
	get := getByScheme[root.Scheme]
	if get == nil {
		panic(fmt.Errorf("unsupported URL scheme %s", root.Scheme))
	}

	return func(ctx context.Context, p string) ([]byte, error) {
		u, err := root.Parse(p)
		if err != nil {
			return nil, err
		}
		return get(ctx, u)
	}
}

var getByScheme = map[string]func(context.Context, *url.URL) ([]byte, error){
	"http":  readHTTP,
	"https": readHTTP,
}

func readHTTP(ctx context.Context, u *url.URL) ([]byte, error) {
	req, err := http.NewRequest("GET", u.String(), nil)
	if err != nil {
		return nil, err
	}
	c := http.Client{
		Timeout: *timeout,
	}
	resp, err := c.Do(req.WithContext(ctx))
	if err != nil {
		return nil, err
	}
	switch resp.StatusCode {
	case 404:
		glog.Infof("Not found: %q", u.String())
		return nil, os.ErrNotExist
	case 200:
		break
	default:
		return nil, fmt.Errorf("unexpected http status %q", resp.Status)
	}
	defer func() {
		if err := resp.Body.Close(); err != nil {
			glog.Errorf("resp.Body.Close(): %v", err)
		}
	}()
	return io.ReadAll(resp.Body)
}