restructured non-verifiying monitor

Author: liamsi

cmd/keytransparency-monitor/main.go

@@ -35,11 +35,15 @@ import (
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/reflection"
 
+	"github.com/google/keytransparency/core/monitor/storage"
 	cmon "github.com/google/keytransparency/core/monitor"
 	kpb "github.com/google/keytransparency/core/proto/keytransparency_v1_types"
+	"github.com/google/keytransparency/impl/monitor/client"
 	spb "github.com/google/keytransparency/impl/proto/keytransparency_v1_service"
 	mopb "github.com/google/keytransparency/impl/proto/monitor_v1_service"
 	mupb "github.com/google/keytransparency/impl/proto/mutation_v1_service"
+	_ "github.com/google/trillian/merkle/coniks"    // Register coniks
+	_ "github.com/google/trillian/merkle/objhasher" // Register objhasher
 )
 
 var (
@@ -121,12 +125,8 @@ func main() {
 		glog.Fatalf("Could not read domain info %v:", err)
 	}
 
-	srv := monitor.New(mcc, crypto.NewSHA256Signer(key), logTree, mapTree, *pollPeriod)
-	mon, err := cmon.New(logTree, mapTree, crypto.NewSHA256Signer(key))
-	if err != nil {
-		glog.Exitf("Failed setting up REST proxy: %v", err)
-	}
-
+	store := storage.New()
+	srv := monitor.New(store)
 	mopb.RegisterMonitorServiceServer(grpcServer, srv)
 	reflection.Register(grpcServer)
 	grpc_prometheus.Register(grpcServer)
@@ -142,6 +142,31 @@ func main() {
 	mux := http.NewServeMux()
 	mux.Handle("/", gwmux)
 
+	// initialize the mutations API client and feed the responses it got
+	// into the monitor:
+	mon, err := cmon.New(logTree, mapTree, crypto.NewSHA256Signer(key), store)
+	if err != nil {
+		glog.Exitf("Failed to initialize monitor: %v", err)
+	}
+	mutCli := client.New(mcc, *pollPeriod)
+	responses, errs := mutCli.StartPolling(1)
+	go func() {
+		for {
+			select {
+			case mutResp := <-responses:
+				glog.Infof("Received mutations response: %v", mutResp.Epoch)
+				if err := mon.Process(mutResp); err != nil {
+					glog.Infof("Error processing mutations response: %v", err)
+				}
+			case err := <-errs:
+				// this is OK if there were no mutations in  between:
+				// TODO(ismail): handle the case when the known maxDuration has
+				// passed and no epoch was issued?
+				glog.Infof("Could not retrieve mutations API response %v", err)
+			}
+		}
+	}()
+
 	// Serve HTTP2 server over TLS.
 	glog.Infof("Listening on %v", *addr)
 	if err := http.ListenAndServeTLS(*addr, *certFile, *keyFile,
@@ -190,7 +215,7 @@ func transportCreds(ktURL string, ktCert string, insecure bool) (credentials.Tra
 }
 
 // config selects a source for and returns the client configuration.
-func getTrees(ctx context.Context, cc *grpc.ClientConn) (mapTree *trillian.Tree, logTree *trillian.Tree, err error) {
+func getTrees(ctx context.Context, cc *grpc.ClientConn) (logTree *trillian.Tree, mapTree *trillian.Tree, err error) {
 	ktClient := spb.NewKeyTransparencyServiceClient(cc)
 	resp, err2 := ktClient.GetDomainInfo(ctx, &kpb.GetDomainInfoRequest{})
 	if err2 != nil {

core/monitor/monitor.go

@@ -17,11 +17,16 @@ package monitor
 import (
 	"crypto"
 	"fmt"
+	"time"
+
+	"github.com/golang/glog"
+
+	"github.com/google/keytransparency/core/monitor/storage"
+	ktpb "github.com/google/keytransparency/core/proto/keytransparency_v1_types"
 
 	"github.com/google/trillian"
 	"github.com/google/trillian/merkle"
 	"github.com/google/trillian/merkle/hashers"
-
 	tcrypto "github.com/google/trillian/crypto"
 )
 
@@ -35,10 +40,11 @@ type Monitor struct {
 	signer      *tcrypto.Signer
 	// TODO(ismail): update last trusted signed log root
 	//trusted     trillian.SignedLogRoot
+	store *storage.Storage
 }
 
 // New creates a new instance of the monitor.
-func New(logTree, mapTree *trillian.Tree, signer *tcrypto.Signer) (*Monitor, error) {
+func New(logTree, mapTree *trillian.Tree, signer *tcrypto.Signer, store *storage.Storage) (*Monitor, error) {
 	logHasher, err := hashers.NewLogHasher(logTree.GetHashStrategy())
 	if err != nil {
 		return nil, fmt.Errorf("Failed creating LogHasher: %v", err)
@@ -53,5 +59,26 @@ func New(logTree, mapTree *trillian.Tree, signer *tcrypto.Signer) (*Monitor, err
 		logPubKey:   logTree.GetPublicKey(),
 		mapPubKey:   mapTree.GetPublicKey(),
 		signer:      signer,
+		store:       store,
 	}, nil
 }
+
+func (m *Monitor) Process(resp *ktpb.GetMutationsResponse) error {
+	var smr *trillian.SignedMapRoot
+	var err error
+	seen := time.Now().Unix()
+	errs := m.VerifyMutationsResponse(resp)
+	if len(errs) == 0 {
+		glog.Infof("Successfully verified mutations response for epoch: %v", resp.Epoch)
+		smr, err = m.signMapRoot(resp)
+		if err != nil {
+			glog.Errorf("Failed to sign map root for epoch %v: %v", resp.Epoch, err)
+			return fmt.Errorf("m.signMapRoot(_): %v", err)
+		}
+	}
+	if err := m.store.Set(resp.Epoch, seen, smr, resp, errs); err != nil {
+		glog.Errorf("m.store.Set(%v, %v, _, _, %v): %v", resp.Epoch, seen, errs, err)
+		return err
+	}
+	return nil
+}

core/monitor/response.go

@@ -12,21 +12,26 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-// Package monitor implements the monitor service. A monitor repeatedly polls a
-// key-transparency server's Mutations API and signs Map Roots if it could
-// reconstruct
-// clients can query.
 package monitor
 
 import (
-	"errors"
-)
+	"fmt"
+
+	ktpb "github.com/google/keytransparency/core/proto/keytransparency_v1_types"
 
-var (
-	// ErrNothingProcessed occurs when the monitor did not process any mutations /
-	// smrs yet.
-	ErrNothingProcessed = errors.New("did not process any mutations yet")
+	"github.com/google/trillian"
 )
 
-// TODO(ismail): call the client, actually process the mutations by calling API
-// in core
+func (m *Monitor) signMapRoot(in *ktpb.GetMutationsResponse) (*trillian.SignedMapRoot, error) {
+	// copy of received SMR:
+	smr := *in.Smr
+	smr.Signature = nil
+
+	sig, err := m.signer.SignObject(smr)
+	if err != nil {
+		return nil, fmt.Errorf("SignObject(): %v", err)
+	}
+	smr.Signature = sig
+
+	return &smr, nil
+}

impl/monitor/process.go renamed to core/monitor/sign.go

@@ -0,0 +1,79 @@
+// Copyright 2017 Google Inc. 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 storage
+
+import (
+	"errors"
+
+	ktpb "github.com/google/keytransparency/core/proto/keytransparency_v1_types"
+	"github.com/google/trillian"
+)
+
+var (
+	ErrAlreadyStored = errors.New("already stored epoch")
+	ErrNotFound      = errors.New("data for epoch not found")
+)
+
+type MonitoringResult struct {
+	// in case of success this contains the map root signed by the monitor
+	Smr *trillian.SignedMapRoot
+	// response contains the original mutations API response from the server
+	// in case at least one verification step failed
+	Response *ktpb.GetMutationsResponse
+	Seen     int64
+	Errors   []error
+}
+
+type Storage struct {
+	store map[int64]*MonitoringResult
+	latest int64
+}
+
+func New() *Storage {
+	return &Storage{
+		store: make(map[int64]*MonitoringResult),
+	}
+}
+
+func (s *Storage) Set(epoch int64,
+	seenNanos int64,
+	smr *trillian.SignedMapRoot,
+	response *ktpb.GetMutationsResponse,
+	errorList []error) error {
+	// see if we already processed this epoch:
+	if _, ok := s.store[epoch]; ok {
+		return ErrAlreadyStored
+	}
+	// if not we just store the value:
+	s.store[epoch] = &MonitoringResult{
+		Smr:      smr,
+		Seen:     seenNanos,
+		Response: response,
+		Errors:   errorList,
+	}
+	s.latest=epoch
+	return nil
+}
+
+func (s *Storage) Get(epoch int64) (*MonitoringResult, error) {
+	if result, ok := s.store[epoch]; ok {
+		return result, nil
+	}
+	return nil, ErrNotFound
+}
+
+func (s *Storage) LatestEpoch() int64 {
+	return s.latest
+}

core/monitor/storage/storage.go

@@ -26,7 +26,6 @@ import (
 // of received mutations may differ from those included in the initial response
 // because of the max. page size. If any verification check failed it returns
 // an error.
-// TODO(ismail): make this return a list of errors
-func (m *Monitor) VerifyMutationsResponse(in *ktpb.GetMutationsResponse, allMuts []*ktpb.Mutation) error {
+func (m *Monitor) VerifyMutationsResponse(in *ktpb.GetMutationsResponse) []error {
 	return nil
 }