Merge pull request #153 from hashcloak/fix-142

Close db connection when shutdown

.github/workflows/integration.yml

@@ -14,13 +14,11 @@ jobs:
 
     - name: Build Server Container
       run: |
-        cd server
-        make build-docker
+        make build-docker-server
 
     - name: Build Katzenmint Container
       run: |
-        cd katzenmint
-        make build-docker
+        make build-docker-katzenmint
 
     # - name: Start local testnet
     #   run: |

.gitignore

@@ -27,4 +27,6 @@ testdb
 data/
 
 # docker genesis documents
-testnet/conf/node*/config/genesis.json
+testnet/conf/node*/config/genesis.json
+
+plugin/ops/__pycache__

katzenmint/Dockerfile → Dockerfile.katzenmint

@@ -6,10 +6,11 @@ RUN apk update && \
     apk add --no-cache git make ca-certificates build-base && \
     update-ca-certificates
 
-WORKDIR /go
+WORKDIR /go/Meson
 
-RUN git clone https://github.com/hashcloak/Meson.git
-RUN cd Meson/katzenmint ; make build ;
+COPY . .
+
+RUN cd katzenmint ; make build ;
 
 FROM alpine
 

server/Dockerfile → Dockerfile.server

@@ -1,6 +1,6 @@
 FROM golang:alpine AS builder
 
-LABEL authors="Christian Muehlhaeuser: [email protected]"
+LABEL authors="Christian Muehlhaeuser: [email protected];"
 
 # Can pass --build-arg warped=true to decrease epoch period
 ARG warped=false
@@ -11,9 +11,10 @@ RUN apk update && \
     apk add --no-cache git make ca-certificates && \
     update-ca-certificates
 
-WORKDIR /go
+WORKDIR /go/Meson
+
+COPY . .
 
