Merge pull request 'New Cwtch Tool App, Extract MakePayment into a standalone function, Fix race condition in engine.' (#468) from indents_and_tools into master

Reviewed-on: #468

.gitignore

@@ -29,3 +29,6 @@ arch/
 testing/encryptedstorage/encrypted_storage_profiles
 testing/encryptedstorage/tordir
 *.tar.gz
+data-dir-cwtchtool/
+tokens
+tordir/

protocol/connections/engine.go

@@ -7,6 +7,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"cwtch.im/cwtch/event"
@@ -63,7 +64,7 @@ type engine struct {
 
 	tokenManagers sync.Map // [tokenService][]TokenManager
 
-	shuttingDown bool
+	shuttingDown atomic.Bool
 }
 
 // Engine (ProtocolEngine) encapsulates the logic necessary to make and receive Cwtch connections.
@@ -139,7 +140,7 @@ func (e *engine) eventHandler() {
 	for {
 		ev := e.queue.Next()
 		// optimistic shutdown...
-		if e.shuttingDown {
+		if e.shuttingDown.Load() {
 			return
 		}
 		switch ev.EventType {
@@ -297,7 +298,7 @@ func (e *engine) createPeerTemplate() *PeerApp {
 // Listen sets up an onion listener to process incoming cwtch messages
 func (e *engine) listenFn() {
 	err := e.service.Listen(e.createPeerTemplate())
-	if !e.shuttingDown {
+	if !e.shuttingDown.Load() {
 		e.eventManager.Publish(event.NewEvent(event.ProtocolEngineStopped, map[event.Field]string{event.Identity: e.identity.Hostname(), event.Error: err.Error()}))
 	}
 }
@@ -307,7 +308,8 @@ func (e *engine) Shutdown() {
 	// don't accept any more events...
 	e.queue.Publish(event.NewEvent(event.ProtocolEngineShutdown, map[event.Field]string{}))
 	e.service.Shutdown()
-	e.shuttingDown = true
+
+	e.shuttingDown.Store(true)
 
 	e.ephemeralServicesLock.Lock()
 	defer e.ephemeralServicesLock.Unlock()
@@ -368,7 +370,7 @@ func (e *engine) makeAntispamPayment(onion string) {
 			tokenManager := tokenManagerPointer.(*TokenManager)
 			log.Debugf("checking antispam tokens %v", tokenManager.NumTokens())
 			if tokenManager.NumTokens() < 5 {
-				go tokenApp.MakePayment()
+				go tokenApp.PurchaseTokens()
 			}
 		}
 	}
@@ -431,7 +433,7 @@ func (e *engine) peerWithTokenServer(onion string, tokenServerOnion string, toke
 
 func (e *engine) ignoreOnShutdown(f func(string)) func(string) {
 	return func(x string) {
-		if !e.shuttingDown {
+		if !e.shuttingDown.Load() {
 			f(x)
 		}
 	}
@@ -439,7 +441,7 @@ func (e *engine) ignoreOnShutdown(f func(string)) func(string) {
 
 func (e *engine) ignoreOnShutdown2(f func(string, string)) func(string, string) {
 	return func(x, y string) {
-		if !e.shuttingDown {
+		if !e.shuttingDown.Load() {
 			f(x, y)
 		}
 	}
@@ -615,7 +617,7 @@ func (e *engine) sendMessageToGroup(groupID string, server string, ct []byte, si
 		if ok {
 			if spent, numtokens := tokenApp.Post(ct, sig); !spent {
 				// we failed to post, probably because we ran out of tokens... so make a payment
-				go tokenApp.MakePayment()
+				go tokenApp.PurchaseTokens()
 				// backoff
 				time.Sleep(time.Second * 5)
 				// try again
@@ -623,7 +625,7 @@ func (e *engine) sendMessageToGroup(groupID string, server string, ct []byte, si
 				e.sendMessageToGroup(groupID, server, ct, sig, attempts+1)
 			} else {
 				if numtokens < 5 {
-					go tokenApp.MakePayment()
+					go tokenApp.PurchaseTokens()
 				}
 			}
 			// regardless we return....
@@ -696,7 +698,7 @@ func (e *engine) handlePeerMessage(hostname string, eventID string, context stri
 		e.eventManager.Publish(event.NewEvent(event.NewMessageFromPeerEngine, map[event.Field]string{event.TimestampReceived: time.Now().Format(time.RFC3339Nano), event.RemotePeer: hostname, event.Data: string(message)}))
 
 		// Send an explicit acknowledgement
-		// Every other protocol should have a explicit acknowledgement message e.g. value lookups have responses, and file handling has an explicit flow
+		// Every other protocol should have an explicit acknowledgement message e.g. value lookups have responses, and file handling has an explicit flow
 		if err := e.sendPeerMessage(hostname, pmodel.PeerMessage{ID: eventID, Context: event.ContextAck, Data: []byte{}}); err != nil {
 			e.eventManager.Publish(event.NewEvent(event.SendMessageToPeerError, map[event.Field]string{event.RemotePeer: hostname, event.EventID: eventID, event.Error: err.Error()}))
 		}

protocol/connections/make_payment.go

@@ -0,0 +1,59 @@
+package connections
+
+import (
+	"cwtch.im/cwtch/utils"
+	"git.openprivacy.ca/cwtch.im/tapir/applications"
+	"git.openprivacy.ca/cwtch.im/tapir/networks/tor"
+	"git.openprivacy.ca/cwtch.im/tapir/primitives"
+	"git.openprivacy.ca/cwtch.im/tapir/primitives/privacypass"
+	"git.openprivacy.ca/openprivacy/connectivity"
+	"git.openprivacy.ca/openprivacy/log"
+	"reflect"
+	"time"
+)
+
+// MakePayment uses the PoW based token protocol to obtain more tokens
+func MakePayment(tokenServiceOnion string, tokenService *privacypass.TokenServer, acn connectivity.ACN, handler TokenBoardHandler) error {
+	log.Debugf("making a payment")
+	id, sk := primitives.InitializeEphemeralIdentity()
+	client := new(tor.BaseOnionService)
+	client.Init(acn, sk, &id)
+	defer client.Shutdown()
+
+	tokenApplication := new(applications.TokenApplication)
+	tokenApplication.TokenService = tokenService
+	powTokenApp := new(applications.ApplicationChain).
+		ChainApplication(new(applications.ProofOfWorkApplication), applications.SuccessfulProofOfWorkCapability).
+		ChainApplication(tokenApplication, applications.HasTokensCapability)
+
+	log.Debugf("waiting for successful PoW auth...")
+	tp := utils.TimeoutPolicy(time.Second * 30)
+	err := tp.ExecuteAction(func() error {
+		connected, err := client.Connect(tokenServiceOnion, powTokenApp)
+		if connected && err == nil {
+			log.Debugf("waiting for successful token acquisition...")
+			conn, err := client.WaitForCapabilityOrClose(tokenServiceOnion, applications.HasTokensCapability)
+			if err == nil {
+				powtapp, ok := conn.App().(*applications.TokenApplication)
+				if ok {
+					log.Debugf("updating tokens")
+					handler.NewTokenHandler(tokenServiceOnion, powtapp.Tokens)
+					log.Debugf("transcript: %v", powtapp.Transcript().OutputTranscriptToAudit())
+					conn.Close()
+					return nil
+				}
+				log.Errorf("invalid cast of powapp. this should never happen %v %v", powtapp, reflect.TypeOf(conn.App()))
+				return nil
+			}
+			return nil
+		}
+		return err
+	})
+
+	// we timed out
+	if err != nil {
+		log.Debugf("make payment timeout...")
+		return err
+	}
+	return err
+}

protocol/connections/tokenboardclientapp.go

@@ -2,18 +2,13 @@ package connections
 
 import (
 	"cwtch.im/cwtch/protocol/groups"
-	"cwtch.im/cwtch/utils"
 	"encoding/json"
 	"git.openprivacy.ca/cwtch.im/tapir"
 	"git.openprivacy.ca/cwtch.im/tapir/applications"
-	"git.openprivacy.ca/cwtch.im/tapir/networks/tor"
-	"git.openprivacy.ca/cwtch.im/tapir/primitives"
 	"git.openprivacy.ca/cwtch.im/tapir/primitives/privacypass"
 	"git.openprivacy.ca/openprivacy/connectivity"
 	"git.openprivacy.ca/openprivacy/log"
 	"github.com/gtank/ristretto255"
-	"reflect"
-	"time"
 )
 
 // TokenBoardHandler encapsulates all the various handlers a client needs to interact with a token board
@@ -73,8 +68,8 @@ func (ta *TokenBoardClient) Init(connection tapir.Connection) {
 		log.Debugf("Successfully Initialized Connection to %v", connection.Hostname())
 		go ta.Listen()
 		// Optimistically acquire many tokens for this server...
-		go ta.MakePayment()
-		go ta.MakePayment()
+		go ta.PurchaseTokens()
+		go ta.PurchaseTokens()
 		ta.Replay()
 	} else {
 		connection.Close()
@@ -152,7 +147,7 @@ func (ta *TokenBoardClient) Replay() {
 
 // PurchaseTokens purchases the given number of tokens from the server (using the provided payment handler)
 func (ta *TokenBoardClient) PurchaseTokens() {
-	ta.MakePayment()
+	MakePayment(ta.tokenServiceOnion, ta.tokenService, ta.acn, ta.tokenBoardHandler)
 }
 
 // Post sends a Post Request to the server
@@ -172,51 +167,6 @@ func (ta *TokenBoardClient) Post(ct []byte, sig []byte) (bool, int) {
 	return false, numTokens
 }
 
-// MakePayment uses the PoW based token protocol to obtain more tokens
-func (ta *TokenBoardClient) MakePayment() error {
-	log.Debugf("Making a Payment")
-	id, sk := primitives.InitializeEphemeralIdentity()
-	client := new(tor.BaseOnionService)
-	client.Init(ta.acn, sk, &id)
-	defer client.Shutdown()
-
-	tokenApplication := new(applications.TokenApplication)
-	tokenApplication.TokenService = ta.tokenService
-	powTokenApp := new(applications.ApplicationChain).
-		ChainApplication(new(applications.ProofOfWorkApplication), applications.SuccessfulProofOfWorkCapability).
-		ChainApplication(tokenApplication, applications.HasTokensCapability)
-
-	log.Debugf("Waiting for successful PoW Auth...")
-
-	connected, err := client.Connect(ta.tokenServiceOnion, powTokenApp)
-	if connected && err == nil {
-		log.Debugf("Waiting for successful Token Acquisition...")
-		tp := utils.TimeoutPolicy(time.Second * 30)
-		err := tp.ExecuteAction(func() error {
-			conn, err := client.WaitForCapabilityOrClose(ta.tokenServiceOnion, applications.HasTokensCapability)
-			if err == nil {
-				powtapp, ok := conn.App().(*applications.TokenApplication)
-				if ok {
-					log.Debugf("Updating Tokens")
-					ta.tokenBoardHandler.NewTokenHandler(ta.tokenServiceOnion, powtapp.Tokens)
-					log.Debugf("Transcript: %v", powtapp.Transcript().OutputTranscriptToAudit())
-					conn.Close()
-					return nil
-				}
-				log.Errorf("invalid cast of powapp. this should never happen %v %v", powtapp, reflect.TypeOf(conn.App()))
-				return nil
-			}
-			return nil
-		})
-
-		// we timed out
-		if err != nil {
-			ta.connection.Close()
-		}
-	}
-	return err
-}
-
 // NextToken retrieves the next token
 func (ta *TokenBoardClient) NextToken(data []byte, hostname string) (privacypass.SpentToken, int, error) {
 	token, numtokens, err := ta.tokenBoardHandler.FetchToken(ta.tokenServiceOnion)

tools/cwtch_tools.go

@@ -0,0 +1,150 @@
+package main
+
+import (
+	"crypto/rand"
+	"cwtch.im/cwtch/model"
+	"cwtch.im/cwtch/model/constants"
+	"cwtch.im/cwtch/protocol/connections"
+	"cwtch.im/cwtch/protocol/groups"
+	"encoding/base64"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"git.openprivacy.ca/cwtch.im/tapir/primitives/privacypass"
+	"git.openprivacy.ca/openprivacy/connectivity/tor"
+	"git.openprivacy.ca/openprivacy/log"
+	"github.com/gtank/ristretto255"
+	mrand "math/rand"
+	"os"
+	path "path/filepath"
+	"strings"
+	"time"
+)
+
+var tool = flag.String("tool", "", "the tool to use")
+var bundle = flag.String("bundle", "", "a server bundle to parse")
+
+func main() {
+	flag.Parse()
+
+	if *tool == "" {
+		fmt.Fprintf(os.Stderr, "usage: cwtch_tools -tool <tool>\n")
+		flag.PrintDefaults()
+	}
+
+	switch *tool {
+	case "bundleparse":
+		bundle, err := bundleParse(*bundle)
+		if err == nil {
+			fmt.Printf("bundle: %s\n", bundle.Serialize())
+		} else {
+			fmt.Printf("error parsing bundle: %v", err)
+		}
+	case "gettokens":
+		getTokens(*bundle)
+	default:
+		fmt.Fprintf(os.Stderr, "unknown tool: %s \n", *tool)
+	}
+
+}
+
+func bundleParse(bundle string) (*model.KeyBundle, error) {
+	if strings.HasPrefix(bundle, constants.ServerPrefix) {
+		bundle, err := base64.StdEncoding.DecodeString(bundle[len(constants.ServerPrefix):])
+		if err != nil {
+			fmt.Printf("invalid server bundle: %v\n", err)
+		}
+		keyBundle, err := model.DeserializeAndVerify([]byte(bundle))
+		return keyBundle, err
+	} else {
+		return nil, fmt.Errorf("unknown bundle prefix")
+	}
+}
+
+func getTokens(bundle string) {
+	log.SetLevel(log.LevelDebug)
+	keyBundle, err := bundleParse(bundle)
+	if err != nil {
+		log.Errorf("error parsing keybundle: %v", err)
+		return
+	}
+
+	privacyPassKey, err := keyBundle.GetKey(model.KeyTypePrivacyPass)
+	if err != nil {
+		log.Errorf("error parsing keybundle: %v", err)
+		return
+	}
+
+	tokenServiceKey, err := keyBundle.GetKey(model.KeyTypeTokenOnion)
+	if err != nil {
+		log.Errorf("error parsing keybundle: %v", err)
+		return
+	}
+
+	os.Mkdir("tordir", 0700)
+	dataDir := path.Join("tordir", "tor")
+	os.MkdirAll(dataDir, 0700)
+
+	// we don't need real randomness for the port, just to avoid a possible conflict...
+	mrand.Seed(int64(time.Now().Nanosecond()))
+	socksPort := mrand.Intn(1000) + 9051
+	controlPort := mrand.Intn(1000) + 9052
+
+	// generate a random password
+	key := make([]byte, 64)
+	_, err = rand.Read(key)
+	if err != nil {
+		log.Errorf("insufficient randomness: %v", err)
+		return
+	}
+
+	torDataDir := "./data-dir-cwtchtool"
+	if err = os.MkdirAll(torDataDir, 0700); err != nil {
+		log.Errorf("could not create data dir")
+		return
+	}
+
+	tor.NewTorrc().WithSocksPort(socksPort).WithOnionTrafficOnly().WithHashedPassword(base64.StdEncoding.EncodeToString(key)).WithControlPort(controlPort).Build("tordir/tor/torrc")
+	acn, err := tor.NewTorACNWithAuth("./tordir", path.Join("..", "tor"), torDataDir, controlPort, tor.HashedPasswordAuthenticator{Password: base64.StdEncoding.EncodeToString(key)})
+	if err != nil {
+		log.Errorf("Could not start Tor: %v", err)
+		return
+	}
+	acn.WaitTillBootstrapped()
+	defer acn.Close()
+
+	Y := new(ristretto255.Element)
+	Y.UnmarshalText([]byte(privacyPassKey))
+
+	tokenServer := privacypass.NewTokenServer()
+	tokenServer.Y = Y
+
+	err = connections.MakePayment(string(tokenServiceKey), tokenServer, acn, Handler{})
+	if err != nil {
+		log.Errorf("timed out trying to get payments")
+	}
+}
+
+type Handler struct {
+}
+
+func (h Handler) GroupMessageHandler(server string, gm *groups.EncryptedGroupMessage) {
+}
+
+func (h Handler) ServerAuthedHandler(server string) {
+}
+
+func (h Handler) ServerSyncedHandler(server string) {
+}
+
+func (h Handler) ServerClosedHandler(server string) {
+}
+
+func (h Handler) NewTokenHandler(tokenService string, tokens []*privacypass.Token) {
+	data, _ := json.Marshal(tokens)
+	os.WriteFile("tokens", data, 0600)
+}
+
+func (h Handler) FetchToken(tokenService string) (*privacypass.Token, int, error) {
+	return nil, 0, nil
+}

utils/timeoutpolicy.go

@@ -12,7 +12,6 @@ type TimeoutPolicy time.Duration
 // ExecuteAction runs a function and returns an error if it hasn't returned
 // by the time specified by TimeoutPolicy
 func (tp *TimeoutPolicy) ExecuteAction(action func() error) error {
-
 	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(*tp))
 	defer cancel()