microsoft
/
electionguard-python
зеркало из https://github.com/microsoft/electionguard-python.git
1
0
Форкнуть 0
Код Задачи Пакеты Проекты Релизы Вики Активность

#690 Guardian Verify Backups (#711) 

* Refactor should_run logic into each stage service

* Guardians perform verifications

* lint the gui with mypy and fix the resulting issues

* Generate init.py's for the gui

* fix more linting errors

* Fix more linting errors

* fix typo per PR
This commit is contained in:
Lee Richardson 2022-07-21 10:27:47 -04:00 коммит произвёл GitHub
Родитель 7f5bffaf80
Коммит 8c70e6e1ec
Не найден ключ, соответствующий данной подписи
Идентификатор ключа GPG: 4AEE18F83AFDEB23
25 изменённых файлов: 702 добавлений и 103 удалений
Показать статистику Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

2
Makefile
Просмотреть файл

@ -97,7 +97,7 @@ blackcheck:
poetry run black --check .
mypy:
poetry run mypy src/electionguard src/electionguard_tools src/electionguard_cli stubs
poetry run mypy src/electionguard src/electionguard_tools src/electionguard_cli src/electionguard_gui stubs
validate:
@echo ✅ VALIDATE

1
pyproject.toml
Просмотреть файл

@ -24,6 +24,7 @@ packages = [
{ include = "electionguard", from = "src" },
{ include = "electionguard_tools", from = "src" },
{ include = "electionguard_cli", from = "src" },
{ include = "electionguard_gui", from = "src" },
]

6
src/electionguard/guardian.py
Просмотреть файл

