Merge branch 'master' into work-fe1-nico-election-notice

be1-go/Makefile

@@ -22,6 +22,9 @@ lint:
 test: protocol
 	go test -v -race ./...
 
+test-no-cache: protocol
+	go test -v -race ./... -count=1
+
 test-cov: protocol
 	go test -v -coverpkg=./... -coverprofile=coverage.out ./... -json > report.json
 

be1-go/channel/chirp/mod_test.go

@@ -348,6 +348,8 @@ func Test_Delete_Chirp(t *testing.T) {
 	// publish add chirp message
 	require.NoError(t, cha.Publish(pub, socket.ClientSocket{}))
 
+	time.Sleep(time.Millisecond)
+
 	// create delete chirp message
 	file = filepath.Join(relativeMsgDataExamplePath, "chirp_delete_publish",
 		"chirp_delete_publish.json")

be1-go/channel/consensus/mod_test.go

@@ -1458,14 +1458,13 @@ func Test_Timeout_Prepare(t *testing.T) {
 	require.NoError(t, err)
 
 	// Verify that a prepare message was sent and empty the socket
-	// Unmarshal the failure message sent to other servers to verify its values
 	var sentPrepare method.Publish
 	err = json.Unmarshal(fakeHub.fakeSock.msg, &sentPrepare)
 	require.NoError(t, err)
 
 	sentMsg := sentPrepare.Params.Message
 
-	// Unmarshal the failure message data to check its values
+	// Unmarshal the prepare message data to check its values
 	jsonData, err := base64.URLEncoding.DecodeString(sentMsg.Data)
 	require.NoError(t, err)
 	var prepare messagedata.ConsensusPrepare

be1-go/channel/election/mod_test.go

@@ -476,13 +476,10 @@ func Test_Sending_Election_Key(t *testing.T) {
 
 	electionKeyMsg := catchupAnswer[1]
 
-	data := messagedata.ElectionKey{}.NewEmpty()
+	var dataKey messagedata.ElectionKey
 
-	err := electionKeyMsg.UnmarshalData(data)
-	require.NoError(t, err)
-
-	dataKey, ok := data.(*messagedata.ElectionKey)
-	require.True(t, ok)
+	err := electionKeyMsg.UnmarshalData(&dataKey)
+	require.NoError(t, err, electionKeyMsg)
 
 	key, err := base64.URLEncoding.DecodeString(dataKey.Key)
 	require.NoError(t, err)

be1-go/channel/lao/mod_test.go

@@ -554,7 +554,7 @@ func TestLAOChannel_Sends_Greeting(t *testing.T) {
 	var laoGreet messagedata.LaoGreet
 
 	err = greetMsg.UnmarshalData(&laoGreet)
-	require.NoError(t, err, laoGreet, catchupAnswer)
+	require.NoError(t, err)
 
 	require.Equal(t, messagedata.LAOObject, laoGreet.Object)
 	require.Equal(t, messagedata.LAOActionGreet, laoGreet.Action)

be1-go/inbox/mod.go

@@ -18,15 +18,17 @@ type messageInfo struct {
 // Inbox represents an in-memory data store to record incoming messages.
 type Inbox struct {
 	mutex     sync.RWMutex
-	msgs      map[string]*messageInfo
+	msgsMap   map[string]*messageInfo
+	msgsArray []*messageInfo
 	channelID string
 }
 
 // NewInbox returns a new initialized inbox
 func NewInbox(channelID string) *Inbox {
 	return &Inbox{
 		mutex:     sync.RWMutex{},
-		msgs:      make(map[string]*messageInfo),
+		msgsMap:   make(map[string]*messageInfo),
+		msgsArray: make([]*messageInfo, 0),
 		channelID: channelID,
 	}
 }
@@ -66,30 +68,25 @@ func (i *Inbox) StoreMessage(msg message.Message) {
 		storedTime: storedTime,
 	}
 
-	i.msgs[msg.MessageID] = messageInfo
+	i.msgsMap[msg.MessageID] = messageInfo
+	i.msgsArray = append(i.msgsArray, messageInfo)
 }
 
 // GetSortedMessages returns all messages stored sorted by stored time.
 func (i *Inbox) GetSortedMessages() []message.Message {
 	i.mutex.RLock()
 	defer i.mutex.RUnlock()
 
-	messages := make([]messageInfo, 0, len(i.msgs))
-	// iterate over map and collect all the values (messageInfo instances)
-	for _, msgInfo := range i.msgs {
-		messages = append(messages, *msgInfo)
-	}
-
 	// sort.Slice on messages based on the timestamp
-	sort.Slice(messages, func(i, j int) bool {
-		return messages[i].storedTime < messages[j].storedTime
+	sort.SliceStable(i.msgsArray, func(k, l int) bool {
+		return i.msgsArray[k].storedTime < i.msgsArray[l].storedTime
 	})
 
-	result := make([]message.Message, len(messages))
+	result := make([]message.Message, len(i.msgsArray))
 
 	// iterate and extract the messages[i].message field and
 	// append it to the result slice
-	for i, msgInfo := range messages {
+	for i, msgInfo := range i.msgsArray {
 		result[i] = msgInfo.message
 	}
 
@@ -102,7 +99,7 @@ func (i *Inbox) GetMessage(messageID string) (*message.Message, bool) {
 	i.mutex.Lock()
 	defer i.mutex.Unlock()
 
-	msgInfo, ok := i.msgs[messageID]
+	msgInfo, ok := i.msgsMap[messageID]
 	if !ok {
 		return nil, false
 	}

be1-go/inbox/mod_test.go

@@ -20,7 +20,7 @@ func TestInbox_AddWitnessSignature(t *testing.T) {
 	// Add a message to the inbox
 	inbox.StoreMessage(msg)
 
-	require.Equal(t, 1, len(inbox.msgs))
+	require.Equal(t, 1, len(inbox.msgsMap))
 
 	// Add the witness signature to the message in the inbox
 	err := inbox.AddWitnessSignature(msg.MessageID, "456", "789")
@@ -47,7 +47,7 @@ func TestInbox_AddSigWrongMessages(t *testing.T) {
 	_, ok := inbox.GetMessage(string(buf))
 	require.False(t, ok)
 
-	require.Equal(t, 0, len(inbox.msgs))
+	require.Equal(t, 0, len(inbox.msgsMap))
 }
 
 func TestInbox_AddWitnessSignatures(t *testing.T) {
@@ -58,7 +58,7 @@ func TestInbox_AddWitnessSignatures(t *testing.T) {
 	// Add a message to the inbox
 	inbox.StoreMessage(msg)
 
-	require.Equal(t, 1, len(inbox.msgs))
+	require.Equal(t, 1, len(inbox.msgsMap))
 
 	signaturesNumber := 100
 	for i := 0; i < signaturesNumber; i++ {

be2-scala/README.md

@@ -65,6 +65,7 @@ The project relies on several sbt dependencies (external libraries) :
 - database : [**leveldb**](https://github.com/codeborui/leveldb-scala) which relies on both [snappy](https://search.maven.org/artifact/org.xerial.snappy/snappy-java/1.1.7.3/jar) (for compression/decompression) and [akka-persistence](https://doc.akka.io/docs/akka/current/persistence.html)
 - Json parser : [**spray-json**](https://github.com/spray/spray-json) for Json encoding/decoding
 - encryption : [**tink**](https://github.com/google/tink/blob/master/docs/JAVA-HOWTO.md) to verify signatures
+- encryption : [**kyber**](https://github.com/dedis/kyber) to encrypt and decrypt messages of an election
 - testing : [**scalatest**](https://www.scalatest.org/) for unit tests
 - Json schema validator : [**networknt**](https://github.com/networknt/json-schema-validator) for Json schema validation
 

be2-scala/docs/README.md

@@ -125,10 +125,10 @@ a source of truth since the validation library checks the messages against it.
 
 ```scala
 register.add(
-  (ObjectType.LAO, ActionType.CREATE), 	
-  SchemaValidator.createSchemaValidator("dataCreateLao.json"), 
-  CreateLao.buildFromJson, 
-  LaoValidator.validateCreateLao, 
+  (ObjectType.LAO, ActionType.CREATE),
+  SchemaValidator.createSchemaValidator("dataCreateLao.json"),
+  CreateLao.buildFromJson,
+  LaoValidator.validateCreateLao,
   LaoHandler.handleCreateLao
 )
 ```
@@ -206,6 +206,8 @@ Summary of the keys used to retrieve data:
 - for a message: `channel#message_id`
 - for ChannelData: `channel`
 - for LaoData: `root/lao_id#laodata`
+- for RollCallData: `root/rollcall/lao_id`
+- for ElectionData: `root/private/election_id`
 
 We use `/` as a separator for parts of a channel and `#` as a separator for data objects when needed.
 
@@ -303,7 +305,26 @@ final case class DbActorReadLaoDataAck(laoData: LaoData) extends DbActorMessage
 
 For the Social Media functionality, each user has their own channel with the identifier `root/lao_id/own_pop_token` and each broadcast containing the message_id of a post will be written to `root/lao_id/posts`.
 
+For the Election functionality, we need to have a key pair stored safely somewhere so that we can encrypt/decrypt messages. That is why we use a `ElectionData` object to store the key pairs for the corresponding election.
+The path root is `root/private/election_id` as stated above.
+The key pair can be stored and retrieved by the following functions.
 
+```scala
+final case class CreateElectionData(id: Hash, keyPair: KeyPair) extends Event
+final case class ReadElectionData(electionId: Hash) extends Event
+
+final case class DbActorReadElectionDataAck(electionData: ElectionData) extends DbActorMessage
+```
+
+The RollCallData is an object that stores the id and state of the latest rollcall action (`CREATE`, `OPEN`, `REOPEN`, or `CLOSE`). It ensures that we cannot open a closed rollcall or close a non-opened rollcall.
+The stored parameters can be modified or retrieved by the following functions.
+
+```scala
+final case class ReadRollCallData(laoId: Hash) extends Event
+final case class WriteRollCallData(laoId: Hash, message: Message) extends Event
+
+final case class DbActorReadRollCallDataAck(rollcallData: RollCallData) extends DbActorMessage
+```
 
 :information_source: the database may easily be reset/purged by deleting the `database` folder entirely. You may add the `-Dclean` flag at compilation for automatic database purge
 

be2-scala/project/build.properties

@@ -1 +1 @@
-sbt.version = 1.6.1
+sbt.version=1.6.1

be2-scala/src/main/scala/ch/epfl/pop/pubsub/graph/handlers/CoinHandler.scala

@@ -1,26 +1,30 @@
 package ch.epfl.pop.pubsub.graph.handlers
 
 import ch.epfl.pop.model.network.JsonRpcRequest
-import ch.epfl.pop.model.network.method.message.Message
-import ch.epfl.pop.model.network.method.message.data.ObjectType
-import ch.epfl.pop.model.network.method.message.data.coin.PostTransaction
+import ch.epfl.pop.model.objects.DbActorNAckException
 import ch.epfl.pop.pubsub.graph.{ErrorCodes, GraphMessage, PipelineError}
 
 import scala.concurrent.ExecutionContext.Implicits.global
-import scala.concurrent.{Await, Future}
+import scala.concurrent.{Await}
+import scala.util.{Failure, Success}
 
 case object CoinHandler extends MessageHandler {
 
   def handlePostTransaction(rpcMessage: JsonRpcRequest): GraphMessage = {
-    rpcMessage.getParamsMessage match {
-      case Some(message: Message) =>
-        dbAskWritePropagate(rpcMessage)
-        Left(rpcMessage)
+    val ask = {
+      for {
+        _ <- extractParameters(rpcMessage, s"Unable to handle coin message $rpcMessage. Not a post message")
+        _ <- dbAskWritePropagate(rpcMessage)
+      } yield ()
+    }
 
-      case _ => Right(PipelineError(
+    Await.ready(ask, duration).value match {
+      case Some(Success(_))                        => Left(rpcMessage)
+      case Some(Failure(ex: DbActorNAckException)) => Right(PipelineError(ex.code, s"handlePostTransaction failed : ${ex.message}", rpcMessage.getId))
+      case reply => Right(PipelineError(
           ErrorCodes.SERVER_ERROR.id,
-          s"Unable to handle coin message $rpcMessage. Not a post message",
-          rpcMessage.id
+          s"handlePostTransaction failed : unexpected DbActor reply '$reply'",
+          rpcMessage.getId
         ))
     }
   }

be2-scala/src/main/scala/ch/epfl/pop/pubsub/graph/handlers/ElectionHandler.scala

@@ -13,7 +13,6 @@ import ch.epfl.pop.pubsub.graph.{ErrorCodes, GraphMessage, PipelineError}
 import ch.epfl.pop.storage.DbActor
 import ch.epfl.pop.storage.DbActor.DbActorReadElectionDataAck
 
-import java.nio.ByteBuffer
 import scala.collection.mutable
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.{Await, Future}
@@ -66,8 +65,10 @@ class ElectionHandler(dbRef: => AskableActorRef) extends MessageHandler {
           case OPEN_BALLOT => Left(rpcMessage)
           case SECRET_BALLOT =>
             val keyElection: KeyElection = KeyElection(electionId, keyPair.publicKey)
-            val broadcastKey: Base64Data = Base64Data.encode(KeyElectionFormat.write(keyElection).toString)
-            Await.result(dbBroadcast(rpcMessage, rpcMessage.getParamsChannel, broadcastKey, electionChannel), duration)
+            Await.result(
+              dbBroadcast(rpcMessage, rpcMessage.getParamsChannel, KeyElectionFormat.write(keyElection), electionChannel),
+              duration
+            )
         }
       case Some(Failure(ex: DbActorNAckException)) => Right(PipelineError(ex.code, s"handleSetupElection failed : ${ex.message}", rpcMessage.getId))
       case reply                                   => Right(PipelineError(ErrorCodes.SERVER_ERROR.id, s"handleSetupElection failed : unexpected DbActor reply '$reply'", rpcMessage.getId))
@@ -113,38 +114,51 @@ class ElectionHandler(dbRef: => AskableActorRef) extends MessageHandler {
       _ <- dbAskWritePropagate(rpcMessage)
       // data to be broadcast
       resultElection: ResultElection = ResultElection(electionQuestionResults, witnessSignatures)
-      data: Base64Data = Base64Data.encode(resultElectionFormat.write(resultElection).toString)
       // create & propagate the resultMessage
-      _ <- dbBroadcast(rpcMessage, electionChannel, data, electionChannel)
+      _ <- dbBroadcast(rpcMessage, electionChannel, resultElectionFormat.write(resultElection), electionChannel)
     } yield ()
     Await.ready(combined, duration).value match {
       case Some(Success(_)) => Left(rpcMessage)
       case _                => Right(PipelineError(ErrorCodes.SERVER_ERROR.id, s"handleEndElection unknown error", rpcMessage.getId))
     }
   }
 
+  /** Helper function to create the list of ElectionQuestionResult
+    *
+    * @param electionChannel
+    *   : the Channel in which we read the data
+    * @return
+    *   the list of ElectionQuestionResult wrapped in a [[scala.concurrent.Future]]
+    */
   private def createElectionQuestionResults(electionChannel: Channel): Future[List[ElectionQuestionResult]] = {
     for {
+      // get the last votes of the CastVotes messages
       castsVotesElections <- electionChannel.getLastVotes(dbActor)
+      // get the setupElection message of the channel
       setupMessage <- electionChannel.getSetupMessage(dbActor)
+      // associate the questions ids to their ballots
       questionToBallots = setupMessage.questions.map(question => question.id -> question.ballot_options).toMap
       DbActorReadElectionDataAck(electionData) <- dbActor ? DbActor.ReadElectionData(setupMessage.id)
     } yield {
+      // set up the table of results
       val resultsTable = mutable.HashMap.from(for {
         (question, ballots) <- questionToBallots
       } yield question -> ballots.map(_ -> 0).toMap)
       for {
         castVoteElection <- castsVotesElections
         voteElection <- castVoteElection.votes
+        // get the index of the vote
         voteIndex = electionChannel.getVoteIndex(electionData, voteElection.vote)
       } {
         val question = voteElection.question
         val ballots = questionToBallots(question).toArray
         val ballot = ballots.apply(voteIndex)
         val questionResult = resultsTable(question)
+        // update the results by adding a vote to the corresponding ballot
         resultsTable.update(question, questionResult.updated(ballot, questionResult(ballot) + 1))
       }
       (for {
+        // from the results saved in resultsTable, we construct the ElectionQuestionResult
         (qid, ballotToCount) <- resultsTable
         electionBallotVotes = List.from(for {
           (ballot, count) <- ballotToCount

be2-scala/src/main/scala/ch/epfl/pop/pubsub/graph/handlers/LaoHandler.scala

@@ -7,7 +7,7 @@ import ch.epfl.pop.model.network.JsonRpcRequest
 import ch.epfl.pop.model.network.method.message.Message
 import ch.epfl.pop.model.network.method.message.data.ObjectType
 import ch.epfl.pop.model.network.method.message.data.lao.{CreateLao, GreetLao, StateLao}
-import ch.epfl.pop.model.objects.{Base64Data, Channel, DbActorNAckException, Hash}
+import ch.epfl.pop.model.objects.{Channel, DbActorNAckException, Hash}
 import ch.epfl.pop.pubsub.graph.{ErrorCodes, GraphMessage, PipelineError}
 import ch.epfl.pop.storage.DbActor
 
@@ -46,8 +46,7 @@ case object LaoHandler extends MessageHandler {
         _ <- dbActor ? DbActor.WriteLaoData(laoChannel, message, address)
         // after creating the lao, we need to send a lao#greet message to the frontend
         greet: GreetLao = GreetLao(data.id, message.sender, address.get, List.empty)
-        broadcastGreet: Base64Data = Base64Data.encode(GreetLaoFormat.write(greet).toString())
-        _ <- dbBroadcast(rpcMessage, laoChannel, broadcastGreet, laoChannel)
+        _ <- dbBroadcast(rpcMessage, laoChannel, GreetLaoFormat.write(greet), laoChannel)
       } yield ()
 
     Await.ready(ask, duration).value.get match {

be2-scala/src/main/scala/ch/epfl/pop/pubsub/graph/handlers/MeetingHandler.scala

@@ -1,21 +1,30 @@
 package ch.epfl.pop.pubsub.graph.handlers
 
 import ch.epfl.pop.model.network.JsonRpcRequest
+import ch.epfl.pop.model.objects.DbActorNAckException
 import ch.epfl.pop.pubsub.graph.{ErrorCodes, GraphMessage, PipelineError}
 
-import scala.concurrent.{Await, Future}
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.{Await}
+import scala.util.{Failure, Success}
 
 case object MeetingHandler extends MessageHandler {
 
   def handleCreateMeeting(rpcMessage: JsonRpcRequest): GraphMessage = {
-    rpcMessage.getParamsChannel.decodeChannelLaoId match {
-      case Some(_) =>
-        val ask: Future[GraphMessage] = dbAskWritePropagate(rpcMessage)
-        Await.result(ask, duration)
-      case _ => Right(PipelineError(
-          ErrorCodes.INVALID_DATA.id,
-          s"Unable to create meeting: invalid encoded laoId '${rpcMessage.getParamsChannel}'",
-          rpcMessage.id
+    val ask = {
+      for {
+        _ <- extractParameters(rpcMessage, s"Unable to create meeting: invalid encoded laoId '${rpcMessage.getParamsChannel}'")
+        _ <- dbAskWritePropagate(rpcMessage)
+      } yield ()
+    }
+
+    Await.ready(ask, duration).value match {
+      case Some(Success(_))                        => Left(rpcMessage)
+      case Some(Failure(ex: DbActorNAckException)) => Right(PipelineError(ex.code, s"handleCreateMeeting failed : ${ex.message}", rpcMessage.getId))
+      case reply => Right(PipelineError(
+          ErrorCodes.SERVER_ERROR.id,
+          s"handleCreateMeeting failed : unexpected DbActor reply '$reply'",
+          rpcMessage.getId
         ))
     }
   }