quantumlayer-emu-prng.go

package quantumlayer

/*
 *
 * This packages "implements", actually it only emulates, the output of a
 * quantum link, i.e., the transmitted random numbers from one quantum
 * sender to a quantum receiver.
 * This relies on crypto/rand to generate the random numbers that will be
 * transmitted to the other end.
 *
 */
import (
	"context"
	"crypto/rand"
	"encoding/json"
	"errors"
	"io"
	"math/big"
	"net"
	"sync"
	"time"

	"github.com/sirupsen/logrus"
	logi "github.com/sirupsen/logrus"
)

type QuantumPayloadElement struct {
	BulkKeyId     int64   `json:"bulk-key-id"`     // the unique ID of this bulk of keys
	BulkKeyLength int     `json:"bulk-key-length"` // the length, counted in bytes, of bulkKey
	BulkKey       *[]byte `json:"bulk-key"`        // the bulk key
}

type QuantumlayerEmuPRNG struct {
	configured       bool
	poweron          bool
	incomingRandNums chan QuantumPayloadElement
	outgoingRandNums chan QuantumPayloadElement
	peerNumbers      *NumberStore
	myNumbers        *NumberStore
	localQLAddress   string
	udpSrvConn       *net.UDPConn
	qlPeer           string
	qlPeerCancel     context.CancelFunc
	qlLocalPort      *net.UDPAddr
	qlPeerMutex      sync.Mutex
}

// We use our own logrus instance, as we would like to have different log levels for different parts.
var log = logrus.New()

func NewQuantumlayerEmuPRNG(logOutput io.Writer, logLevel logi.Level, logInJson bool) (newql *QuantumlayerEmuPRNG) {
	/*
	 * Setup logging
	 */

	//What level
	log.SetLevel(logLevel)
	// Where to send log out put
	log.SetOutput(logOutput)
	// and plain-text (standard) or json
	if !logInJson {
		log.SetFormatter(&logi.TextFormatter{})
	} else {
		log.SetFormatter(&logi.JSONFormatter{})
	}
	// print code function if level is set to Trace
	if logLevel == logi.TraceLevel {
		log.SetReportCaller(true)
	} else {
		log.SetReportCaller(false)
	}

	// Return PRNG Quantum Layer
	return &QuantumlayerEmuPRNG{
		configured:       false,
		poweron:          false,
		incomingRandNums: make(chan QuantumPayloadElement),
		outgoingRandNums: make(chan QuantumPayloadElement),
		peerNumbers:      NewNumberStore(40000),
		myNumbers:        NewNumberStore(40000),
		qlPeer:           "",
	}
}

// Configure the quantum emulation, but do not start if yet
func (qlemuprng *QuantumlayerEmuPRNG) Configure(localQLAddress ...string) {

	// Start receiving numberstores
	go qlemuprng.peerNumbers.receiveNumbers(qlemuprng.incomingRandNums)
	go qlemuprng.myNumbers.receiveNumbers(qlemuprng.outgoingRandNums)

	// Determine if a local UDP address should be used or not
	if len(localQLAddress) == 0 {
		// No input
		qlemuprng.localQLAddress = ":0"
	} else {
		qlemuprng.localQLAddress = localQLAddress[0]
	}
	qlemuprng.configured = true
}

