Elections

.circleci/config.yml

@@ -0,0 +1,21 @@
+version: 2.1
+
+orbs:
+  python: circleci/[email protected]
+
+jobs:
+  build-and-test:
+    executor: python/default
+    steps:
+      - checkout
+      - python/load-cache
+      - python/install-deps
+      - python/save-cache
+      - run:
+          command: python -m pytest
+          name: Test
+
+workflows:
+  main:
+    jobs:
+      - build-and-test

BUILD.txt

@@ -5,3 +5,5 @@
 01/27/20-17:17:06-joris
 01/27/20-17:18:06-joris
 01/27/20-17:22:43-joris
+02/04/20-16:52:22-
+02/04/20-16:53:02-

Database.md

@@ -21,6 +21,7 @@ A sample Election Document looks like this:
 , "closed" : <Bool: is Closed?>
 , "winner" : <Winner Proposal>
 , "type" : "<either machine or human>"
+, "votes" : [{ "username" : "<username of voter>", "vote" : b"<encrypted vote>" }]
 }
 ```
 There can be two kinds of proposals:

Server/.goutputstream-9HC4F0

@@ -0,0 +1,47 @@
+import functools,random
+from typing import List, Dict
+
+def vote(votes : List[List[str]], options : List[str]):
+    options = {key : list(filter(lambda v: v[-1] == key or (key == "NoneOfTheOtherOptions" and v == []), votes)) for key in options}
+    winner = False
+    while(winner == False):
+        result = _vote(options,len(votes))
+        if type(result) == type(""):
+            winner = (result,len(options[result]))
+        elif type(result) == type(None):
+            winner = (None, None)
+        elif type(result) == type({}):
+            options = result
+        else:
+            raise "VotingError: Unknow type retuned by _vote"
+    return winner
+
+def _vote(options : Dict[str,List[List[str]]], votes : int):
+    ordered = sorted(list(options), key=lambda o: options[o], reverse=True)
+
+    if len(options[ordered[0]])/votes > 0.5:
+        return ordered[0]
+    elif len(options[ordered[0]]) == len(options[ordered[1]]) and len(options[ordered[0]]) == 0.5:
+        return None
+    else:
+        least = options[ordered[-1]]
+        print(least)
+        least = options.pop(ordered[-1])
+        options = { key : [a for a in least if a != ordered[-1]] for key in list(options)}
+
+
+        for vote in least:
+            if len(vote) > 1:
+                vote.pop()
+                options[vote[-1]].append(vote)
+                print(vote[-1])
+            elif len(vote) == 1:
+                options["NoneOfTheOtherOptions"].append([])
+            else:
+                raise "VotingError: NoneOfTheOtherOptions seems to be removed from options, because empty lists are supposed to be reallocated, which only appear in NoneOfTheOtherOptions"
+        return options
+
+def test(n):
+    options = [str(x) for x in range(100)]
+    votes = [random.sample(options,k=random.randint(1,len(options))) for i in range(n)]
+    return vote(votes,options)

Server/ELECTION.el

@@ -0,0 +1,8 @@
+# File to represent an election result
+
+Votes:
+# List of votes as python lists
+Thrown:
+# All the options, that were thrown out after each other
+Winner:
+# The Option, that won + the number of votes in support of it.

Server/Elections.py

@@ -5,7 +5,9 @@
 from Server.election import count_votes
 from pymongo import MongoClient
 from Crypto.Hash import SHA256
-
+from Crypto.PublicKey import RSA
+from Crypto.Cipher import PKCS1_OAEP
+from typing import List, Tuple, Mapping
 
 """
 Utility function to get the MongoClient.demnet[<collection>]
@@ -93,21 +95,32 @@ def create(type,deadline,proposals):
         elections.insert_one(election)
         return election['hash']
 
+
+"""Stores a vote byte string, that is encrypted like this:
+Given Keys:
+- Private Key of the User (private_key_of_user)
+- Public Key of the Election Authority (public_key_of_ea)
+Encryption of the vote as string:
+vote_e = E(private_key_of_user, E(public_key_of_ea, vote))
+If you only have the public key of the user you can only
+tell, that a user has voted but not how.
+And only the EA can read the vote, with their private key.
+If you don't trust the EA or the EA has been compromised,
+the election is invalid to you, but this could easily
+be decentralised.
 """
-a vote is a list of all options ranked by
-how much a voter wants them to win. (see alternative vote).
-**This function call cannot leave any trace of the association between
-username and vote.**
-"""
-def vote(election_hash,vote,username):
+def vote(election_hash : str, username : str, vote : List[str], private_key_of_user : RSA.RsaKey, public_key_of_ea : RSA.RsaKey) -> bool:
     elections = collection("elections")
     election = elections.find_one({ "hash" : election_hash })
 
-    if username in election["participants"] and not election['closed']:
+    cipher_ea = PKCS1_OAEP.new(public_key_of_ea)
+    cipher_user = PKCS1_OAEP.new(private_key_of_user)
+
+    if username in election["participants"]:
         return False
     else:
-        elections.update_one({ "hash" : election_hash }, { "$push" : { "participants" : username }})
-        elections.update_one({ "hash" : election_hash }, { "$push" : { "votes" : vote }})
+        vote = cipher_user(cipher_ea.encrypt(json.dumps(vote).encode('utf-8')))
+        elections.update_one({ "hash" : election_hash }, { "$push" : { "votes" : { "username" : username, "vote" : vote_e }}}
         return True
 
 
@@ -118,21 +131,22 @@ def vote(election_hash,vote,username):
 Publishing the votes, to make the independent control
 possible.
 """
-def close(election_hash):
+def close(election_hash : str, private_key_ea : RSA.RsaKey):
     elections = collection('elections')
     election = elections.find_one({ "hash" : election_hash })
 
     if election:
         if election.get('deadline') <= time.time():
-            winner = count_votes(election.votes, len(election.participants), range(0,len(election.proposals)+1))["ballot"]
+            votes = encrypt_votes(election["votes"], private_key_ea)
+            winner = count_votes(election["votes"], len(election.participants), range(0,len(election.proposals)+1))["ballot"]
             winner = election.proposals[winner]
             elections.update_one({ "hash" : election_hash }, { "$set" : { "winner" : winner, "closed" : True } })
 
             if election['type'] == "machine":
                 patches = collection('patches')
                 patch = patches.find_one({ "hash" : winner['patch_id'] })
                 Patches.close(patch['name'], patch['name'], patch['hash'], merge=True)
-            elif election['type'] == "human":
+            else:
                 laws = collection("laws")
                 # Append ammendments to laws
                 for ammendment in winner["ammendment"]:
@@ -150,6 +164,18 @@ def close(election_hash):
                     laws.remove_one({ "title" : removal })
 
 
-            return True
+        return True
     else:
         return False
+
+def encrypt_votes(votes : List[bytes], private_key_ea : RSA.RsaKey) -> List[List[str]]:
+    users   = collection("users")
+    cipher_ea = PKCS1_OAEP.new(private_key_ea)
+    for vote in votes:
+        user    = users.find_one({ "username" : vote["username"] })
+        public_key_of_user = RSA.import_key(user['public_key'])
+        cipher_user = PKCS1_OAEP.new(public_key_of_user)
+        try:
+            vote = cipher_ea.decrypt(cipher_user.decrypt(user["vote"]))
+        except Exception as e:
+            raise

Server/Users.py

@@ -10,7 +10,7 @@
 """
 from Crypto.PublicKey import RSA
 from Crypto.Hash import SHA256
-from Crypto.Cipher import AES
+from Crypto.Cipher import PKCS1_OAEP
 from pymongo import MongoClient
 import datetime, sys, json
 

Server/election.el

@@ -0,0 +1,5 @@
+Votes:
+ [["A", "B", "C"], ["C", "A", "B"], ["A", "B", "C"]]
+Thrown:
+Winner:
+('C', 1.0)

Server/election.py

@@ -1,47 +1,49 @@
-import functools
-
-def count_votes( votes, participant_count, options ):
-    ballot = map( lambda option : { 'option' : option, 'support' : [] }, options )
-    winner = distribute_votes( votes, participant_count, list(ballot) )
-    return { "ballot": winner, "participants" : participant_count, "options" : options }
-
-def distribute_votes( votes, participant_count, ballot ):
-    if len(ballot[0]['support']) > participant_count * 0.5 or len(ballot) <= 1:
-        return ballot[0] # Winner!
-    else:
-        # Distribute votes
-        for option in ballot:
-            option['support'] = option['support'] + list(filter(lambda v: v[-1] == option['option'], votes))
-
-        # Eliminate another low ranking option
-        # Sort by Support
-        ballot = sorted(ballot, key = lambda option : len(option['support']), reverse=True)
-
-        looser = least_popular(ballot)
-        ballot.remove(looser)
-        looser['support'] = list(map(lambda v: v[:-1], looser['support']))
-        return distribute_votes(looser['support'], participant_count, ballot)
-
-def least_popular( ballot ):
-    # If there is a tie, look at the option, that is more popular in the alternative votes.
-    if len(ballot[-1]['support']) == len(ballot[-2]['support']):
-        option1 = ballot[-1]
-        option2 = ballot[-2]
-        alternative_votes = list( map( lambda option: option['support'][:-1], ballot ) )
-        alternative_votes = functools.reduce( lambda x, y : x + y, alternative_votes )
-
-        i = -1
-        least_popular = False
-
-        while(not least_popular):
-            is_tie = alternative_votes[i].count(option1['option']) - alternative_votes.count(option2['option'])
-
-            if is_tie < 0:
-                least_popular = option1
-            elif is_tie > 0:
-                least_popular = option2
+import random, pprint
+from typing import List, Dict
+
+def count(votes : List[List[str]], options : List[str]):
+    options  : Dict[str, List[List[str]]] = { key : list(filter(lambda v: v[-1] == key, votes)) for key in list(options) }
+    votes = len(votes)
+    all_participants = votes
+    thrown_out = 0
+    result = { "winner" : False
+             , "rounds" : []
+             }
+    while result["winner"] == False:
+        # Does a candidate have more than 50%?
+        result["rounds"].append(options)
+        winners = list(filter(lambda o: len(options[o]) >= 0.5*votes, list(options)))
+        if len(winners) == 1:
+            result["winner"] = winners[0]
+            break
+        elif len(winners) >= 2:
+            result["winner"] = None
+            break
+        elif len(winners) == 0:
+            # Is there only one candidate left?
+            if len(list(options)) == 1:
+                result["winner"] = options(list(options)[0])
+                break
+            elif len(list(options)) == 0:
+                result["winner"] = None
+                break
             else:
-                least_popular = False
-        return least_popular
-    else:
-        return ballot[-1]
+                # Drop worst candidate and find out who voters liked next best
+                sorted_options = sorted(list(options), key=lambda o: len(options[o]))
+                worst = sorted_options[-1]
+                worsts_votes = options.pop(worst)
+                options = { key : [list(filter(lambda a: a != worst, vote)) for vote in options[key]] for key in list(options)}
+                options = { key : options[key] for key in list(options) if options[key] != []}
+                sum_votes = sum([len(options[o]) for o in list(options)])
+                thrown_out += (votes - sum_votes)
+                votes = sum_votes
+                continue
+
+    result["thrown_out"] = thrown_out
+    if result["thrown_out"]/all_participants > 0.5:
+        result["winner"] = "NoneOfTheOtherOptions"
+    elif len(result["winner"])/all_participants == 0.5 and (result["thrown_out"]/all_participants == 0.5):
+        result["winner"] = None
+
+    result["winner"] = "NoneOfTheOtherOptions" if (result["thrown_out"]/all_participants) > 0.5 else result["winner"]
+    return result

Server/sample_election.el

@@ -0,0 +1,7 @@
+Votes:
+ [['A', 'B', 'C'], ['C', 'B', 'A'], ['D', 'A', 'B', 'C'], ['D', 'A', 'C', 'B'], ['NoneOfTheOtherOptions']]
+Thrown:
+A
+D
+Winner:
+B, 3

Server/test_election.py

@@ -0,0 +1,20 @@
+from election import count
+import random,os
+
+def test_count(n=50, repeat_for=10**5,seed=None):
+    for i in range(repeat_for):
+        options = [str(i) for i in range(random.randint(4,20))]
+        votes = [random.sample(options,k=random.randint(1,len(options))) for i in range(n)]
+        result = count(votes,options)
+
+        # On AV there is one option deleted every round
+        # thus after len(options) rounds, AV has to stop.
+        assert len(result["rounds"]) <= len(options)
+        assert len(result["rounds"]) > 0
+
+        assert result["thrown_out"] >= 0
+
+        if result["winner"] != "NoneOfTheOtherOptions":
+            assert result["thrown_out"]/len(votes) < 0.5
+        else:
+            assert result["thrown_out"] > 0.5*len(votes)