@ -363,8 +363,10 @@ class Guardian:
"""
backup = self._guardian_election_partial_key_backups.get(guardian_id)
public_key = self._guardian_election_public_keys.get(guardian_id)
if backup is None or public_key is None:
return None
if backup is None:
raise ValueError(f"No backup exists for {guardian_id}")
if public_key is None:
raise ValueError(f"No public key exists for {guardian_id}")
return verify_election_partial_key_backup(
self.id, backup, public_key, self._election_keys
)

172
src/electionguard_gui/__init__.py Normal file
Просмотреть файл

@ -0,0 +1,172 @@
from electionguard_gui import components
from electionguard_gui import containers
from electionguard_gui import eel_utils
from electionguard_gui import gui_setup_election
from electionguard_gui import main_app
from electionguard_gui import models
from electionguard_gui import services
from electionguard_gui import start
from electionguard_gui.components import (
ComponentBase,
CreateKeyCeremonyComponent,
KeyCeremonyDetailsComponent,
KeyCeremonyListComponent,
SetupElectionComponent,
component_base,
create_key_ceremony_component,
key_ceremony_details_component,
key_ceremony_list_component,
make_js_key_ceremony,
send_key_ceremonies_to_ui,
setup_election_component,
)
from electionguard_gui.containers import (
Container,
)
from electionguard_gui.eel_utils import (
convert_utc_to_local,
eel_fail,
eel_success,
utc_to_str,
)
from electionguard_gui.gui_setup_election import (
GuiSetupInputRetrievalStep,
gui_setup_input_retrieval_step,
)
from electionguard_gui.main_app import (
MainApp,
)
from electionguard_gui.models import (
KeyCeremonyDto,
KeyCeremonyStates,
key_ceremony_dto,
key_ceremony_states,
)
from electionguard_gui.services import (
AuthorizationService,
DB_HOST_KEY,
DB_PASSWORD_KEY,
DbService,
EelLogService,
GuardianService,
IS_ADMIN_KEY,
KeyCeremonyS1JoinService,
KeyCeremonyS2AnnounceService,
KeyCeremonyS3MakeBackupService,
KeyCeremonyS4ShareBackupService,
KeyCeremonyS5VerifyBackupService,
KeyCeremonyService,
KeyCeremonyStageBase,
KeyCeremonyStateService,
ServiceBase,
announce_guardians,
authorization_service,
backup_to_dict,
configuration_service,
db_serialization_service,
db_service,
eel_log_service,
get_db_host,
get_db_password,
get_is_admin,
get_key_ceremony_status,
guardian_service,
key_ceremony_s1_join_service,
key_ceremony_s2_announce_service,
key_ceremony_s3_make_backup_service,
key_ceremony_s4_share_backup_service,
key_ceremony_s5_verify_backup_service,
key_ceremony_service,
key_ceremony_stage_base,
key_ceremony_stages,
key_ceremony_state_service,
make_guardian,
make_mediator,
public_key_to_dict,
service_base,
status_descriptions,
verification_to_dict,
)
from electionguard_gui.start import (
run,
)
__all__ = [
"AuthorizationService",
"ComponentBase",
"Container",
"CreateKeyCeremonyComponent",
"DB_HOST_KEY",
"DB_PASSWORD_KEY",
"DbService",
"EelLogService",
"GuardianService",
"GuiSetupInputRetrievalStep",
"IS_ADMIN_KEY",
"KeyCeremonyDetailsComponent",
"KeyCeremonyDto",
"KeyCeremonyListComponent",
"KeyCeremonyS1JoinService",
"KeyCeremonyS2AnnounceService",
"KeyCeremonyS3MakeBackupService",
"KeyCeremonyS4ShareBackupService",
"KeyCeremonyS5VerifyBackupService",
"KeyCeremonyService",
"KeyCeremonyStageBase",
"KeyCeremonyStateService",
"KeyCeremonyStates",
"MainApp",
"ServiceBase",
"SetupElectionComponent",
"announce_guardians",
"authorization_service",
"backup_to_dict",
"component_base",
"components",
"configuration_service",
"containers",
"convert_utc_to_local",
"create_key_ceremony_component",
"db_serialization_service",
"db_service",
"eel_fail",
"eel_log_service",
"eel_success",
"eel_utils",
"get_db_host",
"get_db_password",
"get_is_admin",
"get_key_ceremony_status",
"guardian_service",
"gui_setup_election",
"gui_setup_input_retrieval_step",
"key_ceremony_details_component",
"key_ceremony_dto",
"key_ceremony_list_component",
"key_ceremony_s1_join_service",
"key_ceremony_s2_announce_service",
"key_ceremony_s3_make_backup_service",
"key_ceremony_s4_share_backup_service",
"key_ceremony_s5_verify_backup_service",
"key_ceremony_service",
"key_ceremony_stage_base",
"key_ceremony_stages",
"key_ceremony_state_service",
"key_ceremony_states",
"main_app",
"make_guardian",
"make_js_key_ceremony",
"make_mediator",
"models",
"public_key_to_dict",
"run",
"send_key_ceremonies_to_ui",
"service_base",
"services",
"setup_election_component",
"start",
"status_descriptions",
"utc_to_str",
"verification_to_dict",
]

38
src/electionguard_gui/components/__init__.py Normal file
Просмотреть файл

@ -0,0 +1,38 @@
from electionguard_gui.components import component_base
from electionguard_gui.components import create_key_ceremony_component
from electionguard_gui.components import key_ceremony_details_component
from electionguard_gui.components import key_ceremony_list_component
from electionguard_gui.components import setup_election_component
from electionguard_gui.components.component_base import (
ComponentBase,
)
from electionguard_gui.components.create_key_ceremony_component import (
CreateKeyCeremonyComponent,
)
from electionguard_gui.components.key_ceremony_details_component import (
KeyCeremonyDetailsComponent,
)
from electionguard_gui.components.key_ceremony_list_component import (
KeyCeremonyListComponent,
make_js_key_ceremony,
send_key_ceremonies_to_ui,
)
from electionguard_gui.components.setup_election_component import (
SetupElectionComponent,
)
__all__ = [
"ComponentBase",
"CreateKeyCeremonyComponent",
"KeyCeremonyDetailsComponent",
"KeyCeremonyListComponent",
"SetupElectionComponent",
"component_base",
"create_key_ceremony_component",
"key_ceremony_details_component",
"key_ceremony_list_component",
"make_js_key_ceremony",
"send_key_ceremonies_to_ui",
"setup_election_component",
]

13
src/electionguard_gui/components/create_key_ceremony_component.py
Просмотреть файл

@ -30,7 +30,10 @@ class CreateKeyCeremonyComponent(ComponentBase):
self, key_ceremony_name: str, guardian_count: int, quorum: int
) -> dict[str, Any]:
if guardian_count < quorum:
return eel_fail("Guardian count must be greater than or equal to quorum")
result: dict[str, Any] = eel_fail(
"Guardian count must be greater than or equal to quorum"
)
return result
self.log.debug(
"Starting ceremony: "
@ -44,8 +47,8 @@ class CreateKeyCeremonyComponent(ComponentBase):
)
if existing_key_ceremonies:
self.log.debug(f"record '{key_ceremony_name}' already exists")
result: dict[str, Any] = eel_fail("Key ceremony name already exists")
return result
fail_result: dict[str, Any] = eel_fail("Key ceremony name already exists")
return fail_result
key_ceremony = {
"key_ceremony_name": key_ceremony_name,
"guardian_count": guardian_count,
@ -56,10 +59,12 @@ class CreateKeyCeremonyComponent(ComponentBase):
"other_keys": [],
"backups": [],
"shared_backups": [],
"verifications": [],
"created_by": self._auth_service.get_user_id(),
"created_at": datetime.utcnow(),
}
inserted_id = db.key_ceremonies.insert_one(key_ceremony).inserted_id
self.log.debug(f"created '{key_ceremony_name}' record, id: {inserted_id}")
self._key_ceremony_service.notify_changed(db, inserted_id)
return eel_success(str(inserted_id))
result = eel_success(str(inserted_id))
return result

57
src/electionguard_gui/components/key_ceremony_details_component.py
Просмотреть файл

@ -1,3 +1,4 @@
from typing import List
import eel
from pymongo.database import Database
@ -16,13 +17,16 @@ from electionguard_gui.services.key_ceremony_stages.key_ceremony_s4_share_backup
)
from electionguard_gui.models.key_ceremony_dto import KeyCeremonyDto
from electionguard_gui.services.key_ceremony_stages.key_ceremony_s5_verify_backup_service import (
KeyCeremonyS5VerifyBackupService,
)
from electionguard_gui.services.key_ceremony_stages.key_ceremony_stage_base import (
KeyCeremonyStageBase,
)
from electionguard_gui.services.key_ceremony_state_service import (
KeyCeremonyStateService,
get_key_ceremony_status,
)
from electionguard_gui.models.key_ceremony_states import (
KeyCeremonyStates,
)
from electionguard_gui.services.authorization_service import AuthorizationService
from electionguard_gui.components.component_base import ComponentBase
from electionguard_gui.services.key_ceremony_service import (
@ -36,9 +40,7 @@ class KeyCeremonyDetailsComponent(ComponentBase):
_auth_service: AuthorizationService
_ceremony_state_service: KeyCeremonyStateService
_key_ceremony_s1_join_service: KeyCeremonyS1JoinService
_key_ceremony_s2_announce_service: KeyCeremonyS2AnnounceService
_key_ceremony_s3_make_backup_service: KeyCeremonyS3MakeBackupService
_key_ceremony_s4_share_backup_service: KeyCeremonyS4ShareBackupService
key_ceremony_watch_stages: List[KeyCeremonyStageBase]
def __init__(
self,
@ -49,29 +51,25 @@ class KeyCeremonyDetailsComponent(ComponentBase):
key_ceremony_s2_announce_service: KeyCeremonyS2AnnounceService,
key_ceremony_s3_make_backup_service: KeyCeremonyS3MakeBackupService,
key_ceremony_s4_share_backup_service: KeyCeremonyS4ShareBackupService,
key_ceremony_s5_verification_service: KeyCeremonyS5VerifyBackupService,
) -> None:
super().__init__()
self._key_ceremony_service = key_ceremony_service
self._ceremony_state_service = key_ceremony_state_service
self._auth_service = auth_service
self._key_ceremony_s1_join_service = key_ceremony_s1_join_service
self._key_ceremony_s2_announce_service = key_ceremony_s2_announce_service
self._key_ceremony_s3_make_backup_service = key_ceremony_s3_make_backup_service
self._key_ceremony_s4_share_backup_service = (
key_ceremony_s4_share_backup_service
)
self.key_ceremony_watch_stages = [
key_ceremony_s2_announce_service,
key_ceremony_s3_make_backup_service,
key_ceremony_s4_share_backup_service,
key_ceremony_s5_verification_service,
]
def expose(self) -> None:
eel.expose(self.join_key_ceremony)
eel.expose(self.watch_key_ceremony)
eel.expose(self.stop_watching_key_ceremony)
def can_join_key_ceremony(self, key_ceremony: KeyCeremonyDto) -> bool:
user_id = self._auth_service.get_user_id()
already_joined = user_id in key_ceremony.guardians_joined
is_admin = self._auth_service.is_admin()
return not already_joined and not is_admin
def watch_key_ceremony(self, key_ceremony_id: str) -> None:
db = self.db_service.get_db()
# retrieve and send the key ceremony to the client
@ -87,26 +85,15 @@ class KeyCeremonyDetailsComponent(ComponentBase):
self.log.debug(
f"on_key_ceremony_changed key_ceremony_id: '{key_ceremony_id}', current_user_id: '{current_user_id}'"
)
is_admin = self._auth_service.is_admin()
is_guardian = not is_admin
db = self.db_service.get_db()
key_ceremony = self.get_ceremony(db, key_ceremony_id)
state = self._ceremony_state_service.get_key_ceremony_state(key_ceremony)
self.log.debug(f"{key_ceremony_id} state = '{state}'")
if is_admin and state == KeyCeremonyStates.PendingAdminAnnounce:
self._key_ceremony_s2_announce_service.run(db, key_ceremony)
current_user_backups = key_ceremony.get_backup_count_for_user(current_user_id)
current_user_backup_exists = current_user_backups > 0
if (
is_guardian
and state == KeyCeremonyStates.PendingGuardianBackups
and not current_user_backup_exists
):
self._key_ceremony_s3_make_backup_service.run(db, key_ceremony)
if is_admin and state == KeyCeremonyStates.PendingAdminToShareBackups:
self._key_ceremony_s4_share_backup_service.run(db, key_ceremony)
for stage in self.key_ceremony_watch_stages:
if stage.should_run(key_ceremony, state):
stage.run(db, key_ceremony)
break
key_ceremony = self.get_ceremony(db, key_ceremony_id)
new_state = self._ceremony_state_service.get_key_ceremony_state(key_ceremony)
@ -128,3 +115,9 @@ class KeyCeremonyDetailsComponent(ComponentBase):
key_ceremony = self._key_ceremony_service.get(db, id)
key_ceremony.can_join = self.can_join_key_ceremony(key_ceremony)
return key_ceremony
def can_join_key_ceremony(self, key_ceremony: KeyCeremonyDto) -> bool:
user_id = self._auth_service.get_user_id()
already_joined = user_id in key_ceremony.guardians_joined
is_admin = self._auth_service.is_admin()
return not already_joined and not is_admin

10
src/electionguard_gui/components/setup_election_component.py
Просмотреть файл

@ -1,4 +1,5 @@
import tempfile
from typing import Any, Dict
import eel
from electionguard_cli.cli_steps import KeyCeremonyStep
@ -24,7 +25,7 @@ class SetupElectionComponent(ComponentBase):
# pylint: disable=no-self-use
def setup_election(
self, guardian_count: int, quorum: int, verification_url: str, manifest: str
) -> str:
) -> Dict[str, Any]:
election_inputs = GuiSetupInputRetrievalStep().get_gui_inputs(
guardian_count, quorum, verification_url, manifest
)
@ -41,6 +42,7 @@ class SetupElectionComponent(ComponentBase):
self.log.debug(
f"Setup complete, context: {context_file}, constants: {constants_file}"
)
with open(context_file, "r", encoding="utf-8") as context_file:
context_raw: str = context_file.read()
return eel_success(context_raw)
with open(context_file, "r", encoding="utf-8") as context_file_io:
context_raw: str = context_file_io.read()
result: Dict[Any, str] = eel_success(context_raw)
return result

70
src/electionguard_gui/containers.py
Просмотреть файл

@ -1,4 +1,5 @@
from dependency_injector import containers, providers
from dependency_injector.providers import Factory, Singleton
from electionguard_gui.components.create_key_ceremony_component import (
CreateKeyCeremonyComponent,
)
@ -14,6 +15,7 @@ from electionguard_gui.services.authorization_service import AuthorizationServic
from electionguard_gui.services.db_service import DbService
from electionguard_gui.services.eel_log_service import EelLogService
from electionguard_gui.services.guardian_service import GuardianService
from electionguard_gui.services.key_ceremony_service import KeyCeremonyService
from electionguard_gui.services.key_ceremony_stages.key_ceremony_s1_join_service import (
KeyCeremonyS1JoinService,
@ -27,6 +29,9 @@ from electionguard_gui.services.key_ceremony_stages.key_ceremony_s3_make_backup_
from electionguard_gui.services.key_ceremony_stages.key_ceremony_s4_share_backup_service import (
KeyCeremonyS4ShareBackupService,
)
from electionguard_gui.services.key_ceremony_stages.key_ceremony_s5_verify_backup_service import (
KeyCeremonyS5VerifyBackupService,
)
from electionguard_gui.services.key_ceremony_state_service import (
KeyCeremonyStateService,
)
@ -36,58 +41,92 @@ class Container(containers.DeclarativeContainer):
"""Responsible for dependency injection and how components are wired together"""
# services
log_service = providers.Factory(EelLogService)
db_service = providers.Singleton(DbService, log_service=log_service)
key_ceremony_service = providers.Factory(KeyCeremonyService, db_service=db_service)
authorization_service = providers.Singleton(AuthorizationService)
key_ceremony_state_service = providers.Factory(
log_service: Factory[EelLogService] = providers.Factory(EelLogService)
db_service: Singleton[DbService] = providers.Singleton(
DbService, log_service=log_service
)
key_ceremony_service: Factory[KeyCeremonyService] = providers.Factory(
KeyCeremonyService, db_service=db_service
)
authorization_service: Singleton[AuthorizationService] = providers.Singleton(
AuthorizationService
)
key_ceremony_state_service: Factory[KeyCeremonyStateService] = providers.Factory(
KeyCeremonyStateService, log_service=log_service
)
guardian_service: Factory[GuardianService] = providers.Factory(
GuardianService, log_service=log_service
)
# key ceremony services
key_ceremony_s1_join_service = providers.Factory(
key_ceremony_s1_join_service: Factory[KeyCeremonyS1JoinService] = providers.Factory(
KeyCeremonyS1JoinService,
log_service=log_service,
db_service=db_service,
key_ceremony_service=key_ceremony_service,
auth_service=authorization_service,
key_ceremony_state_service=key_ceremony_state_service,
guardian_service=guardian_service,
)
key_ceremony_s2_announce_service = providers.Factory(
key_ceremony_s2_announce_service: Factory[
KeyCeremonyS2AnnounceService
] = providers.Factory(
KeyCeremonyS2AnnounceService,
log_service=log_service,
db_service=db_service,
key_ceremony_service=key_ceremony_service,
auth_service=authorization_service,
key_ceremony_state_service=key_ceremony_state_service,
guardian_service=guardian_service,
)
key_ceremony_s3_make_backup_service = providers.Factory(
key_ceremony_s3_make_backup_service: Factory[
KeyCeremonyS3MakeBackupService
] = providers.Factory(
KeyCeremonyS3MakeBackupService,
log_service=log_service,
db_service=db_service,
key_ceremony_service=key_ceremony_service,
auth_service=authorization_service,
key_ceremony_state_service=key_ceremony_state_service,
guardian_service=guardian_service,
)
key_ceremony_s4_share_backup_service = providers.Factory(
key_ceremony_s4_share_backup_service: Factory[
KeyCeremonyS4ShareBackupService
] = providers.Factory(
KeyCeremonyS4ShareBackupService,
log_service=log_service,
db_service=db_service,
key_ceremony_service=key_ceremony_service,
auth_service=authorization_service,
key_ceremony_state_service=key_ceremony_state_service,
guardian_service=guardian_service,
)
key_ceremony_s5_verification_service: Factory[
KeyCeremonyS5VerifyBackupService
] = providers.Factory(
KeyCeremonyS5VerifyBackupService,
log_service=log_service,
db_service=db_service,
key_ceremony_service=key_ceremony_service,
auth_service=authorization_service,
key_ceremony_state_service=key_ceremony_state_service,
guardian_service=guardian_service,
)
# components
guardian_home_component = providers.Factory(
guardian_home_component: Factory[KeyCeremonyListComponent] = providers.Factory(
KeyCeremonyListComponent, key_ceremony_service=key_ceremony_service
)
create_key_ceremony_component = providers.Factory(
create_key_ceremony_component: Factory[
CreateKeyCeremonyComponent
] = providers.Factory(
CreateKeyCeremonyComponent,
key_ceremony_service=key_ceremony_service,
auth_service=authorization_service,
)
key_ceremony_details_component = providers.Factory(
key_ceremony_details_component: Factory[
KeyCeremonyDetailsComponent
] = providers.Factory(
KeyCeremonyDetailsComponent,
key_ceremony_service=key_ceremony_service,
auth_service=authorization_service,
@ -96,11 +135,14 @@ class Container(containers.DeclarativeContainer):
key_ceremony_s2_announce_service=key_ceremony_s2_announce_service,
key_ceremony_s3_make_backup_service=key_ceremony_s3_make_backup_service,
key_ceremony_s4_share_backup_service=key_ceremony_s4_share_backup_service,
key_ceremony_s5_verification_service=key_ceremony_s5_verification_service,
)
setup_election_component: Factory[SetupElectionComponent] = providers.Factory(
SetupElectionComponent
)
setup_election_component = providers.Factory(SetupElectionComponent)
# main
main_app = providers.Factory(
main_app: Factory[MainApp] = providers.Factory(
MainApp,
log_service=log_service,
db_service=db_service,

7
src/electionguard_gui/gui_setup_election/__init__.py Normal file
Просмотреть файл

@ -0,0 +1,7 @@
from electionguard_gui.gui_setup_election import gui_setup_input_retrieval_step
from electionguard_gui.gui_setup_election.gui_setup_input_retrieval_step import (
GuiSetupInputRetrievalStep,
)
__all__ = ["GuiSetupInputRetrievalStep", "gui_setup_input_retrieval_step"]

16
src/electionguard_gui/models/__init__.py Normal file
Просмотреть файл

@ -0,0 +1,16 @@
from electionguard_gui.models import key_ceremony_dto
from electionguard_gui.models import key_ceremony_states
from electionguard_gui.models.key_ceremony_dto import (
KeyCeremonyDto,
)
from electionguard_gui.models.key_ceremony_states import (
KeyCeremonyStates,
)
__all__ = [
"KeyCeremonyDto",
"KeyCeremonyStates",
"key_ceremony_dto",
"key_ceremony_states",
]

21
src/electionguard_gui/models/key_ceremony_dto.py
Просмотреть файл

@ -25,6 +25,7 @@ class KeyCeremonyDto:
self.created_at_utc = key_ceremony["created_at"]
self.created_at_str = utc_to_str(self.created_at_utc)
self.keys = [_dict_to_election_public_key(key) for key in key_ceremony["keys"]]
self.verifications = key_ceremony["verifications"]
def to_dict(self) -> dict[str, Any]:
return {
@ -64,6 +65,26 @@ class KeyCeremonyDto:
backups = [backup for backup in self.backups if backup["owner_id"] == user_id]
return len(backups)
def get_verification_count_for_user(self, user_id: str) -> int:
return len(
[
verification
for verification in self.verifications
if verification["designated_id"] == user_id
]
)
def get_shared_backups_for_guardian(
self, guardian_id: str
) -> List[ElectionPartialKeyBackup]:
shared_backup_wrapper = next(
filter(
lambda backup: backup["owner_id"] == guardian_id, self.shared_backups
)
)
backups = shared_backup_wrapper["backups"]
return [_dict_to_backup(backup) for backup in backups]
def get_backups(self) -> List[ElectionPartialKeyBackup]:
return [_dict_to_backup(backup) for backup in self.backups]

110
src/electionguard_gui/services/__init__.py Normal file
Просмотреть файл

@ -0,0 +1,110 @@
from electionguard_gui.services import authorization_service
from electionguard_gui.services import configuration_service
from electionguard_gui.services import db_serialization_service
from electionguard_gui.services import db_service
from electionguard_gui.services import eel_log_service
from electionguard_gui.services import guardian_service
from electionguard_gui.services import key_ceremony_service
from electionguard_gui.services import key_ceremony_stages
from electionguard_gui.services import key_ceremony_state_service
from electionguard_gui.services import service_base
from electionguard_gui.services.authorization_service import (
AuthorizationService,
)
from electionguard_gui.services.configuration_service import (
DB_HOST_KEY,
DB_PASSWORD_KEY,
IS_ADMIN_KEY,
get_db_host,
get_db_password,
get_is_admin,
)
from electionguard_gui.services.db_serialization_service import (
backup_to_dict,
public_key_to_dict,
verification_to_dict,
)
from electionguard_gui.services.db_service import (
DbService,
)
from electionguard_gui.services.eel_log_service import (
EelLogService,
)
from electionguard_gui.services.guardian_service import (
GuardianService,
announce_guardians,
make_guardian,
make_mediator,
)
from electionguard_gui.services.key_ceremony_service import (
KeyCeremonyService,
)
from electionguard_gui.services.key_ceremony_stages import (
KeyCeremonyS1JoinService,
KeyCeremonyS2AnnounceService,
KeyCeremonyS3MakeBackupService,
KeyCeremonyS4ShareBackupService,
KeyCeremonyS5VerifyBackupService,
KeyCeremonyStageBase,
key_ceremony_s1_join_service,
key_ceremony_s2_announce_service,
key_ceremony_s3_make_backup_service,
key_ceremony_s4_share_backup_service,
key_ceremony_s5_verify_backup_service,
key_ceremony_stage_base,
)
from electionguard_gui.services.key_ceremony_state_service import (
KeyCeremonyStateService,
get_key_ceremony_status,
status_descriptions,
)
from electionguard_gui.services.service_base import (
ServiceBase,
)
__all__ = [
"AuthorizationService",
"DB_HOST_KEY",
"DB_PASSWORD_KEY",
"DbService",
"EelLogService",
"GuardianService",
"IS_ADMIN_KEY",
"KeyCeremonyS1JoinService",
"KeyCeremonyS2AnnounceService",
"KeyCeremonyS3MakeBackupService",
"KeyCeremonyS4ShareBackupService",
"KeyCeremonyS5VerifyBackupService",
"KeyCeremonyService",
"KeyCeremonyStageBase",
"KeyCeremonyStateService",
"ServiceBase",
"announce_guardians",
"authorization_service",
"backup_to_dict",
"configuration_service",
"db_serialization_service",
"db_service",
"eel_log_service",
"get_db_host",
"get_db_password",
"get_is_admin",
"get_key_ceremony_status",
"guardian_service",
"key_ceremony_s1_join_service",
"key_ceremony_s2_announce_service",
"key_ceremony_s3_make_backup_service",
"key_ceremony_s4_share_backup_service",
"key_ceremony_s5_verify_backup_service",
"key_ceremony_service",
"key_ceremony_stage_base",
"key_ceremony_stages",
"key_ceremony_state_service",
"make_guardian",
"make_mediator",
"public_key_to_dict",
"service_base",
"status_descriptions",
"verification_to_dict",
]

8
src/electionguard_gui/services/authorization_service.py
Просмотреть файл

@ -16,6 +16,11 @@ class AuthorizationService(ServiceBase):
eel.expose(self.set_user_id)
eel.expose(self.is_admin)
def get_required_user_id(self) -> str:
if self.user_id is None:
raise Exception("User must be logged in")
return self.user_id
def get_user_id(self) -> Optional[str]:
return self.user_id
@ -24,4 +29,5 @@ class AuthorizationService(ServiceBase):
# pylint: disable=no-self-use
def is_admin(self) -> bool:
return get_is_admin()
is_admin: bool = get_is_admin()
return is_admin

17
src/electionguard_gui/services/db_serialization_service.py
Просмотреть файл

@ -1,5 +1,9 @@
from typing import Any
from electionguard.key_ceremony import ElectionPartialKeyBackup, ElectionPublicKey
from electionguard.key_ceremony import (
ElectionPartialKeyBackup,
ElectionPartialKeyVerification,
ElectionPublicKey,
)
def public_key_to_dict(key: ElectionPublicKey) -> dict[str, Any]:
@ -33,3 +37,14 @@ def backup_to_dict(backup: ElectionPartialKeyBackup) -> dict[str, Any]:
"mac": coordinate.mac,
},
}
def verification_to_dict(
verification: ElectionPartialKeyVerification,
) -> dict[str, Any]:
return {
"owner_id": verification.owner_id,
"designated_id": verification.designated_id,
"verifier_id": verification.verifier_id,
"verified": verification.verified,
}

45
src/electionguard_gui/services/guardian_service.py
Просмотреть файл

@ -1,7 +1,50 @@
from electionguard.guardian import Guardian
from os import getcwd, path
from electionguard.serialize import from_file, to_file
from electionguard.guardian import Guardian, PrivateGuardianRecord
from electionguard.key_ceremony import CeremonyDetails
from electionguard.key_ceremony_mediator import KeyCeremonyMediator
from electionguard_gui.models.key_ceremony_dto import KeyCeremonyDto
from electionguard_gui.services.eel_log_service import EelLogService
from electionguard_gui.services.service_base import ServiceBase
from electionguard_tools.helpers.export import GUARDIAN_PREFIX
class GuardianService(ServiceBase):
"""Responsible for functionality related to guardians"""
_log: EelLogService
def __init__(self, log_service: EelLogService) -> None:
self._log = log_service
def save_guardian(self, guardian: Guardian, key_ceremony: KeyCeremonyDto) -> None:
private_guardian_record = guardian.export_private_data()
file_name = GUARDIAN_PREFIX + private_guardian_record.guardian_id
file_path = path.join(getcwd(), "gui_private_keys", key_ceremony.id)
file = to_file(private_guardian_record, file_name, file_path)
self._log.warn(
f"Guardian private data saved to {file}. This data should be carefully protected and never shared."
)
def load_guardian(self, guardian_id: str, key_ceremony: KeyCeremonyDto) -> Guardian:
file_name = GUARDIAN_PREFIX + guardian_id + ".json"
file_path = path.join(getcwd(), "gui_private_keys", key_ceremony.id, file_name)
self._log.debug(f"loading guardian from {file_path}")
private_guardian_record = from_file(PrivateGuardianRecord, file_path)
return Guardian.from_private_record(
private_guardian_record,
key_ceremony.guardian_count,
key_ceremony.quorum,
)
def load_other_keys(
self, key_ceremony: KeyCeremonyDto, current_user_id: str, guardian: Guardian
) -> None:
current_user_other_keys = key_ceremony.find_other_keys_for_user(current_user_id)
for other_key in current_user_other_keys:
other_user = other_key.owner_id
self._log.debug(f"saving other_key from {other_user} for {current_user_id}")
guardian.save_guardian_key(other_key)
def make_guardian(

21
src/electionguard_gui/services/key_ceremony_service.py
Просмотреть файл

@ -4,11 +4,16 @@ from pymongo.database import Database
from pymongo import CursorType
from bson import ObjectId
import eel
from electionguard.key_ceremony import ElectionPartialKeyBackup, ElectionPublicKey
from electionguard.key_ceremony import (
ElectionPartialKeyBackup,
ElectionPartialKeyVerification,
ElectionPublicKey,
)
from electionguard_gui.models.key_ceremony_dto import KeyCeremonyDto
from electionguard_gui.services.db_serialization_service import (
backup_to_dict,
public_key_to_dict,
verification_to_dict,
)
from electionguard_gui.services.db_service import DbService
@ -135,3 +140,17 @@ class KeyCeremonyService(ServiceBase):
{"_id": ObjectId(key_ceremony_id)},
{"$push": {"shared_backups": {"$each": shared_backups}}},
)
def append_verifications(
self,
db: Database,
key_ceremony_id: str,
verifications: List[ElectionPartialKeyVerification],
) -> None:
verifications_dict = [
verification_to_dict(verification) for verification in verifications
]
db.key_ceremonies.update_one(
{"_id": ObjectId(key_ceremony_id)},
{"$push": {"verifications": {"$each": verifications_dict}}},
)

48
src/electionguard_gui/services/key_ceremony_stages/__init__.py Normal file
Просмотреть файл

@ -0,0 +1,48 @@
from electionguard_gui.services.key_ceremony_stages import key_ceremony_s1_join_service
from electionguard_gui.services.key_ceremony_stages import (
key_ceremony_s2_announce_service,
)
from electionguard_gui.services.key_ceremony_stages import (
key_ceremony_s3_make_backup_service,
)
from electionguard_gui.services.key_ceremony_stages import (
key_ceremony_s4_share_backup_service,
)
from electionguard_gui.services.key_ceremony_stages import (
key_ceremony_s5_verify_backup_service,
)
from electionguard_gui.services.key_ceremony_stages import key_ceremony_stage_base
from electionguard_gui.services.key_ceremony_stages.key_ceremony_s1_join_service import (
KeyCeremonyS1JoinService,
)
from electionguard_gui.services.key_ceremony_stages.key_ceremony_s2_announce_service import (
KeyCeremonyS2AnnounceService,
)
from electionguard_gui.services.key_ceremony_stages.key_ceremony_s3_make_backup_service import (
KeyCeremonyS3MakeBackupService,
)
from electionguard_gui.services.key_ceremony_stages.key_ceremony_s4_share_backup_service import (
KeyCeremonyS4ShareBackupService,
)
from electionguard_gui.services.key_ceremony_stages.key_ceremony_s5_verify_backup_service import (
KeyCeremonyS5VerifyBackupService,
)
from electionguard_gui.services.key_ceremony_stages.key_ceremony_stage_base import (
KeyCeremonyStageBase,
)
__all__ = [
"KeyCeremonyS1JoinService",
"KeyCeremonyS2AnnounceService",
"KeyCeremonyS3MakeBackupService",
"KeyCeremonyS4ShareBackupService",
"KeyCeremonyS5VerifyBackupService",
"KeyCeremonyStageBase",
"key_ceremony_s1_join_service",
"key_ceremony_s2_announce_service",
"key_ceremony_s3_make_backup_service",
"key_ceremony_s4_share_backup_service",
"key_ceremony_s5_verify_backup_service",
"key_ceremony_stage_base",
]

17
src/electionguard_gui/services/key_ceremony_stages/key_ceremony_s1_join_service.py
Просмотреть файл

@ -1,14 +1,10 @@
from os import getcwd, path
from pymongo.database import Database
from electionguard import to_file
from electionguard.guardian import Guardian
from electionguard_gui.models.key_ceremony_dto import KeyCeremonyDto
from electionguard_gui.services.key_ceremony_stages.key_ceremony_stage_base import (
KeyCeremonyStageBase,
)
from electionguard_gui.services.guardian_service import make_guardian
from electionguard_tools import GUARDIAN_PREFIX
class KeyCeremonyS1JoinService(KeyCeremonyStageBase):
@ -16,7 +12,7 @@ class KeyCeremonyS1JoinService(KeyCeremonyStageBase):
def run(self, db: Database, key_ceremony: KeyCeremonyDto) -> None:
key_ceremony_id = key_ceremony.id
user_id = self._auth_service.get_user_id()
user_id = self._auth_service.get_required_user_id()
self._key_ceremony_service.append_guardian_joined(db, key_ceremony_id, user_id)
# refresh key ceremony to get the list of guardians with the authoritative order they joined in
key_ceremony = self._key_ceremony_service.get(db, key_ceremony_id)
@ -27,19 +23,10 @@ class KeyCeremonyS1JoinService(KeyCeremonyStageBase):
f"user {user_id} about to join key ceremony {key_ceremony_id} as guardian #{guardian_number}"
)
guardian = make_guardian(user_id, guardian_number, key_ceremony)
self.save_guardian(guardian, key_ceremony)
self._guardian_service.save_guardian(guardian, key_ceremony)
public_key = guardian.share_key()
self._key_ceremony_service.append_key(db, key_ceremony_id, public_key)
self.log.debug(
f"{user_id} joined key ceremony {key_ceremony_id} as guardian #{guardian_number}"
)
self._key_ceremony_service.notify_changed(db, key_ceremony_id)
def save_guardian(self, guardian: Guardian, key_ceremony: KeyCeremonyDto) -> None:
private_guardian_record = guardian.export_private_data()
file_name = GUARDIAN_PREFIX + private_guardian_record.guardian_id
file_path = path.join(getcwd(), "gui_private_keys", key_ceremony.id)
file = to_file(private_guardian_record, file_name, file_path)
self.log.warn(
f"Guardian private data saved to {file}. This data should be carefully protected and never shared."
)

8
src/electionguard_gui/services/key_ceremony_stages/key_ceremony_s2_announce_service.py
Просмотреть файл

@ -3,6 +3,7 @@ from pymongo.database import Database
from electionguard.key_ceremony import ElectionPublicKey
from electionguard.utils import get_optional
from electionguard_gui.models.key_ceremony_dto import KeyCeremonyDto
from electionguard_gui.models.key_ceremony_states import KeyCeremonyStates
from electionguard_gui.services.db_serialization_service import public_key_to_dict
from electionguard_gui.services.guardian_service import (
announce_guardians,
@ -16,6 +17,13 @@ from electionguard_gui.services.key_ceremony_stages.key_ceremony_stage_base impo
class KeyCeremonyS2AnnounceService(KeyCeremonyStageBase):
"""Responsible for stage 2 of the key ceremony where admins announce the key ceremony"""
def should_run(
self, key_ceremony: KeyCeremonyDto, state: KeyCeremonyStates
) -> bool:
is_admin = self._auth_service.is_admin()
should_run: bool = is_admin and state == KeyCeremonyStates.PendingAdminAnnounce
return should_run
def run(self, db: Database, key_ceremony: KeyCeremonyDto) -> None:
key_ceremony_id = key_ceremony.id
self.log.info("all guardians have joined, announcing guardians")

42
src/electionguard_gui/services/key_ceremony_stages/key_ceremony_s3_make_backup_service.py
Просмотреть файл

@ -1,43 +1,35 @@
from os import getcwd, path
from pymongo.database import Database
from electionguard import Guardian
from electionguard.guardian import PrivateGuardianRecord
from electionguard.serialize import from_file
from electionguard_gui.models.key_ceremony_dto import KeyCeremonyDto
from electionguard_gui.models.key_ceremony_states import KeyCeremonyStates
from electionguard_gui.services.key_ceremony_stages.key_ceremony_stage_base import (
KeyCeremonyStageBase,
)
from electionguard_tools import GUARDIAN_PREFIX
class KeyCeremonyS3MakeBackupService(KeyCeremonyStageBase):
"""Responsible for stage 3 of the key ceremony where guardians create backups to send to the admin."""
def should_run(
self, key_ceremony: KeyCeremonyDto, state: KeyCeremonyStates
) -> bool:
is_guardian = not self._auth_service.is_admin()
current_user_id = self._auth_service.get_required_user_id()
current_user_backups = key_ceremony.get_backup_count_for_user(current_user_id)
current_user_backup_exists = current_user_backups > 0
return (
is_guardian
and state == KeyCeremonyStates.PendingGuardianBackups
and not current_user_backup_exists
)
def run(self, db: Database, key_ceremony: KeyCeremonyDto) -> None:
current_user_id = self._auth_service.get_user_id()
current_user_id = self._auth_service.get_required_user_id()
key_ceremony_id = key_ceremony.id
self.log.debug(f"creating backups for guardian {current_user_id}")
guardian = self.load_guardian(current_user_id, key_ceremony)
current_user_other_keys = key_ceremony.find_other_keys_for_user(current_user_id)
for other_key in current_user_other_keys:
other_user = other_key.owner_id
self.log.debug(f"saving other_key from {other_user} for {current_user_id}")
guardian.save_guardian_key(other_key)
guardian = self._guardian_service.load_guardian(current_user_id, key_ceremony)
self._guardian_service.load_other_keys(key_ceremony, current_user_id, guardian)
guardian.generate_election_partial_key_backups()
backups = guardian.share_election_partial_key_backups()
self._key_ceremony_service.append_backups(db, key_ceremony_id, backups)
# notify the admin that a new guardian has backups
self._key_ceremony_service.notify_changed(db, key_ceremony_id)
def load_guardian(self, guardian_id: str, key_ceremony: KeyCeremonyDto) -> Guardian:
file_name = GUARDIAN_PREFIX + guardian_id + ".json"
file_path = path.join(getcwd(), "gui_private_keys", key_ceremony.id, file_name)
self.log.debug(f"loading guardian from {file_path}")
private_guardian_record = from_file(PrivateGuardianRecord, file_path)
return Guardian.from_private_record(
private_guardian_record,
key_ceremony.guardian_count,
key_ceremony.quorum,
)

9
src/electionguard_gui/services/key_ceremony_stages/key_ceremony_s4_share_backup_service.py
Просмотреть файл

@ -1,6 +1,7 @@
from typing import Any, List
from pymongo.database import Database
from electionguard_gui.models.key_ceremony_dto import KeyCeremonyDto
from electionguard_gui.models.key_ceremony_states import KeyCeremonyStates
from electionguard_gui.services.db_serialization_service import backup_to_dict
from electionguard_gui.services.guardian_service import (
announce_guardians,
@ -17,6 +18,12 @@ class KeyCeremonyS4ShareBackupService(KeyCeremonyStageBase):
back to guardians for verification.
"""
def should_run(
self, key_ceremony: KeyCeremonyDto, state: KeyCeremonyStates
) -> bool:
is_admin: bool = self._auth_service.is_admin()
return is_admin and state == KeyCeremonyStates.PendingAdminToShareBackups
def run(self, db: Database, key_ceremony: KeyCeremonyDto) -> None:
current_user_id = self._auth_service.get_user_id()
self.log.debug(f"sharing backups for admin {current_user_id}")
@ -34,6 +41,8 @@ class KeyCeremonyS4ShareBackupService(KeyCeremonyStageBase):
for guardian_id in key_ceremony.guardians_joined:
self.log.debug(f"sharing backups for guardian {guardian_id}")
guardian_backups = mediator.share_backups(guardian_id)
if guardian_backups is None:
raise Exception("Error sharing backups")
backups_as_dict = [backup_to_dict(backup) for backup in guardian_backups]
shared_backups.append({"owner_id": guardian_id, "backups": backups_as_dict})
return shared_backups