// Power on the quantum layer, i.e., open up the communication ports for the
// other quantum module
func (qlemuprng *QuantumlayerEmuPRNG) PowerOn() {
	if qlemuprng.configured == false {
		// nothing do here move on
		log.Errorf("QuantumlayerEmuPRNG: Sorry, the quantum layer is not configured for action. You've missed Configure()")
		return
	}
	//qlemuprng.poweron = false
	log.Infof("QuantumlayerEmuPRNG: is powering on...charging.")

	// serve UDP incoming
	if qlemuprng.udpSrvConn == nil {
		go func() {
			// Get UDP server part going...
			log.Debugf("QuantumlayerEmuPRNG: localQLAddress is %s", qlemuprng.localQLAddress)

			// This reads random numbers from other Quantum end
			udpSrvPort, err := net.ResolveUDPAddr("udp", qlemuprng.localQLAddress)
			if err != nil {
				log.Fatalf("QuantumlayerEmuPRNG: UDP failure: %s", err)
				return
			}

			qlemuprng.udpSrvConn, err = net.ListenUDP("udp", udpSrvPort)
			if err != nil {
				log.Fatalf("QuantumlayerEmuPRNG: UDP failure: %s", err)
				return
			}
			defer qlemuprng.udpSrvConn.Close()

			// Retrieve local UDP address and store it for further actions.
			qlemuprng.qlLocalPort = qlemuprng.udpSrvConn.LocalAddr().(*net.UDPAddr)

			log.Infof("QuantumlayerEmuPRNG: started server, waiting for incoming rands on port %s \n", qlemuprng.udpSrvConn.LocalAddr().(*net.UDPAddr).String())
			inBuffer := make([]byte, 1500)
			for {
				// Buffer for reading from "Quantum link"
				n, addr, err := qlemuprng.udpSrvConn.ReadFromUDP(inBuffer)
				if err != nil {
					log.Errorf("QuantumlayerEmuPRNG: Could not read from UDP: %s", err)
				} else {
					log.Debugf("QuantumlayerEmuPRNG: read %d bytes from %s\n", n, addr)

					// Check if sender of datagram is qlPeer
					// Warning this is not checking the validity of the sender, i.e., spoofing is possible
					if addr.String() == qlemuprng.qlPeer {
						log.Debugf("QuantumlayerEmuPRNG: Peer %s listed", addr)
						//dumb the received data into the channel and carry on
						// TODO/XXX: no vetting for anything
						// Unmarshall out of JSON
						var inQBuffer QuantumPayloadElement
						unmarshallErr := json.Unmarshal(inBuffer[0:n], &inQBuffer)
						if unmarshallErr == nil {
							qlemuprng.incomingRandNums <- inQBuffer
						}
					} else {
						log.Infof("QuantumlayerEmuPRNG: Peer %s NOT listed", addr)
					}
				}
			}
		}()
	}

	// Wait for listening UDP socket in the above go-routine to get ready
	for qlemuprng.udpSrvConn == nil {
	}

	// Ready, set, go!
	qlemuprng.poweron = true

	log.Infof("QuantumlayerEmuPRNG: is charged and powered on.")
}

// Power off the quantum layer, i.e., close the communication ports for the
// other quantum module
func (qlemuprng *QuantumlayerEmuPRNG) PowerOff() {
	qlemuprng.poweron = false
	log.Println("QuantumlayerEmuPRNG: is powered off...discharging.")
}

func (qlemuprng *QuantumlayerEmuPRNG) AddPeer(addr *net.UDPAddr) {
	if qlemuprng.poweron == false {
		return
	}
	//TODO/XXX check the incoming addr

	// Add  peer to the ....
	qlemuprng.qlPeerMutex.Lock()
	qlemuprng.qlPeer = addr.String()
	qlemuprng.qlPeerMutex.Unlock()

	ctx, cancel := context.WithCancel(context.Background())
	qlemuprng.qlPeerCancel = cancel

	// Start the generation and shipping of random numbers
	go func(ctx context.Context) {
		for {
			select {
			case <-ctx.Done():
				return
			default:
				if qlemuprng.poweron == true {
					// retrieve a new back of random numbers
					newNumberBatch := qlemuprng.GenerateRandomNumbers()
					// TODO: Replace this by some generic encapsulation reader and not just JSON
					//Get JSON for transmission ready
					qpe := QuantumPayloadElement{time.Now().UnixNano(), len(newNumberBatch), &newNumberBatch}

					// XXX/TODO: error must be handled
					jsonPayload, err := json.Marshal(qpe)
					if err != nil {
						log.Errorf("QuantumlayerEmuPRNG: json.Marshal error %s", err)
					}

					_, _, err = qlemuprng.udpSrvConn.WriteMsgUDP(jsonPayload, nil, addr)
					if err != nil {
						log.Fatalf("QuantumlayerEmuPRNG: WriteMsgUDPAddrPort failed: %s", err)
					}
					qlemuprng.outgoingRandNums <- qpe
				}
				// TODO: This sleep timer has to replaced by something for clever.
				time.Sleep(5 * time.Second)
			}
		}
	}(ctx)
}

