Build encryption into election workflow
This commit is contained in:
parent
9505a03044
commit
b2bc6980cb
GPG Key ID:
7234E476BF21C61A
|
|
@ -21,6 +21,8 @@ Eos aims to be implementation-agnostic with respect to cryptographic details. Th
|
|||
* SCHNORR, Claus Peter and Markus JAKOBSSON. ‘Security of Signed ElGamal Encryption’. In: T. OKAMOTO, ed. *Advances in Cryptology – ASIACRYPT 2000*. Berlin: Springer-Verlag, 2000. pp. 73–89. Lecture Notes in Computer Science, vol. 1976. ISBN 978-3-540-44448-0. Available from: https://doi.org/10.1007/3-540-44448-3_7
|
||||
* **R**andomised partial checking (RPC) due to Jakobsson, Juels and Rivest (2002)
|
||||
* JAKOBSSON, Markus, Ari JUELS and Ronald L. RIVEST. ‘Making Mix Nets Robust For Electronic Voting By Randomized Partial Checking’. In: *Proceedings of the 11th USENIX Security Symposium*. pp. 339–353. Berkeley: USENIX Association, 2002. Available from: https://www.usenix.org/event/sec02/full_papers/jakobsson/jakobsson.pdf
|
||||
* Taking note of points raised by Khazaei and Wikström (2013)
|
||||
* KHAZAEI, Shahram and Douglas WIKSTRÖM. ‘Randomized Partial Checking Revisited’. In: E. DAWSON, ed. *Topics in Cryptology – CT-RSA 2013*. Berlin: Springer-Verlag, 2013. pp. 115–128. Lecture Notes in Computer Science, vol. 7779. ISBN 978-3-642-36095-4. Available from: https://doi.org/10.1007/978-3-642-36095-4_8
|
||||
|
||||
## Mother of all disclaimers
|
||||
|
||||
|
|
|
|||
|
|
@ -45,23 +45,12 @@ class Result(EmbeddedObject):
|
|||
|
||||
class ApprovalQuestion(Question):
|
||||
choices = ListField(StringField())
|
||||
|
||||
def compute_result(self):
|
||||
result = ApprovalResult(choices=([0] * len(self.choices)))
|
||||
for voter in self.recurse_parents(Election).voters:
|
||||
for ballot in voter.ballots:
|
||||
# TODO: Separate decryption phase
|
||||
encrypted_answer = ballot.encrypted_answers[self._instance[1]] # _instance[1] is the question number
|
||||
answer = encrypted_answer.decrypt()
|
||||
for choice in answer.choices:
|
||||
result.choices[choice] += 1
|
||||
return result
|
||||
|
||||
class ApprovalAnswer(Answer):
|
||||
choices = ListField(IntField())
|
||||
|
||||
class ApprovalResult(Result):
|
||||
choices = ListField(IntField())
|
||||
class RawResult(Result):
|
||||
answers = EmbeddedObjectListField()
|
||||
|
||||
class Election(TopLevelObject):
|
||||
_id = UUIDField()
|
||||
|
|
|
|||
|
|
@ -23,19 +23,16 @@ from eos.core.objects import *
|
|||
class ElectionTestCase(EosTestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
if is_python:
|
||||
client.drop_database('test')
|
||||
client.drop_database('test')
|
||||
|
||||
def exit_task_assert(self, election, task, next_task):
|
||||
def do_task_assert(self, election, task, next_task):
|
||||
self.assertEqual(election.workflow.get_task(task).status, WorkflowTask.Status.READY)
|
||||
self.assertEqual(election.workflow.get_task(next_task).status, WorkflowTask.Status.NOT_READY)
|
||||
election.workflow.get_task(task).exit()
|
||||
if next_task is not None:
|
||||
self.assertEqual(election.workflow.get_task(next_task).status, WorkflowTask.Status.NOT_READY)
|
||||
election.workflow.get_task(task).enter()
|
||||
self.assertEqual(election.workflow.get_task(task).status, WorkflowTask.Status.EXITED)
|
||||
self.assertEqual(election.workflow.get_task(next_task).status, WorkflowTask.Status.READY)
|
||||
|
||||
def save_if_python(self, obj):
|
||||
if is_python:
|
||||
obj.save()
|
||||
if next_task is not None:
|
||||
self.assertEqual(election.workflow.get_task(next_task).status, WorkflowTask.Status.READY)
|
||||
|
||||
@py_only
|
||||
def test_run_election(self):
|
||||
|
|
@ -65,18 +62,17 @@ class ElectionTestCase(EosTestCase):
|
|||
question = ApprovalQuestion(prompt='Chairman', choices=['John Doe', 'Andrew Citizen'])
|
||||
election.questions.append(question)
|
||||
|
||||
self.save_if_python(election)
|
||||
election.save()
|
||||
|
||||
# Check that it saved
|
||||
if is_python:
|
||||
self.assertEqual(db[Election._name].find_one()['value'], election.serialise())
|
||||
self.assertEqual(EosObject.deserialise_and_unwrap(db[Election._name].find_one()).serialise(), election.serialise())
|
||||
self.assertEqual(db[Election._db_name].find_one()['value'], election.serialise())
|
||||
self.assertEqual(EosObject.deserialise_and_unwrap(db[Election._db_name].find_one()).serialise(), election.serialise())
|
||||
|
||||
self.assertEqualJSON(EosObject.deserialise_and_unwrap(EosObject.serialise_and_wrap(election)).serialise(), election.serialise())
|
||||
|
||||
# Freeze election
|
||||
self.exit_task_assert(election, 'eos.base.workflow.TaskConfigureElection', 'eos.base.workflow.TaskOpenVoting')
|
||||
self.save_if_python(election)
|
||||
self.do_task_assert(election, 'eos.base.workflow.TaskConfigureElection', 'eos.base.workflow.TaskOpenVoting')
|
||||
election.save()
|
||||
|
||||
# Try to freeze it again
|
||||
try:
|
||||
|
|
@ -85,6 +81,10 @@ class ElectionTestCase(EosTestCase):
|
|||
except Exception:
|
||||
pass
|
||||
|
||||
# Open voting
|
||||
self.do_task_assert(election, 'eos.base.workflow.TaskOpenVoting', 'eos.base.workflow.TaskCloseVoting')
|
||||
election.save()
|
||||
|
||||
# Cast ballots
|
||||
VOTES = [[[0], [0]], [[0, 1], [1]], [[2], [0]]]
|
||||
|
||||
|
|
@ -96,19 +96,23 @@ class ElectionTestCase(EosTestCase):
|
|||
ballot.encrypted_answers.append(encrypted_answer)
|
||||
election.voters[i].ballots.append(ballot)
|
||||
|
||||
self.save_if_python(election)
|
||||
election.save()
|
||||
|
||||
# Close voting
|
||||
self.exit_task_assert(election, 'eos.base.workflow.TaskOpenVoting', 'eos.base.workflow.TaskCloseVoting')
|
||||
self.save_if_python(election)
|
||||
self.do_task_assert(election, 'eos.base.workflow.TaskCloseVoting', 'eos.base.workflow.TaskDecryptVotes')
|
||||
election.save()
|
||||
|
||||
# Compute result
|
||||
election.results = [None, None]
|
||||
for i in range(2):
|
||||
result = election.questions[i].compute_result()
|
||||
election.results[i] = result
|
||||
# "Decrypt" votes
|
||||
self.do_task_assert(election, 'eos.base.workflow.TaskDecryptVotes', 'eos.base.workflow.TaskReleaseResults')
|
||||
election.save()
|
||||
|
||||
self.save_if_python(election)
|
||||
# Check result
|
||||
RESULTS = [[[0], [0, 1], [2]], [[0], [1], [0]]]
|
||||
for i in range(len(RESULTS)):
|
||||
votes1 = RESULTS[i]
|
||||
votes2 = [x.choices for x in election.results[i].answers]
|
||||
self.assertEqual(sorted(votes1), sorted(votes2))
|
||||
|
||||
self.assertEqual(election.results[0].choices, [2, 1, 1])
|
||||
self.assertEqual(election.results[1].choices, [2, 1])
|
||||
# Release result
|
||||
self.do_task_assert(election, 'eos.base.workflow.TaskReleaseResults', None)
|
||||
election.save()
|
||||
|
|
|
|||
|
|
@ -20,7 +20,7 @@ class WorkflowTask(EmbeddedObject):
|
|||
class Status:
|
||||
NOT_READY = 10
|
||||
READY = 20
|
||||
#ENTERED = 30
|
||||
ENTERED = 30
|
||||
#COMPLETE = 40
|
||||
EXITED = 50
|
||||
|
||||
|
|
@ -40,6 +40,7 @@ class WorkflowTask(EmbeddedObject):
|
|||
self.status = WorkflowTask.Status.READY if self.are_dependencies_met() else WorkflowTask.Status.NOT_READY
|
||||
|
||||
self.listeners = {
|
||||
'enter': [],
|
||||
'exit': []
|
||||
}
|
||||
|
||||
|
|
@ -62,16 +63,31 @@ class WorkflowTask(EmbeddedObject):
|
|||
def satisfies(cls, descriptor):
|
||||
return cls._name == descriptor or descriptor in cls.provides
|
||||
|
||||
def on_enter(self):
|
||||
self.exit()
|
||||
|
||||
def enter(self):
|
||||
if self.status is not WorkflowTask.Status.READY:
|
||||
raise Exception('Attempted to enter a task when not ready')
|
||||
|
||||
self.status = WorkflowTask.Status.ENTERED
|
||||
self.fire_event('enter')
|
||||
self.on_enter()
|
||||
|
||||
def fire_event(self, event):
|
||||
for listener in self.listeners[event]:
|
||||
listener()
|
||||
|
||||
def on_exit(self):
|
||||
pass
|
||||
|
||||
def exit(self):
|
||||
if self.status is not WorkflowTask.Status.READY:
|
||||
raise Exception('Attempted to exit a task when not ready')
|
||||
if self.status is not WorkflowTask.Status.ENTERED:
|
||||
raise Exception('Attempted to exit a task when not entered')
|
||||
|
||||
self.status = WorkflowTask.Status.EXITED
|
||||
self.fire_event('exit')
|
||||
self.on_exit()
|
||||
|
||||
class Workflow(EmbeddedObject):
|
||||
tasks = EmbeddedObjectListField()
|
||||
|
|
@ -109,6 +125,26 @@ class TaskOpenVoting(WorkflowTask):
|
|||
class TaskCloseVoting(WorkflowTask):
|
||||
depends_on = ['eos.base.workflow.TaskOpenVoting']
|
||||
|
||||
class TaskDecryptVotes(WorkflowTask):
|
||||
depends_on = ['eos.base.workflow.TaskCloseVoting']
|
||||
|
||||
def on_enter(self):
|
||||
election = self.recurse_parents('eos.base.election.Election')
|
||||
|
||||
for _ in range(len(election.questions)):
|
||||
election.results.append(EosObject.objects['eos.base.election.RawResult']())
|
||||
|
||||
for voter in election.voters:
|
||||
for ballot in voter.ballots:
|
||||
for i in range(len(ballot.encrypted_answers)):
|
||||
answer = ballot.encrypted_answers[i].decrypt()
|
||||
election.results[i].answers.append(answer)
|
||||
|
||||
self.exit()
|
||||
|
||||
class TaskReleaseResults(WorkflowTask):
|
||||
depends_on = ['eos.base.workflow.TaskDecryptVotes']
|
||||
|
||||
# Concrete workflows
|
||||
# ==================
|
||||
|
||||
|
|
@ -119,3 +155,5 @@ class WorkflowBase(Workflow):
|
|||
self.tasks.append(TaskConfigureElection())
|
||||
self.tasks.append(TaskOpenVoting())
|
||||
self.tasks.append(TaskCloseVoting())
|
||||
self.tasks.append(TaskDecryptVotes())
|
||||
self.tasks.append(TaskReleaseResults())
|
||||
|
|
|
|||
|
|
@ -129,6 +129,9 @@ class EosObjectType(type):
|
|||
cls._name = ((cls.__module__ if is_python else meta.__next_class_module__) + '.' + cls.__name__).replace('.js.', '.').replace('.python.', '.') #TNYI: module and qualname
|
||||
if name != 'EosObject':
|
||||
EosObject.objects[cls._name] = cls
|
||||
if '_db_name' not in attrs:
|
||||
# Don't inherit _db_name, use only if explicitly given
|
||||
cls._db_name = cls._name
|
||||
return cls
|
||||
|
||||
class EosObject(metaclass=EosObjectType):
|
||||
|
|
@ -142,6 +145,9 @@ class EosObject(metaclass=EosObjectType):
|
|||
self._inited = True
|
||||
|
||||
def recurse_parents(self, cls):
|
||||
if not isinstance(cls, type):
|
||||
cls = EosObject.objects[cls]
|
||||
|
||||
if isinstance(self, cls):
|
||||
return self
|
||||
if self._instance[0]:
|
||||
|
|
@ -207,6 +213,9 @@ class EosList(EosObject):
|
|||
return self.impl[idx]
|
||||
def __setitem__(self, idx, val):
|
||||
self.impl[idx] = val
|
||||
val._instance = (self, idx)
|
||||
if not val._inited:
|
||||
val.post_init()
|
||||
def __contains__(self, val):
|
||||
return val in self.impl
|
||||
|
||||
|
|
@ -311,7 +320,7 @@ class DocumentObject(EosObject, metaclass=DocumentObjectType):
|
|||
class TopLevelObject(DocumentObject):
|
||||
def save(self):
|
||||
#res = db[self._name].replace_one({'_id': self.serialise()['_id']}, self.serialise(), upsert=True)
|
||||
res = db[self._name].replace_one({'_id': self._fields['_id'].serialise(self._id)}, EosObject.serialise_and_wrap(self), upsert=True)
|
||||
res = db[self._db_name].replace_one({'_id': self._fields['_id'].serialise(self._id)}, EosObject.serialise_and_wrap(self), upsert=True)
|
||||
|
||||
class EmbeddedObject(DocumentObject):
|
||||
pass
|
||||
|
|
|
|||
|
|
@ -139,73 +139,3 @@ class SEGCiphertext(EGCiphertext):
|
|||
_, c = EosObject.to_sha256(str(gs), str(self.gamma), str(self.delta))
|
||||
|
||||
return self.c == c
|
||||
|
||||
class BlockEncryptedAnswer(EncryptedAnswer):
|
||||
blocks = EmbeddedObjectListField()
|
||||
|
||||
def decrypt(self):
|
||||
# TODO
|
||||
raise Exception('NYI')
|
||||
|
||||
class RPCMixnet:
|
||||
def __init__(self):
|
||||
self.params = []
|
||||
|
||||
def random_permutation(self, n):
|
||||
permutation = list(range(n))
|
||||
# Fisher-Yates shuffle
|
||||
i = n
|
||||
while i != 0:
|
||||
rnd = BigInt.crypto_random(0, i - 1)
|
||||
rnd = rnd.__int__()
|
||||
i -= 1
|
||||
permutation[rnd], permutation[i] = permutation[i], permutation[rnd]
|
||||
return permutation
|
||||
|
||||
def shuffle(self, encrypted_answers):
|
||||
shuffled_answers = [None] * len(encrypted_answers)
|
||||
permutations = self.random_permutation(len(encrypted_answers))
|
||||
|
||||
permutations_and_reenc = []
|
||||
|
||||
for i in range(len(encrypted_answers)):
|
||||
encrypted_answer = encrypted_answers[i]
|
||||
|
||||
# Reencrypt the answer
|
||||
shuffled_blocks = []
|
||||
block_reencryptions = []
|
||||
for block in encrypted_answer.blocks:
|
||||
block2, reenc = block.reencrypt()
|
||||
shuffled_blocks.append(block2)
|
||||
block_reencryptions.append(reenc)
|
||||
# And shuffle it to the new position
|
||||
shuffled_answers[permutations[i]] = BlockEncryptedAnswer(blocks=shuffled_blocks)
|
||||
# Record the parameters
|
||||
permutations_and_reenc.append([permutations[i], block_reencryptions, block.public_key.group.random_element(), block.public_key.group.random_element()])
|
||||
|
||||
commitments_left = []
|
||||
for i in range(len(permutations_and_reenc)):
|
||||
val = permutations_and_reenc[i]
|
||||
val_json = [val[0], [str(x) for x in val[1]], str(val[2])]
|
||||
commitments_left.append(EosObject.to_sha256(EosObject.to_json(val_json))[0])
|
||||
|
||||
commitments_right = []
|
||||
for i in range(len(permutations_and_reenc)):
|
||||
# Find the answer that went to 'i'
|
||||
idx = next(idx for idx in range(len(permutations_and_reenc)) if permutations_and_reenc[idx][0] == i)
|
||||
val = permutations_and_reenc[idx]
|
||||
|
||||
val_json = [idx, [str(x) for x in val[1]], str(val[3])]
|
||||
commitments_right.append(EosObject.to_sha256(EosObject.to_json(val_json))[0])
|
||||
|
||||
self.params = permutations_and_reenc
|
||||
return shuffled_answers, commitments_left, commitments_right
|
||||
|
||||
def challenge(self, i, is_left):
|
||||
if is_left:
|
||||
val = self.params[i]
|
||||
return [val[0], val[1], val[2]]
|
||||
else:
|
||||
idx = next(idx for idx in range(len(self.params)) if self.params[idx][0] == i)
|
||||
val = self.params[idx]
|
||||
return [idx, val[1], val[3]]
|
||||
|
|
|
|||
|
|
@ -0,0 +1,57 @@
|
|||
# Eos - Verifiable elections
|
||||
# Copyright © 2017 RunasSudo (Yingtong Li)
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Affero General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
from eos.core.objects import *
|
||||
from eos.base.election import *
|
||||
from eos.psr.bitstream import *
|
||||
from eos.psr.crypto import *
|
||||
|
||||
class BlockEncryptedAnswer(EncryptedAnswer):
|
||||
blocks = EmbeddedObjectListField()
|
||||
|
||||
@classmethod
|
||||
def encrypt(cls, pk, obj):
|
||||
pt = EosObject.to_json(EosObject.serialise_and_wrap(obj))
|
||||
bs = BitStream()
|
||||
bs.write_string(pt)
|
||||
bs.multiple_of(pk.group.p.nbits() - 1, True)
|
||||
ct = bs.map(pk.encrypt, pk.group.p.nbits() - 1)
|
||||
|
||||
return cls(blocks=ct)
|
||||
|
||||
def decrypt(self, sk=None):
|
||||
if sk is None:
|
||||
sk = self.recurse_parents(PSRElection).sk
|
||||
|
||||
bs = BitStream.unmap(self.blocks, sk.decrypt, sk.public_key.group.p.nbits() - 1)
|
||||
m = bs.read_string()
|
||||
obj = EosObject.deserialise_and_unwrap(EosObject.from_json(m))
|
||||
|
||||
return obj
|
||||
|
||||
class Trustee(EmbeddedObject):
|
||||
name = StringField()
|
||||
email = StringField()
|
||||
|
||||
class MixingTrustee(Trustee):
|
||||
mix_order = IntField()
|
||||
mixed_questions = EmbeddedObjectListField()
|
||||
|
||||
class PSRElection(Election):
|
||||
_db_name = Election._name
|
||||
|
||||
sk = EmbeddedObjectField(SEGPrivateKey) # TODO: Threshold
|
||||
mixing_trustees = EmbeddedObjectListField(MixingTrustee)
|
||||
|
|
@ -0,0 +1,82 @@
|
|||
# Eos - Verifiable elections
|
||||
# Copyright © 2017 RunasSudo (Yingtong Li)
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Affero General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
from eos.core.bigint import *
|
||||
from eos.core.objects import *
|
||||
from eos.psr.election import *
|
||||
|
||||
class RPCMixnet:
|
||||
def __init__(self):
|
||||
self.params = []
|
||||
|
||||
def random_permutation(self, n):
|
||||
permutation = list(range(n))
|
||||
# Fisher-Yates shuffle
|
||||
i = n
|
||||
while i != 0:
|
||||
rnd = BigInt.crypto_random(0, i - 1)
|
||||
rnd = rnd.__int__()
|
||||
i -= 1
|
||||
permutation[rnd], permutation[i] = permutation[i], permutation[rnd]
|
||||
return permutation
|
||||
|
||||
def shuffle(self, encrypted_answers):
|
||||
shuffled_answers = [None] * len(encrypted_answers)
|
||||
permutations = self.random_permutation(len(encrypted_answers))
|
||||
|
||||
permutations_and_reenc = []
|
||||
|
||||
for i in range(len(encrypted_answers)):
|
||||
encrypted_answer = encrypted_answers[i]
|
||||
|
||||
# Reencrypt the answer
|
||||
shuffled_blocks = []
|
||||
block_reencryptions = []
|
||||
for block in encrypted_answer.blocks:
|
||||
block2, reenc = block.reencrypt()
|
||||
shuffled_blocks.append(block2)
|
||||
block_reencryptions.append(reenc)
|
||||
# And shuffle it to the new position
|
||||
shuffled_answers[permutations[i]] = BlockEncryptedAnswer(blocks=shuffled_blocks)
|
||||
# Record the parameters
|
||||
permutations_and_reenc.append([permutations[i], block_reencryptions, block.public_key.group.random_element(), block.public_key.group.random_element()])
|
||||
|
||||
commitments_left = []
|
||||
for i in range(len(permutations_and_reenc)):
|
||||
val = permutations_and_reenc[i]
|
||||
val_json = [val[0], [str(x) for x in val[1]], str(val[2])]
|
||||
commitments_left.append(EosObject.to_sha256(EosObject.to_json(val_json))[0])
|
||||
|
||||
commitments_right = []
|
||||
for i in range(len(permutations_and_reenc)):
|
||||
# Find the answer that went to 'i'
|
||||
idx = next(idx for idx in range(len(permutations_and_reenc)) if permutations_and_reenc[idx][0] == i)
|
||||
val = permutations_and_reenc[idx]
|
||||
|
||||
val_json = [idx, [str(x) for x in val[1]], str(val[3])]
|
||||
commitments_right.append(EosObject.to_sha256(EosObject.to_json(val_json))[0])
|
||||
|
||||
self.params = permutations_and_reenc
|
||||
return shuffled_answers, commitments_left, commitments_right
|
||||
|
||||
def challenge(self, i, is_left):
|
||||
if is_left:
|
||||
val = self.params[i]
|
||||
return [val[0], val[1], val[2]]
|
||||
else:
|
||||
idx = next(idx for idx in range(len(self.params)) if self.params[idx][0] == i)
|
||||
val = self.params[idx]
|
||||
return [idx, val[1], val[3]]
|
||||
|
|
@ -19,6 +19,9 @@ from eos.core.tests import *
|
|||
from eos.core.bigint import *
|
||||
from eos.psr.bitstream import *
|
||||
from eos.psr.crypto import *
|
||||
from eos.psr.election import *
|
||||
from eos.psr.mixnet import *
|
||||
from eos.psr.workflow import *
|
||||
|
||||
class EGTestCase(EosTestCase):
|
||||
def test_eg(self):
|
||||
|
|
@ -102,15 +105,11 @@ class BlockEGTestCase(EosTestCase):
|
|||
|
||||
def test_object(self):
|
||||
obj = self.Person(name='John Smith')
|
||||
pt = EosObject.to_json(EosObject.serialise_and_wrap(obj))
|
||||
bs = BitStream()
|
||||
bs.write_string(pt)
|
||||
bs.multiple_of(self.test_group.p.nbits() - 1, True)
|
||||
ct = bs.map(self.sk.public_key.encrypt, self.test_group.p.nbits() - 1)
|
||||
bs2 = BitStream.unmap(ct, self.sk.decrypt, self.test_group.p.nbits() - 1)
|
||||
m = bs2.read_string()
|
||||
obj2 = EosObject.deserialise_and_unwrap(EosObject.from_json(m))
|
||||
self.assertEqualJSON(obj, obj2)
|
||||
|
||||
ct = BlockEncryptedAnswer.encrypt(self.sk.public_key, obj)
|
||||
m = ct.decrypt(self.sk)
|
||||
|
||||
self.assertEqualJSON(obj, m)
|
||||
|
||||
class MixnetTestCase(EosTestCase):
|
||||
@py_only
|
||||
|
|
@ -167,3 +166,79 @@ class MixnetTestCase(EosTestCase):
|
|||
val_json = [perm, [str(x) for x in reencs], str(rand)]
|
||||
self.assertEqual(commitments_right[i], EosObject.to_sha256(EosObject.to_json(val_json))[0])
|
||||
verify_shuffle(perm, i, reencs)
|
||||
|
||||
class ElectionTestCase(EosTestCase):
|
||||
def do_task_assert(self, election, task, next_task):
|
||||
self.assertEqual(election.workflow.get_task(task).status, WorkflowTask.Status.READY)
|
||||
if next_task is not None:
|
||||
self.assertEqual(election.workflow.get_task(next_task).status, WorkflowTask.Status.NOT_READY)
|
||||
election.workflow.get_task(task).enter()
|
||||
self.assertEqual(election.workflow.get_task(task).status, WorkflowTask.Status.EXITED)
|
||||
if next_task is not None:
|
||||
self.assertEqual(election.workflow.get_task(next_task).status, WorkflowTask.Status.READY)
|
||||
|
||||
@py_only
|
||||
def test_run_election(self):
|
||||
# Set up election
|
||||
election = PSRElection()
|
||||
election.workflow = PSRWorkflow()
|
||||
|
||||
# Set election details
|
||||
election.name = 'Test Election'
|
||||
|
||||
for i in range(3):
|
||||
voter = Voter()
|
||||
election.voters.append(voter)
|
||||
|
||||
for i in range(3):
|
||||
mixing_trustee = MixingTrustee(mix_order=i)
|
||||
election.mixing_trustees.append(mixing_trustee)
|
||||
|
||||
election.sk = EGPrivateKey.generate()
|
||||
|
||||
question = ApprovalQuestion(prompt='President', choices=['John Smith', 'Joe Bloggs', 'John Q. Public'])
|
||||
election.questions.append(question)
|
||||
|
||||
question = ApprovalQuestion(prompt='Chairman', choices=['John Doe', 'Andrew Citizen'])
|
||||
election.questions.append(question)
|
||||
|
||||
election.save()
|
||||
|
||||
# Freeze election
|
||||
self.do_task_assert(election, 'eos.base.workflow.TaskConfigureElection', 'eos.base.workflow.TaskOpenVoting')
|
||||
|
||||
# Open voting
|
||||
self.do_task_assert(election, 'eos.base.workflow.TaskOpenVoting', 'eos.base.workflow.TaskCloseVoting')
|
||||
election.save()
|
||||
|
||||
# Cast ballots
|
||||
VOTES = [[[0], [0]], [[0, 1], [1]], [[2], [0]]]
|
||||
|
||||
for i in range(3):
|
||||
ballot = Ballot()
|
||||
for j in range(2):
|
||||
answer = ApprovalAnswer(choices=VOTES[i][j])
|
||||
encrypted_answer = BlockEncryptedAnswer.encrypt(election.sk.public_key, answer)
|
||||
ballot.encrypted_answers.append(encrypted_answer)
|
||||
election.voters[i].ballots.append(ballot)
|
||||
|
||||
election.save()
|
||||
|
||||
# Close voting
|
||||
self.do_task_assert(election, 'eos.base.workflow.TaskCloseVoting', 'eos.base.workflow.TaskDecryptVotes')
|
||||
election.save()
|
||||
|
||||
# Decrypt votes, for realsies
|
||||
self.do_task_assert(election, 'eos.base.workflow.TaskDecryptVotes', 'eos.base.workflow.TaskReleaseResults')
|
||||
election.save()
|
||||
|
||||
# Check result
|
||||
RESULTS = [[[0], [0, 1], [2]], [[0], [1], [0]]]
|
||||
for i in range(len(RESULTS)):
|
||||
votes1 = RESULTS[i]
|
||||
votes2 = [x.choices for x in election.results[i].answers]
|
||||
self.assertEqual(sorted(votes1), sorted(votes2))
|
||||
|
||||
# Release result
|
||||
self.do_task_assert(election, 'eos.base.workflow.TaskReleaseResults', None)
|
||||
election.save()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,34 @@
|
|||
# Eos - Verifiable elections
|
||||
# Copyright © 2017 RunasSudo (Yingtong Li)
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Affero General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
from eos.core.objects import *
|
||||
from eos.base.workflow import *
|
||||
|
||||
# Concrete tasks
|
||||
# ==============
|
||||
|
||||
# Concrete workflows
|
||||
# ==================
|
||||
|
||||
class PSRWorkflow(Workflow):
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
self.tasks.append(TaskConfigureElection())
|
||||
self.tasks.append(TaskOpenVoting())
|
||||
self.tasks.append(TaskCloseVoting())
|
||||
self.tasks.append(TaskDecryptVotes())
|
||||
self.tasks.append(TaskReleaseResults())
|
||||
|
|
@ -79,14 +79,14 @@ for dirpath, dirnames, filenames in os.walk('eos'):
|
|||
method_val = getattr(impl, method)
|
||||
if isinstance(method_val, types.MethodType) and not hasattr(cls_py, method):
|
||||
# Python
|
||||
if not getattr(method_val, '_js_only', False):
|
||||
if not (len(sys.argv) > 2 and sys.argv[2] == 'js') and not getattr(method_val, '_js_only', False):
|
||||
cls_py.add_method(method)
|
||||
if method.startswith('test_'):
|
||||
test_case = cls_py(method)
|
||||
test_suite.addTest(test_case)
|
||||
|
||||
# Javascript
|
||||
if not getattr(method_val, '_py_only', False):
|
||||
if not (len(sys.argv) > 2 and sys.argv[2] == 'py') and not getattr(method_val, '_py_only', False):
|
||||
if method.startswith('test_'):
|
||||
cls_js.add_method(method)
|
||||
test_case = cls_js(method)
|
||||
|
|
|
|||
Loading…
Reference in New Issue