<commit_before>// Copyright 2014-2015 The Dename Authors.
//
// 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 server

import (
	"crypto/rand"
	"crypto/sha256"
	"crypto/tls"
	"encoding/binary"
	"log"
	"net"
	"runtime"
	"sync"
	"time"

	"golang.org/x/net/context"

	"github.com/agl/ed25519"
	"github.com/andres-erbsen/clock"
	"github.com/yahoo/coname/common"
	"github.com/yahoo/coname/common/vrf"
	"github.com/yahoo/coname/proto"
	"github.com/yahoo/coname/server/kv"
	"github.com/yahoo/coname/server/replication"
	"github.com/yahoo/coname/server/replication/kvlog"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

// Config encapsulates everything that needs to be specified about a single
// replica of one realm's keyserver to operate it.
// TODO: make this a protobuf, Unmarshal from JSON
type Config struct {
	Realm string

	ServerID, ReplicaID uint64
	RatificationKey     *[ed25519.PrivateKeySize]byte // [32]byte: secret; [32]byte: public
	VRFSecret           *[vrf.SecretKeySize]byte

	UpdateAddr, LookupAddr, VerifierAddr string
	UpdateTLS, LookupTLS, VerifierTLS    *tls.Config

	MinEpochInterval, MaxEpochInterval, RetryEpochInterval time.Duration
	// FIXME: tls.Config is not serializable, replicate relevant fields
}

// Keyserver manages a single end-to-end keyserver realm.
type Keyserver struct {
	realm               string
	serverID, replicaID uint64

	thresholdSigningIndex uint32
	sehKey                *[ed25519.PrivateKeySize]byte
	vrfSecret             *[vrf.SecretKeySize]byte

	db  kv.DB
	log replication.LogReplicator
	rs  proto.ReplicaState

	updateServer, lookupServer, verifierServer *grpc.Server
	updateListen, lookupListen, verifierListen net.Listener

	minEpochInterval, maxEpochInterval, retryEpochInterval time.Duration

	clk clock.Clock
	// state used for determining whether we should start a new epoch.
	// see replication.proto for explanation.
	leaderHint, canEpoch, mustEpoch       bool
	leaderHintSet                         <-chan bool
	canEpochSet, mustEpochSet, retryEpoch *clock.Timer
	// rs.PendingUpdates is used as well

	vmb *VerifierBroadcast
	wr  *WaitingRoom

	stopOnce sync.Once
	stop     chan struct{}
	waitStop sync.WaitGroup
}

// Open initializes a new keyserver based on cfg, reads the persistent state and
// binds to the specified ports. It does not handle input: requests will block.
func Open(cfg *Config, db kv.DB, clk clock.Clock) (ks *Keyserver, err error) {
	log, err := kvlog.New(db, []byte{tableReplicationLogPrefix})
	if err != nil {
		return nil, err
	}

	ks = &Keyserver{
		realm:              cfg.Realm,
		serverID:           cfg.ServerID,
		replicaID:          cfg.ReplicaID,
		sehKey:             cfg.RatificationKey,
		vrfSecret:          cfg.VRFSecret,
		minEpochInterval:   cfg.MinEpochInterval,
		maxEpochInterval:   cfg.MaxEpochInterval,
		retryEpochInterval: cfg.RetryEpochInterval,
		db:                 db,
		log:                log,
		stop:               make(chan struct{}),
		wr:                 NewWaitingRoom(),

		// TODO: change when using actual replication
		leaderHint:    true,
		leaderHintSet: nil,

		clk:          clk,
		canEpochSet:  clk.Timer(0),
		mustEpochSet: clk.Timer(0),
		retryEpoch:   clk.Timer(0),
	}

	switch replicaStateBytes, err := db.Get(tableReplicaState); err {
	case ks.db.ErrNotFound():
		// ReplicaState zero value is valid initialization
	case nil:
		if err := ks.rs.Unmarshal(replicaStateBytes); err != nil {
			return nil, err
		}
	default:
		return nil, err
	}
	ks.resetEpochTimers(ks.rs.LastEpochDelimiter.Timestamp.Time())

	ks.vmb = NewVerifierBroadcast(ks.rs.NextIndexVerifier)

	if cfg.UpdateAddr != "" {
		ks.updateServer = grpc.NewServer(grpc.Creds(credentials.NewTLS(cfg.UpdateTLS)))
		proto.RegisterE2EKSUpdateServer(ks.updateServer, ks)
		ks.updateListen, err = net.Listen("tcp", cfg.UpdateAddr)
		if err != nil {
			return nil, err
		}
	}
	if cfg.LookupAddr != "" {
		ks.lookupServer = grpc.NewServer(grpc.Creds(credentials.NewTLS(cfg.LookupTLS)))
		proto.RegisterE2EKSLookupServer(ks.lookupServer, ks)
		ks.lookupListen, err = net.Listen("tcp", cfg.LookupAddr)
		if err != nil {
			ks.updateListen.Close()
			return nil, err
		}
	}
	if cfg.VerifierAddr != "" {
		ks.verifierServer = grpc.NewServer(grpc.Creds(credentials.NewTLS(cfg.VerifierTLS)))
		proto.RegisterE2EKSVerificationServer(ks.verifierServer, ks)
		ks.verifierListen, err = net.Listen("tcp", cfg.VerifierAddr)
		if err != nil {
			ks.updateListen.Close()
			ks.lookupListen.Close()
			return nil, err
		}
	}
	return ks, nil
}

// Start makes the keyserver start handling requests (forks goroutines).
func (ks *Keyserver) Start() {
	ks.log.Start(ks.rs.NextIndexLog)
	if ks.updateServer != nil {
		go ks.updateServer.Serve(ks.updateListen)
	}
	if ks.lookupServer != nil {
		go ks.lookupServer.Serve(ks.lookupListen)
	}
	if ks.verifierServer != nil {
		go ks.verifierServer.Serve(ks.verifierListen)
	}
	ks.waitStop.Add(1)
	go func() { ks.run(); ks.waitStop.Done() }()
}

// Stop cleanly shuts down the keyserver and then returns.
// TODO: figure out what will happen to connected clients?
func (ks *Keyserver) Stop() {
	ks.stopOnce.Do(func() {
		if ks.updateServer != nil {<commit_msg>less brittle close on error pattern in Keyserver.Open<commit_after>// Copyright 2014-2015 The Dename Authors.
//
// 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 server

import (
	"crypto/rand"
	"crypto/sha256"
	"crypto/tls"
	"encoding/binary"
	"log"
	"net"
	"runtime"
	"sync"
	"time"

	"golang.org/x/net/context"

	"github.com/agl/ed25519"
	"github.com/andres-erbsen/clock"
	"github.com/yahoo/coname/common"
	"github.com/yahoo/coname/common/vrf"
	"github.com/yahoo/coname/proto"
	"github.com/yahoo/coname/server/kv"
	"github.com/yahoo/coname/server/replication"
	"github.com/yahoo/coname/server/replication/kvlog"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

// Config encapsulates everything that needs to be specified about a single
// replica of one realm's keyserver to operate it.
// TODO: make this a protobuf, Unmarshal from JSON
type Config struct {
	Realm string

	ServerID, ReplicaID uint64
	RatificationKey     *[ed25519.PrivateKeySize]byte // [32]byte: secret; [32]byte: public
	VRFSecret           *[vrf.SecretKeySize]byte

	UpdateAddr, LookupAddr, VerifierAddr string
	UpdateTLS, LookupTLS, VerifierTLS    *tls.Config

	MinEpochInterval, MaxEpochInterval, RetryEpochInterval time.Duration
	// FIXME: tls.Config is not serializable, replicate relevant fields
}

// Keyserver manages a single end-to-end keyserver realm.
type Keyserver struct {
	realm               string
	serverID, replicaID uint64

	thresholdSigningIndex uint32
	sehKey                *[ed25519.PrivateKeySize]byte
	vrfSecret             *[vrf.SecretKeySize]byte

	db  kv.DB
	log replication.LogReplicator
	rs  proto.ReplicaState

	updateServer, lookupServer, verifierServer *grpc.Server
	updateListen, lookupListen, verifierListen net.Listener

	minEpochInterval, maxEpochInterval, retryEpochInterval time.Duration

	clk clock.Clock
	// state used for determining whether we should start a new epoch.
	// see replication.proto for explanation.
	leaderHint, canEpoch, mustEpoch       bool
	leaderHintSet                         <-chan bool
	canEpochSet, mustEpochSet, retryEpoch *clock.Timer
	// rs.PendingUpdates is used as well

	vmb *VerifierBroadcast
	wr  *WaitingRoom

	stopOnce sync.Once
	stop     chan struct{}
	waitStop sync.WaitGroup
}

// Open initializes a new keyserver based on cfg, reads the persistent state and
// binds to the specified ports. It does not handle input: requests will block.
func Open(cfg *Config, db kv.DB, clk clock.Clock) (ks *Keyserver, err error) {
	log, err := kvlog.New(db, []byte{tableReplicationLogPrefix})
	if err != nil {
		return nil, err
	}

	ks = &Keyserver{
		realm:              cfg.Realm,
		serverID:           cfg.ServerID,
		replicaID:          cfg.ReplicaID,
		sehKey:             cfg.RatificationKey,
		vrfSecret:          cfg.VRFSecret,
		minEpochInterval:   cfg.MinEpochInterval,
		maxEpochInterval:   cfg.MaxEpochInterval,
		retryEpochInterval: cfg.RetryEpochInterval,
		db:                 db,
		log:                log,
		stop:               make(chan struct{}),
		wr:                 NewWaitingRoom(),

		// TODO: change when using actual replication
		leaderHint:    true,
		leaderHintSet: nil,

		clk:          clk,
		canEpochSet:  clk.Timer(0),
		mustEpochSet: clk.Timer(0),
		retryEpoch:   clk.Timer(0),
	}

	switch replicaStateBytes, err := db.Get(tableReplicaState); err {
	case ks.db.ErrNotFound():
		// ReplicaState zero value is valid initialization
	case nil:
		if err := ks.rs.Unmarshal(replicaStateBytes); err != nil {
			return nil, err
		}
	default:
		return nil, err
	}
	ks.resetEpochTimers(ks.rs.LastEpochDelimiter.Timestamp.Time())

	ks.vmb = NewVerifierBroadcast(ks.rs.NextIndexVerifier)

	ok := false
	if cfg.UpdateAddr != "" {
		ks.updateServer = grpc.NewServer(grpc.Creds(credentials.NewTLS(cfg.UpdateTLS)))
		proto.RegisterE2EKSUpdateServer(ks.updateServer, ks)
		ks.updateListen, err = net.Listen("tcp", cfg.UpdateAddr)
		if err != nil {
			return nil, err
		}
		defer func() {
			if !ok {
				ks.updateListen.Close()
			}
		}()
	}
	if cfg.LookupAddr != "" {
		ks.lookupServer = grpc.NewServer(grpc.Creds(credentials.NewTLS(cfg.LookupTLS)))
		proto.RegisterE2EKSLookupServer(ks.lookupServer, ks)
		ks.lookupListen, err = net.Listen("tcp", cfg.LookupAddr)
		if err != nil {
			return nil, err
		}
		defer func() {
			if !ok {
				ks.lookupListen.Close()
			}
		}()
	}
	if cfg.VerifierAddr != "" {
		ks.verifierServer = grpc.NewServer(grpc.Creds(credentials.NewTLS(cfg.VerifierTLS)))
		proto.RegisterE2EKSVerificationServer(ks.verifierServer, ks)
		ks.verifierListen, err = net.Listen("tcp", cfg.VerifierAddr)
		if err != nil {
			return nil, err
		}
		defer func() {
			if !ok {
				ks.verifierListen.Close()
			}
		}()
	}
	ok = true
	return ks, nil
}

// Start makes the keyserver start handling requests (forks goroutines).
func (ks *Keyserver) Start() {
	ks.log.Start(ks.rs.NextIndexLog)
	if ks.updateServer != nil {
		go ks.updateServer.Serve(ks.updateListen)
	}
	if ks.lookupServer != nil {
		go ks.lookupServer.Serve(ks.lookupListen)
	}
	if ks.verifierServer != nil {
		go ks.verifierServer.Serve(ks.verifierListen)
	}
	ks.waitStop.Add(1)
	go func() { ks.run(); ks.waitStop.Done() }()
}

// Stop cleanly shuts down the keyserver and then returns.
// TODO: figure out what will happen to connected clients?
func (ks *Keyserver) Stop() {
	ks.stopOnce.Do(func() {
		if ks.updateServer != nil {
{'fraction_non_alphanumeric': 0.10852512714421171, 'fraction_numerical': 0.00810274976295147, 'mean_word_length': 3.4486196319018405, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5584201', 'n_tokens_mistral': 4125, 'n_tokens_neox': 3869, 'n_words': 1001}