func (qlemuprng *QuantumlayerEmuPRNG) RemovePeer() {
	if qlemuprng.poweron == false {
		return
	}

	// Stop the generation and shipping of random numbers
	qlemuprng.qlPeerCancel()

	// delete peer
	qlemuprng.qlPeerMutex.Lock()
	qlemuprng.qlPeer = ""
	qlemuprng.qlPeerMutex.Unlock()
}

func (qlemuprng *QuantumlayerEmuPRNG) GetLocalQLPort() (myAddr net.UDPAddr) {
	return *qlemuprng.qlLocalPort
}

func (qlemuprng *QuantumlayerEmuPRNG) GenerateRandomNumbers() (randNums []byte) {
	numRands, randError := rand.Int(rand.Reader, big.NewInt(1000))
	if randError != nil {
		log.Fatalf("QuantumlayerEmuPRNG: %s", randError)
		return
	}

	b := make([]byte, numRands.Uint64())
	_, randError = rand.Read(b)
	if randError != nil {
		log.Fatalf("QuantumlayerEmuPRNG: %s", randError)
		return
	}
	return b
}

func (qlemuprng *QuantumlayerEmuPRNG) GetKeyBatchPeer() (QuantumLayerBulkKey, error) {
	return qlemuprng.peerNumbers.GetBulk()
}

func (qlemuprng *QuantumlayerEmuPRNG) GetKeyBatchLocal() (QuantumLayerBulkKey, error) {
	return qlemuprng.myNumbers.GetBulk()
}

func (qlemuprng *QuantumlayerEmuPRNG) GetStatus() (poweredOn bool) {
	return qlemuprng.poweron
}

type NumberStore struct {
	mu             sync.Mutex
	maxBytes       int
	storage        []byte
	bulkKeyStorage []QuantumLayerBulkKey
	topOfStorage   int
}

// Generates a new store with given maximum number of bytes
func NewNumberStore(maxBytes int) (newNS *NumberStore) {
	return &NumberStore{
		maxBytes:     maxBytes,
		storage:      make([]byte, maxBytes),
		topOfStorage: 0,
	}
}

func (store *NumberStore) GetBulk() (bulk QuantumLayerBulkKey, err error) {
	store.mu.Lock()
	defer store.mu.Unlock()

	for nextID := range store.bulkKeyStorage {
		next := store.bulkKeyStorage[nextID]
		// Preprare to return
		a := QuantumLayerBulkKey{
			BulkKeyId:     next.BulkKeyId,
			BulkKeyLength: next.BulkKeyLength,
			BulkKey:       nil,
		}

		bulkKey := make([]byte, next.BulkKeyLength)
		copy(bulkKey, *next.BulkKey)
		a.BulkKey = &bulkKey

		// Delete from key store
		store.bulkKeyStorage = store.bulkKeyStorage[1:]
		// and return
		return a, nil
	}
	returnErr := errors.New("no bulk key to retrieve")
	b := QuantumLayerBulkKey{}
	return b, returnErr
}

func (store *NumberStore) GetBatch() (batch []byte) {
	store.mu.Lock()
	defer store.mu.Unlock()

	if store.topOfStorage != 0 {
		log.Debugf("QuantumlayerEmuPRNG: Have Storage in my belly")
	}
	// prepare to return full batch of numbers
	batchReturn := make([]byte, store.topOfStorage)
	copy(batchReturn, store.storage)
	store.topOfStorage = 0

	return batchReturn
}

func (store *NumberStore) receiveNumbers(incoming chan QuantumPayloadElement) {
	for {
		receivedNumbers := <-incoming
		// add received to our buffer, if buffer still last
		store.mu.Lock()
		mem := QuantumLayerBulkKey{
			BulkKeyId:     receivedNumbers.BulkKeyId,
			BulkKeyLength: receivedNumbers.BulkKeyLength,
			BulkKey:       receivedNumbers.BulkKey,
		}
		//store.bulkKeyStorage[receivedNumbers.BulkKeyId] = mem
		store.bulkKeyStorage = append(store.bulkKeyStorage, mem)

		store.mu.Unlock()
	}
}