48
src/electionguard_gui/services/key_ceremony_stages/key_ceremony_s5_verify_backup_service.py Normal file
Просмотреть файл

@ -0,0 +1,48 @@
from typing import List
from pymongo.database import Database
from electionguard.key_ceremony import ElectionPartialKeyVerification
from electionguard_gui.models.key_ceremony_dto import KeyCeremonyDto
from electionguard_gui.models.key_ceremony_states import KeyCeremonyStates
from electionguard_gui.services.key_ceremony_stages.key_ceremony_stage_base import (
KeyCeremonyStageBase,
)
class KeyCeremonyS5VerifyBackupService(KeyCeremonyStageBase):
"""Responsible for stage 5 of the key ceremony where guardians verify backups."""
def should_run(
self, key_ceremony: KeyCeremonyDto, state: KeyCeremonyStates
) -> bool:
is_guardian = not self._auth_service.is_admin()
current_user_id = self._auth_service.get_required_user_id()
current_user_verifications = key_ceremony.get_verification_count_for_user(
current_user_id
)
current_user_verification_exists = current_user_verifications > 0
return (
is_guardian
and state == KeyCeremonyStates.PendingGuardiansVerifyBackups
and not current_user_verification_exists
)
def run(self, db: Database, key_ceremony: KeyCeremonyDto) -> None:
current_user_id = self._auth_service.get_required_user_id()
shared_backups = key_ceremony.get_shared_backups_for_guardian(current_user_id)
guardian = self._guardian_service.load_guardian(current_user_id, key_ceremony)
self._guardian_service.load_other_keys(key_ceremony, current_user_id, guardian)
verifications: List[ElectionPartialKeyVerification] = []
for backup in shared_backups:
self.log.debug(
f"verifying backup from {backup.owner_id} to {current_user_id}"
)
guardian.save_election_partial_key_backup(backup)
verification = guardian.verify_election_partial_key_backup(backup.owner_id)
if verification is None:
raise Exception("Error verifying backup")
verifications.append(verification)
self._key_ceremony_service.append_verifications(
db, key_ceremony.id, verifications
)
# notify the admin that a new verification was created
self._key_ceremony_service.notify_changed(db, key_ceremony.id)

