Eos/eos/psr/election.py

279 lines
9.8 KiB
Python
Raw Normal View History

# Eos - Verifiable elections
# Copyright © 2017-2021 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/>.
2017-09-28 19:37:10 +10:00
from eos.core.bigint import *
from eos.core.objects import *
2017-09-29 13:06:37 +10:00
from eos.core.hashing import *
from eos.base.election import *
from eos.psr.bitstream import *
from eos.psr.crypto import *
2017-11-24 19:37:48 +11:00
from eos.core.objects import __pragma__
class BlockEncryptedAnswer(EncryptedAnswer):
blocks = EmbeddedObjectListField()
@classmethod
def encrypt(cls, pk, obj, nbits=None):
pt = EosObject.to_json(EosObject.serialise_and_wrap(obj))
bs = BitStream()
bs.write_string(pt)
if nbits is not None:
if bs.nbits > nbits:
raise Exception('Message is too big')
bs.pad_to(nbits, True)
bs.multiple_of(pk.nbits(), True)
ct = bs.map(pk.encrypt, pk.nbits())
return cls(blocks=ct)
def decrypt(self, sk=None):
if sk is None:
sk = self.recurse_parents(PSRElection).sk
plaintexts = EosList([sk.decrypt_and_prove(block) for block in self.blocks])
bs = BitStream.unmap(plaintexts, lambda plaintext: plaintext.message, sk.public_key.nbits())
m = bs.read_string()
obj = EosObject.deserialise_and_unwrap(EosObject.from_json(m))
return plaintexts, obj
2017-11-28 22:43:32 +11:00
def deaudit(self):
blocks_deaudit = EosList()
for block in self.blocks:
blocks_deaudit.append(block.deaudit())
2017-11-28 22:43:32 +11:00
return BlockEncryptedAnswer(blocks=blocks_deaudit)
class Trustee(EmbeddedObject):
name = StringField()
email = StringField()
2017-09-28 19:37:10 +10:00
class MixChallengeResponse(EmbeddedObject):
2017-09-29 13:06:37 +10:00
challenge_index = IntField()
response_index = IntField()
2017-09-28 19:37:10 +10:00
reenc = EmbeddedObjectListField(BigInt)
rand = EmbeddedObjectField(BigInt)
class MixingTrustee(Trustee):
2017-11-24 19:37:48 +11:00
mixed_questions = ListField(EmbeddedObjectListField(BlockEncryptedAnswer), is_hashed=False)
commitments = ListField(EmbeddedObjectListField(BigInt), is_hashed=False)
challenge = EmbeddedObjectListField(BigInt, is_hashed=False)
response = ListField(EmbeddedObjectListField(MixChallengeResponse), is_hashed=False)
2017-09-29 13:06:37 +10:00
def compute_challenge(self, question_num):
2017-09-29 16:01:15 +10:00
if self._instance[1] % 2 == 1:
return self.recurse_parents(Election).mixing_trustees[self._instance[1] - 1].compute_challenge(question_num)
2017-09-29 13:06:37 +10:00
sha = SHA256()
trustees = self.recurse_parents(Election).mixing_trustees
for trustee in trustees:
sha.update_text(EosObject.to_json(MixingTrustee._fields['mixed_questions'].element_field.serialise(trustee.mixed_questions[question_num])))
2017-09-29 13:06:37 +10:00
for i in range(self._instance[1]):
sha.update_text(EosObject.to_json(MixingTrustee._fields['response'].element_field.serialise(trustees[i].response[question_num])))
2017-09-29 16:01:15 +10:00
return sha.hash_as_bigint()
def get_input_answers(self, question_num):
if self._instance[1] > 0:
# Use the previous mixnet's output
return self.recurse_parents(Election).mixing_trustees[self._instance[1] - 1].mixed_questions[question_num]
else:
# Use the raw ballots from voters
orig_answers = []
for voter in self.recurse_parents(Election).voters:
if len(voter.votes.get_all()) > 0:
vote = voter.votes.get_all()[-1]
ballot = vote.ballot
2017-09-29 16:01:15 +10:00
orig_answers.append(ballot.encrypted_answers[question_num])
return orig_answers
def verify(self, question_num):
# Verify challenge
challenge = self.compute_challenge(question_num)
if challenge != self.challenge[question_num]:
raise Exception('Invalid challenge')
orig_answers = self.get_input_answers(question_num)
# Prepare challenge bits
challenge_bs = InfiniteHashBitStream(challenge)
# Check each challenge response
for k in range(len(self.mixed_questions[question_num])):
2017-11-24 19:37:48 +11:00
response = self.response[question_num][k]
2017-09-29 16:01:15 +10:00
challenge_bit = challenge_bs.read(1)
should_reveal = ((self._instance[1] % 2) == (challenge_bit % 2))
if should_reveal:
2017-11-24 19:37:48 +11:00
if response is None:
raise Exception('Response inconsistent with challenge')
2017-09-29 16:01:15 +10:00
# Check the commitment matches
if self.commitments[question_num][k] != SHA256().update_obj(response).hash_as_bigint():
raise Exception('Invalid commitment')
# Check the correct challenge/response pair
if response.challenge_index != k:
raise Exception('Invalid response')
if self._instance[1] % 2 == 0:
idx_left = response.challenge_index
idx_right = response.response_index
else:
idx_right = response.challenge_index
idx_left = response.response_index
# Check the shuffle
claimed_blocks = self.mixed_questions[question_num][idx_right].blocks
for k in range(len(orig_answers[idx_left].blocks)):
reencrypted_block, _ = orig_answers[idx_left].blocks[k].reencrypt(response.reenc[k])
if claimed_blocks[k].gamma != reencrypted_block.gamma:
raise Exception('Reencryption not consistent with challenge response')
if claimed_blocks[k].delta != reencrypted_block.delta:
raise Exception('Reencryption not consistent with challenge response')
2017-11-24 19:37:48 +11:00
else:
if response is not None:
raise Exception('Response inconsistent with challenge')
2017-09-29 16:01:15 +10:00
# Check the responses are consistent with a permutation
challenge_indexes = []
response_indexes = []
for response in self.response[question_num]:
2017-11-24 19:37:48 +11:00
if response is None:
continue
2017-09-29 16:01:15 +10:00
if response.challenge_index in challenge_indexes:
raise Exception('Response not consistent with a permutation')
if response.response_index in response_indexes:
raise Exception('Response not consistent with a permutation')
challenge_indexes.append(response.challenge_index)
response_indexes.append(response.response_index)
# Check the outputs are all different
blocks = []
for output in self.mixed_questions[question_num]:
for block in output.blocks:
block = (str(block.gamma), str(block.delta))
if block in blocks:
raise Exception('Duplicate ciphertexts in output')
blocks.append(block)
2017-11-24 19:37:48 +11:00
def mix_votes(self, question=0):
return False
def prove_mixes(self, question=0):
return False
class InternalMixingTrustee(MixingTrustee):
2017-12-07 16:04:24 +11:00
mixnets = EmbeddedObjectListField(is_protected=True)
2017-11-24 19:37:48 +11:00
def mix_votes(self, question=0):
__pragma__('skip')
from eos.psr.mixnet import RPCMixnet
__pragma__('noskip')
election = self.recurse_parents('eos.base.election.Election')
index = self._instance[1]
2017-12-07 16:04:24 +11:00
self.mixnets.append(RPCMixnet(mix_order=index))
2017-11-24 19:37:48 +11:00
if index > 0:
orig_answers = election.mixing_trustees[index - 1].mixed_questions[question]
else:
orig_answers = []
for voter in election.voters:
if len(voter.votes.get_all()) > 0:
ballot = voter.votes.get_all()[-1].ballot
2017-11-24 19:37:48 +11:00
orig_answers.append(ballot.encrypted_answers[question])
shuffled_answers, commitments = self.mixnets[question].shuffle(orig_answers)
self.mixed_questions.append(EosList(shuffled_answers))
self.commitments.append(EosList(commitments))
return True
def prove_mixes(self, question=0):
election = self.recurse_parents('eos.base.election.Election')
index = self._instance[1]
self.challenge.append(self.compute_challenge(question))
challenge_bs = InfiniteHashBitStream(self.challenge[question])
self.response.append(EosList())
for k in range(len(self.mixed_questions[question])):
challenge_bit = challenge_bs.read(1)
should_reveal = ((index % 2) == (challenge_bit % 2))
if should_reveal:
response = self.mixnets[question].challenge(k)
self.response[question].append(response)
else:
self.response[question].append(None)
return True
class PSRElection(Election):
2021-10-16 20:44:31 +11:00
is_voters_public = BooleanField(is_hashed=False, default=True)
is_votes_public = BooleanField(is_hashed=False, default=True)
sk = EmbeddedObjectField(SEGPrivateKey, is_protected=True) # TODO: Threshold
public_key = EmbeddedObjectField(SEGPublicKey)
2017-11-24 19:37:48 +11:00
mixing_trustees = EmbeddedObjectListField()
def can_audit(self):
"""Overrides Election.can_audit"""
return True
def verify(self):
"""Overrides Election.verify"""
# Verify ballots
super().verify()
# Verify mixes
for i in range(len(self.questions)):
for j in range(len(self.mixing_trustees)):
self.mixing_trustees[j].verify(i)
# Verify decryption proofs
for q_num in range(len(self.questions)):
raw_result = self.results[q_num]
2017-12-16 21:47:17 +11:00
if isinstance(raw_result, MultipleResult):
raw_result = next(result for result in raw_result.results if isinstance(result, RawResult))
for answer_num in range(len(raw_result.plaintexts)):
# Input and output blocks:
plaintexts = raw_result.plaintexts[answer_num]
ciphertexts = self.mixing_trustees[-1].mixed_questions[q_num][answer_num].blocks
# Verify ciphertexts
if len(plaintexts) != len(ciphertexts):
raise Exception('Different number of plaintexts and ciphertexts')
for i in range(len(ciphertexts)):
if ciphertexts[i] != plaintexts[i].ciphertext:
raise Exception('Ciphertext does not match mixnet output')
# Verify decryption
for plaintext in plaintexts:
if not plaintext.is_proof_valid():
raise Exception('Proof of decryption is not valid')
# Verify block combination
bs = BitStream.unmap(plaintexts, lambda plaintext: plaintext.message, self.public_key.nbits())
m = bs.read_string()
answer = EosObject.deserialise_and_unwrap(EosObject.from_json(m))
if answer != raw_result.answers[answer_num]:
raise Exception('Result does not match claimed decryption')