added basic server-side cosigning (work in progress)

1 files changed, 115 insertions, 42 deletions

diff --git a/server/main.go b/server/main.go
index 8f5ab48..7402fb3 100644
--- a/server/main.go
+++ b/server/main.go
@@ -2,13 +2,19 @@
 package main
 
 import (
+	"context"
 	"flag"
+	"fmt"
+	"os"
 	"strings"
+	"sync"
+	"syscall"
 	"time"
 
 	"crypto/ed25519"
 	"encoding/base64"
 	"net/http"
+	"os/signal"
 
 	"github.com/golang/glog"
 	"github.com/google/trillian"
@@ -23,78 +29,145 @@ var (
 	rpcBackend   = flag.String("log_rpc_server", "localhost:6962", "host:port specification of where Trillian serves clients")
 	prefix       = flag.String("prefix", "st/v1", "a prefix that proceeds each endpoint path")
 	trillianID   = flag.Int64("trillian_id", 5991359069696313945, "log identifier in the Trillian database")
-	rpcDeadline  = flag.Duration("rpc_deadline", time.Second*10, "deadline for backend RPC requests")
+	deadline     = flag.Duration("deadline", time.Second*10, "deadline for backend requests")
 	key          = flag.String("key", "8gzezwrU/2eTrO6tEYyLKsoqn5V54URvKIL9cTE7jUYUqXVX4neJvcBq/zpSAYPsZFG1woh0OGBzQbi9UP9MZw==", "base64-encoded Ed25519 signing key")
-	namespaces   = flag.String("namespaces", "AAEgHOQFUkKNWpjYAhNKTyWCzahlI7RDtf5123kHD2LACj0=,AAEgLqrWb9JwQUTk/SwTNDdMH8aRmy3mbmhwEepO5WSgb+A=", "comma-separated list of trusted namespaces in base64 (default: testdata.Ed25519{Vk,Vk2})")
+	submitters   = flag.String("submitters", "AAEgHOQFUkKNWpjYAhNKTyWCzahlI7RDtf5123kHD2LACj0=,AAEgLqrWb9JwQUTk/SwTNDdMH8aRmy3mbmhwEepO5WSgb+A=", "comma-separated list of trusted submitter namespaces in base64 (default: testdata.Ed25519{Vk,Vk2})")
+	witnesses    = flag.String("witnesses", "", "comma-separated list of trusted submitter namespaces in base64 (default: none")
 	maxRange     = flag.Int64("max_range", 2, "maximum number of entries that can be retrived in a single request")
+	interval     = flag.Duration("interval", time.Second*30, "interval used to rotate the log's cosigned STH")
 )
 
 func main() {
 	flag.Parse()
+	defer glog.Flush()
 
-	glog.Info("Dialling Trillian gRPC log server")
-	dialOpts := []grpc.DialOption{grpc.WithInsecure(), grpc.WithBlock(), grpc.WithTimeout(*rpcDeadline)}
+	// wait for clean-up before exit
+	var wg sync.WaitGroup
+	defer wg.Wait()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	glog.V(3).Infof("configuring stfe instance...")
+	instance, err := setupInstanceFromFlags()
+	if err != nil {
+		glog.Errorf("setupInstance: %v", err)
+		return
+	}
+
+	glog.V(3).Infof("spawning SthSource")
+	go func() {
+		wg.Add(1)
+		defer wg.Done()
+		instance.SthSource.Run(ctx)
+		glog.Errorf("SthSource shutdown")
+		cancel() // must have SthSource running
+	}()
+
+	glog.V(3).Infof("spawning await")
+	server := http.Server{Addr: *httpEndpoint}
+	go await(ctx, func() {
+		wg.Add(1)
+		defer wg.Done()
+		ctxInner, _ := context.WithTimeout(ctx, time.Second*60)
+		glog.Infof("Shutting down HTTP server...")
+		server.Shutdown(ctxInner)
+		glog.V(3).Infof("HTTP server shutdown")
+		glog.Infof("Shutting down spawned go routines...")
+		cancel()
+	})
+
+	glog.Infof("Serving on %v/%v", *httpEndpoint, *prefix)
+	if err = server.ListenAndServe(); err != http.ErrServerClosed {
+		glog.Errorf("ListenAndServe: %v", err)
+	}
+}
+
+// SetupInstance sets up a new STFE instance from flags
+func setupInstanceFromFlags() (*stfe.Instance, error) {
+	// Trillian gRPC connection
+	dialOpts := []grpc.DialOption{grpc.WithInsecure(), grpc.WithBlock(), grpc.WithTimeout(*deadline)}
 	conn, err := grpc.Dial(*rpcBackend, dialOpts...)
 	if err != nil {
-		glog.Fatal(err)
+		return nil, fmt.Errorf("Dial: %v", err)
 	}
 	client := trillian.NewTrillianLogClient(conn)
-
-	glog.Info("Creating HTTP request multiplexer")
+	// HTTP multiplexer
 	mux := http.NewServeMux()
 	http.Handle("/", mux)
-
-	glog.Info("Adding prometheus handler on path: /metrics")
+	// Prometheus metrics
+	glog.V(3).Infof("Adding prometheus handler on path: /metrics")
 	http.Handle("/metrics", promhttp.Handler())
-
-	glog.Infof("Creating namespace pool")
-	var anchors []*namespace.Namespace
-	for _, b64 := range strings.Split(*namespaces, ",") {
-		b, err := base64.StdEncoding.DecodeString(b64)
-		if err != nil {
-			glog.Fatalf("invalid namespace: %s: %v", b64, err)
-		}
-		var namespace namespace.Namespace
-		if err := namespace.Unmarshal(b); err != nil {
-			glog.Fatalf("invalid namespace: %s: %v", b64, err)
-		}
-		anchors = append(anchors, &namespace)
+	// Trusted submitters
+	submitters, err := newNamespacePoolFromString(*submitters)
+	if err != nil {
+		return nil, fmt.Errorf("submitters: newNamespacePoolFromString: %v", err)
 	}
-	pool, err := namespace.NewNamespacePool(anchors)
+	// Trusted witnesses
+	witnesses, err := newNamespacePoolFromString(*witnesses)
 	if err != nil {
-		glog.Fatalf("invalid namespace pool: %v", err)
+		return nil, fmt.Errorf("witnesses: NewNamespacePool: %v", err)
 	}
-
-	glog.Infof("Creating log signer and identifier")
+	// Log identity
 	sk, err := base64.StdEncoding.DecodeString(*key)
 	if err != nil {
-		glog.Fatalf("invalid signing key: %v", err)
+		return nil, fmt.Errorf("sk: DecodeString: %v", err)
 	}
 	signer := ed25519.PrivateKey(sk)
 	logId, err := namespace.NewNamespaceEd25519V1([]byte(ed25519.PrivateKey(sk).Public().(ed25519.PublicKey)))
 	if err != nil {
-		glog.Fatalf("failed creating log id from secret key: %v", err)
+		return nil, fmt.Errorf("NewNamespaceEd25519V1: %v", err)
 	}
-
-	glog.Infof("Initializing log parameters")
-	lp, err := stfe.NewLogParameters(signer, logId, *trillianID, *prefix, pool, *maxRange)
+	// Setup log parameters
+	lp, err := stfe.NewLogParameters(signer, logId, *trillianID, *prefix, submitters, witnesses, *maxRange, *interval, *deadline)
 	if err != nil {
-		glog.Fatalf("failed setting up log parameters: %v", err)
+		return nil, fmt.Errorf("NewLogParameters: %v", err)
 	}
-
-	i := stfe.NewInstance(lp, client, *rpcDeadline)
+	// Setup STH source
+	source, err := stfe.NewActiveSthSource(client, lp)
+	if err != nil {
+		return nil, fmt.Errorf("NewActiveSthSource: %v", err)
+	}
+	// Setup log instance
+	i := stfe.NewInstance(lp, client, source)
 	for _, handler := range i.Handlers() {
-		glog.Infof("adding handler: %s", handler.Path())
+		glog.V(3).Infof("adding handler: %s", handler.Path())
 		mux.Handle(handler.Path(), handler)
 	}
-	glog.Infof("Configured: %s", i)
+	return i, nil
+}
 
-	glog.Infof("Serving on %v/%v", *httpEndpoint, *prefix)
-	srv := http.Server{Addr: *httpEndpoint}
-	err = srv.ListenAndServe()
-	if err != http.ErrServerClosed {
-		glog.Warningf("Server exited: %v", err)
+// newNamespacePoolFromString creates a new namespace pool from a
+// comma-separated list of serialized and base64-encoded namespaces.
+func newNamespacePoolFromString(str string) (*namespace.NamespacePool, error) {
+	var namespaces []*namespace.Namespace
+	if len(str) > 0 {
+		for _, b64 := range strings.Split(str, ",") {
+			b, err := base64.StdEncoding.DecodeString(b64)
+			if err != nil {
+				return nil, fmt.Errorf("DecodeString: %v", err)
+			}
+			var namespace namespace.Namespace
+			if err := namespace.Unmarshal(b); err != nil {
+				return nil, fmt.Errorf("Unmarshal: %v", err)
+			}
+			namespaces = append(namespaces, &namespace)
+		}
 	}
+	pool, err := namespace.NewNamespacePool(namespaces)
+	if err != nil {
+		return nil, fmt.Errorf("NewNamespacePool: %v", err)
+	}
+	return pool, nil
+}
 
-	glog.Flush()
+// await waits for a shutdown signal and then runs a clean-up function
+func await(ctx context.Context, done func()) {
+	sigs := make(chan os.Signal, 1)
+	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
+	select {
+	case <-sigs:
+	case <-ctx.Done():
+	}
+	glog.V(3).Info("received shutdown signal")
+	done()
 }