10
src/electionguard_gui/services/key_ceremony_stages/key_ceremony_stage_base.py
Просмотреть файл

@ -2,9 +2,11 @@ from abc import ABC
from pymongo.database import Database
from electionguard_gui.models.key_ceremony_dto import KeyCeremonyDto
from electionguard_gui.models.key_ceremony_states import KeyCeremonyStates
from electionguard_gui.services.authorization_service import AuthorizationService
from electionguard_gui.services.db_service import DbService
from electionguard_gui.services.eel_log_service import EelLogService
from electionguard_gui.services.guardian_service import GuardianService
from electionguard_gui.services.key_ceremony_service import KeyCeremonyService
from electionguard_gui.services.key_ceremony_state_service import (
KeyCeremonyStateService,
@ -19,6 +21,7 @@ class KeyCeremonyStageBase(ABC):
_key_ceremony_service: KeyCeremonyService
_auth_service: AuthorizationService
_key_ceremony_state_service: KeyCeremonyStateService
_guardian_service: GuardianService
def __init__(
self,
@ -27,12 +30,19 @@ class KeyCeremonyStageBase(ABC):
key_ceremony_service: KeyCeremonyService,
auth_service: AuthorizationService,
key_ceremony_state_service: KeyCeremonyStateService,
guardian_service: GuardianService,
):
self._db_service = db_service
self._key_ceremony_service = key_ceremony_service
self._auth_service = auth_service
self._key_ceremony_state_service = key_ceremony_state_service
self.log = log_service
self._guardian_service = guardian_service
def should_run(
self, key_ceremony: KeyCeremonyDto, state: KeyCeremonyStates
) -> bool:
pass
def run(self, db: Database, key_ceremony: KeyCeremonyDto) -> None:
pass

9
src/electionguard_gui/services/key_ceremony_state_service.py
Просмотреть файл

@ -19,11 +19,14 @@ class KeyCeremonyStateService(ServiceBase):
backups = len(key_ceremony.backups)
shared_backups = len(key_ceremony.shared_backups)
expected_backups = pow(guardian_count, 2)
verifications = len(key_ceremony.verifications)
expected_verifications = pow(guardian_count, 2) - guardian_count
self.log.debug(
f"guardians: {guardians_joined}/{guardian_count}; "
+ f"other_keys: {other_keys}/{guardian_count}; "
+ f"backups: {backups}/{expected_backups}; "
+ f"shared_backups: {shared_backups}/{guardian_count}"
+ f"shared_backups: {shared_backups}/{guardian_count}; "
+ f"verifications: {verifications}/{expected_verifications}"
)
if guardians_joined < guardian_count:
return KeyCeremonyStates.PendingGuardiansJoin
@ -33,7 +36,9 @@ class KeyCeremonyStateService(ServiceBase):
return KeyCeremonyStates.PendingGuardianBackups
if shared_backups == 0:
return KeyCeremonyStates.PendingAdminToShareBackups
return KeyCeremonyStates.PendingGuardiansVerifyBackups
if verifications < expected_verifications:
return KeyCeremonyStates.PendingGuardiansVerifyBackups
return KeyCeremonyStates.PendingAdminToPublishJointKey
status_descriptions = {