-RUN cd /go ; git clone https://github.com/hashcloak/Meson.git
 RUN cd /go/Meson/server && go build -o meson-server -ldflags "$ldflags" cmd/meson-server/*.go
 RUN cd /go ; git clone https://github.com/katzenpost/memspool.git
 RUN cd /go ; cd memspool/server/cmd/memspool ;  go build -ldflags "$ldflags"

Makefile

@@ -0,0 +1,14 @@
+.PHONY: build-docker-katzenmint
+build-docker-katzenmint:
+	docker build --no-cache -t katzenmint/pki -f Dockerfile.katzenmint .
+
+.PHONY: build-docker-server
+build-docker-server:
+	docker build --no-cache -t meson/server -f Dockerfile.server .
+
+.PHONY: build-docker-containers
+build-docker-containers: build-docker-katzenmint build-docker-server
+
+.PHONY: clean-docker-images
+clean-docker-images:
+	docker rmi -f $$(docker images | grep '^<none>' | awk '{print $$3}')

README.md

@@ -1,18 +1,25 @@
 # Meson
-[![Build Status](https://travis-ci.com/hashcloak/Meson.svg?branch=master)](https://travis-ci.com/hashcloak/Meson)
+[![Integration](https://github.com/hashcloak/Meson/actions/workflows/integration.yml/badge.svg)](https://github.com/hashcloak/Meson/actions/workflows/integration.yml)
+[![Plugin](https://github.com/hashcloak/Meson/actions/workflows/plugin.yml/badge.svg)](https://github.com/hashcloak/Meson/actions/workflows/plugin.yml)
+[![Katzenmint](https://github.com/hashcloak/Meson/actions/workflows/katzenmint.yml/badge.svg)](https://github.com/hashcloak/Meson/actions/workflows/katzenmint.yml)
+[![Server](https://github.com/hashcloak/Meson/actions/workflows/server.yml/badge.svg)](https://github.com/hashcloak/Meson/actions/workflows/server.yml)
+[![Client](https://github.com/hashcloak/Meson/actions/workflows/client.yml/badge.svg)](https://github.com/hashcloak/Meson/actions/workflows/client.yml)
+[![pages-build-deployment](https://github.com/hashcloak/Meson/actions/workflows/pages/pages-build-deployment/badge.svg)](https://github.com/hashcloak/Meson/actions/workflows/pages/pages-build-deployment)
 
 This is the main repository related to the Meson project. 
 Meson is a mixnet for cryptocurrency transactions. Meson is based on the [Katzenpost software project](https://katzenpost.mixnetworks.org/).
 
 We have forked important parts of the Katzenpost mix network project in order to better compliment our upcoming changes and our cryptocurrency transaction usecase. Here are the places in which you can expect activity related to Meson's ongoing development:
-- [Katzenmint](https://github.com/hashcloak/katzenmint-pki): An in-progress new PKI design for mix networks
-- [Meson-server](https://github.com/hashcloak/Meson-server): Forked from the original Katzenpost server in which we make changes better suited for our Katzenmint PKI design and upcoming changes to how nodes (providers and mixes) will work in Meson.
-- [Meson-client](https://github.com/hashcloak/Meson-client): Forked from the original Katzenpost client in which we make changes better suited for our Katzenmint PKI design and upcoming changes to how clients will work in Meson.
+- [Katzenmint](https://github.com/hashcloak/Meson/tree/master/katzenmint): An in-progress new PKI design for mix networks
+- [Meson-server](https://github.com/hashcloak/Meson/tree/master/server): Forked from the original Katzenpost server in which we make changes better suited for our Katzenmint PKI design and upcoming changes to how nodes (providers and mixes) will work in Meson.
+- [Meson-client](https://github.com/hashcloak/Meson/tree/master/client) : Forked from the original Katzenpost client in which we make changes better suited for our Katzenmint PKI design and upcoming changes to how clients will work in Meson.
+
 
 ## Docs
-TODO
+
+Plesase go to [docs.mesonmix.net](https://docs.mesonmix.net)
 
 ## Usage
 
-TODO
+Plesase go to [docs.mesonmix.net/docs/](https://docs.mesonmix.net/docs/)
 

client/README.md

@@ -1,5 +1,5 @@
 # Meson-client
-[![Go](https://github.com/hashcloak/Meson/client/actions/workflows/go.yml/badge.svg)](https://github.com/hashcloak/Meson/client/actions/workflows/go.yml)
+[![Client](https://github.com/hashcloak/Meson/actions/workflows/client.yml/badge.svg)](https://github.com/hashcloak/Meson/actions/workflows/client.yml)
 
 A simple client for use with the Meson mixnet software
 

client/client.go

@@ -80,7 +80,7 @@ func AutoRegisterRandomClient(cfg *config.Config) *ecdh.PrivateKey {
 		panic(err)
 	}
 	account := &config.Account{
-		User:           fmt.Sprintf("%x", linkKey.PublicKey().Bytes()),
+		User:           fmt.Sprintf("%s.%s", linkKey.PublicKey().String()[:6], cfg.UpstreamProxy.User),
 		Provider:       registrationProvider.Name,
 		ProviderKeyPin: registrationProvider.IdentityKey,
 	}

client/minclient/connection.go

@@ -508,7 +508,7 @@ func (c *connection) onWireConn(w *wire.Session) {
 					c.log.Debugf("Failed to send RetrieveMessage: %v", wireErr)
 					return
 				}
-				c.log.Debugf("Sent RetrieveMessage: %d", seq)
+				//c.log.Debugf("Sent RetrieveMessage: %d", seq)
 				nrReqs++
 			}
 			fetchDelay = c.c.GetPollInterval()
@@ -536,7 +536,7 @@ func (c *connection) onWireConn(w *wire.Session) {
 			wireErr = newProtocolError("peer send Disconnect")
 			return
 		case *commands.MessageEmpty:
-			c.log.Debugf("Received MessageEmpty: %v", cmd.Sequence)
+			//c.log.Debugf("Received MessageEmpty: %v", cmd.Sequence)
 			if wireErr = checkSeq(cmd.Sequence); wireErr != nil {
 				c.log.Errorf("MessageEmpty sequence unexpected: %v", cmd.Sequence)
 				return

client/minclient/pki.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/hashcloak/Meson/client/pkiclient/epochtime"
+	"github.com/hashcloak/Meson/katzenmint"
 	cpki "github.com/katzenpost/core/pki"
 	"github.com/katzenpost/core/worker"
 	"gopkg.in/op/go-logging.v1"
@@ -100,8 +101,10 @@ func (p *pki) currentDocument() *cpki.Document {
 		p.log.Debugf("Couldn't find epoch: %+v", err)
 		return nil
 	}
-	if d, _ := p.docs.Load(now); d != nil {
-		return d.(*cpki.Document)
+	for i := 0; i < katzenmint.LifeCycle; i++ {
+		if d, _ := p.docs.Load(now - uint64(i)); d != nil {
+			return d.(*cpki.Document)
+		}
 	}
 	p.forceUpdateCh <- true
 	<-p.doneWorkerCh
@@ -142,7 +145,7 @@ func (p *pki) worker() {
 			p.log.Debugf("Couldn't find epoch: %+v", err)
 			continue
 		}
-		epochs := []uint64{now - 1, now}
+		epochs := []uint64{now}
 
 		// Fetch the documents that we are missing.
 		didUpdate := false

client/pkiclient/cache.go

@@ -25,6 +25,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/hashcloak/Meson/katzenmint"
 	"github.com/katzenpost/core/crypto/eddsa"
 	"github.com/katzenpost/core/pki"
 	"github.com/katzenpost/core/worker"
@@ -53,9 +54,9 @@ type Cache struct {
 	docs map[uint64]*list.Element
 	lru  list.List
 
-	lastSyncTime time.Time
-	memEpoch     uint64
-	memHeight    uint64
+	timer     *time.Timer
+	memEpoch  uint64
+	memHeight uint64
 
 	fetchQueue chan *fetchOp
 }
@@ -69,6 +70,7 @@ type fetchOp struct {
 // Halt tears down the Client instance.
 func (c *Cache) Halt() {
 	c.Worker.Halt()
+	c.timer.Stop()
 
 	// Clean out c.fetchQueue.
 	for {
@@ -83,16 +85,7 @@ func (c *Cache) Halt() {
 
 // GetEpoch returns the epoch information of PKI.
 func (c *Cache) GetEpoch(ctx context.Context) (epoch uint64, ellapsedHeight uint64, err error) {
-	if time.Now().Before(c.lastSyncTime.Add(epochRetrieveInterval)) {
-		return c.memEpoch, c.memHeight, nil
-	}
-	epoch, ellapsedHeight, err = c.impl.GetEpoch(ctx)
-	if err == nil {
-		c.memEpoch = epoch
-		c.memHeight = ellapsedHeight
-		c.lastSyncTime = time.Now()
-	}
-	return
+	return c.memEpoch, c.memHeight, nil
 }
 
 // GetDoc returns the PKI document for the provided epoch.
@@ -102,14 +95,21 @@ func (c *Cache) GetDoc(ctx context.Context, epoch uint64) (*pki.Document, []byte
 		return d.doc, d.raw, nil
 	}
 
+	// Exit upon halt
+	select {
+	case <-c.HaltCh():
+		return nil, nil, fmt.Errorf("pki client is halted, cannot get new document")
+	default:
+	}
+
+	// Slow path
 	op := &fetchOp{
 		ctx:    ctx,
 		epoch:  epoch,
 		doneCh: make(chan interface{}),
 	}
 	c.fetchQueue <- op
-	v := <-op.doneCh
-	switch r := v.(type) {
+	switch r := (<-op.doneCh).(type) {
 	case error:
 		return nil, nil, r
 	case *cacheEntry:
@@ -122,7 +122,7 @@ func (c *Cache) GetDoc(ctx context.Context, epoch uint64) (*pki.Document, []byte
 
 // Post posts the node's descriptor to the PKI for the provided epoch.
 func (c *Cache) Post(ctx context.Context, epoch uint64, signingKey *eddsa.PrivateKey, d *pki.MixDescriptor) error {
-	return errNotSupported
+	return c.impl.Post(ctx, epoch, signingKey, d)
 }
 
 // Deserialize returns PKI document given the raw bytes.
@@ -159,49 +159,86 @@ func (c *Cache) insertLRU(newEntry *cacheEntry) {
 }
 
 func (c *Cache) worker() {
+	const retryTime = time.Second / 2
+
+	var epoch, height uint64
+	var ctx context.Context
+	var err error
 	for {
 		var op *fetchOp
 		select {
 		case <-c.HaltCh():
 			return
+		case <-c.timer.C:
+			if c.memHeight < uint64(katzenmint.EpochInterval) {
+				c.memHeight++
+				c.timer.Reset(katzenmint.HeightPeriod)
+				continue
+			}
+			ctx = context.Background()
+			epoch, height, err = c.impl.GetEpoch(context.Background())
+			if epoch == c.memEpoch && height < uint64(katzenmint.EpochInterval) {
+				c.memHeight = height
+			}
+			if err != nil || epoch == c.memEpoch {
+				c.timer.Reset(retryTime)
+				continue
+			}
+			c.memEpoch = epoch
+			c.memHeight = height
+			c.timer.Reset(katzenmint.HeightPeriod)
 		case op = <-c.fetchQueue:
+			ctx = op.ctx
+			epoch = op.epoch
 		}
 
 		// The fetch may have been in progress while the op was sitting in
 		// queue, check again.
-		if d := c.cacheGet(op.epoch); d != nil {
-			op.doneCh <- d
+		if d := c.cacheGet(epoch); d != nil {
+			if op != nil {
+				op.doneCh <- d
+			}
 			continue
 		}
 
 		// Slow path, have to call into the PKI client.
 		//
 		// TODO: This could allow concurrent fetches at some point, but for
 		// most common client use cases, this shouldn't matter much.
-		d, raw, err := c.impl.GetDoc(op.ctx, op.epoch)
+		d, raw, err := c.impl.GetDoc(ctx, epoch)
 		if err != nil {
-			op.doneCh <- err
+			if op != nil {
+				op.doneCh <- err
+			}
 			continue
 		}
 		e := &cacheEntry{doc: d, raw: raw}
 		c.insertLRU(e)
-		op.doneCh <- e
+		if op != nil {
+			op.doneCh <- e
+		}
 	}
 }
 
 // New constructs a new Client backed by an existing pki.Client instance.
-func NewCacheClient(impl Client) *Cache {
+func NewCacheClient(impl Client) (*Cache, error) {
+	var err error
 	c := new(Cache)
 	c.impl = impl
 	c.docs = make(map[uint64]*list.Element)
 	c.fetchQueue = make(chan *fetchOp, fetchBacklog)
-	c.lastSyncTime = time.Time{}
+	c.memEpoch, c.memHeight, err = c.impl.GetEpoch(context.Background())
+	if err != nil {
+		return nil, err
+	}
+	c.timer = time.NewTimer(katzenmint.HeightPeriod)
 
 	c.Go(c.worker)
-	return c
+	return c, nil
 }
 
 // Shutdown the client
 func (c *Cache) Shutdown() {
 	c.Halt()
+	c.impl.Shutdown()
 }

client/pkiclient/katzenmint_test.go

@@ -103,7 +103,9 @@ func TestMockPKIClientGetDocument(t *testing.T) {
 	tree, err := iavl.NewMutableTree(dbm.NewMemDB(), 100)
 	require.NoError(err)
 
-	tree.Set(key, value)
+	isUpdated, err := tree.Set(key, value)
+	require.NoError(err)
+	require.False(isUpdated)
 
 	rawDoc, proof, err := tree.GetWithProof(key)
 	require.NoError(err)

client/send.go

@@ -112,7 +112,7 @@ func (s *Session) sendDropDecoy() {
 		s.fatalErrCh <- errors.New("failure to get loop service")
 		return
 	}
-	payload := [constants.UserForwardPayloadLength]byte{}
+	payload := make([]byte, constants.UserForwardPayloadLength)
 	id := [cConstants.MessageIDLength]byte{}
 	_, err = io.ReadFull(rand.Reader, id[:])
 	if err != nil {
@@ -137,7 +137,7 @@ func (s *Session) sendLoopDecoy() {
 		s.fatalErrCh <- errors.New("failure to get loop service")
 		return
 	}
-	payload := [constants.UserForwardPayloadLength]byte{}
+	payload := make([]byte, constants.UserForwardPayloadLength)
 	id := [cConstants.MessageIDLength]byte{}
 	_, err = io.ReadFull(rand.Reader, id[:])
 	if err != nil {
@@ -161,7 +161,7 @@ func (s *Session) composeMessage(recipient, provider string, message []byte, isB
 	if len(message) > constants.UserForwardPayloadLength-4 {
 		return nil, fmt.Errorf("invalid message size: %v", len(message))
 	}
-	payload := [constants.UserForwardPayloadLength]byte{}
+	payload := make([]byte, constants.UserForwardPayloadLength)
 	binary.BigEndian.PutUint32(payload[:4], uint32(len(message)))
 	copy(payload[4:], message)
 	id := [cConstants.MessageIDLength]byte{}