118 lines
3.3 KiB
Python
118 lines
3.3 KiB
Python
"""
|
|
Routes pour le vote et les bulletins.
|
|
"""
|
|
|
|
from fastapi import APIRouter, HTTPException, status, Depends, Request
|
|
from sqlalchemy.orm import Session
|
|
import base64
|
|
from .. import schemas, services
|
|
from ..dependencies import get_db, get_current_voter
|
|
from ..models import Voter
|
|
from ..crypto.hashing import SecureHash
|
|
|
|
router = APIRouter(prefix="/api/votes", tags=["votes"])
|
|
|
|
|
|
@router.post("/submit", response_model=schemas.VoteResponse)
|
|
async def submit_vote(
|
|
vote_bulletin: schemas.VoteBulletin,
|
|
current_voter: Voter = Depends(get_current_voter),
|
|
db: Session = Depends(get_db),
|
|
request: Request = None
|
|
):
|
|
"""
|
|
Soumettre un vote chiffré.
|
|
|
|
Le vote doit être:
|
|
- Chiffré avec ElGamal
|
|
- Accompagné d'une preuve ZK de validité
|
|
"""
|
|
|
|
# Vérifier que l'électeur n'a pas déjà voté
|
|
if services.VoteService.has_voter_voted(
|
|
db,
|
|
current_voter.id,
|
|
vote_bulletin.election_id
|
|
):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Voter has already voted in this election"
|
|
)
|
|
|
|
# Vérifier que l'élection existe
|
|
election = services.ElectionService.get_election(
|
|
db,
|
|
vote_bulletin.election_id
|
|
)
|
|
if not election:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Election not found"
|
|
)
|
|
|
|
# Vérifier que le candidat existe
|
|
from ..models import Candidate
|
|
candidate = db.query(Candidate).filter(
|
|
Candidate.id == vote_bulletin.candidate_id,
|
|
Candidate.election_id == vote_bulletin.election_id
|
|
).first()
|
|
if not candidate:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Candidate not found"
|
|
)
|
|
|
|
# Décoder le vote chiffré
|
|
try:
|
|
encrypted_vote_bytes = base64.b64decode(vote_bulletin.encrypted_vote)
|
|
except Exception:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Invalid encrypted vote format"
|
|
)
|
|
|
|
# Générer le hash du bulletin
|
|
import time
|
|
ballot_hash = SecureHash.hash_bulletin(
|
|
vote_id=current_voter.id,
|
|
candidate_id=vote_bulletin.candidate_id,
|
|
timestamp=int(time.time())
|
|
)
|
|
|
|
# Enregistrer le vote
|
|
vote = services.VoteService.record_vote(
|
|
db=db,
|
|
voter_id=current_voter.id,
|
|
election_id=vote_bulletin.election_id,
|
|
candidate_id=vote_bulletin.candidate_id,
|
|
encrypted_vote=encrypted_vote_bytes,
|
|
ballot_hash=ballot_hash,
|
|
ip_address=request.client.host if request else None
|
|
)
|
|
|
|
# Marquer l'électeur comme ayant voté
|
|
services.VoterService.mark_as_voted(db, current_voter.id)
|
|
|
|
return schemas.VoteResponse(
|
|
id=vote.id,
|
|
ballot_hash=ballot_hash,
|
|
timestamp=vote.timestamp
|
|
)
|
|
|
|
|
|
@router.get("/status")
|
|
def get_vote_status(
|
|
election_id: int,
|
|
current_voter: Voter = Depends(get_current_voter),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""Vérifier si l'électeur a déjà voté pour une élection"""
|
|
|
|
has_voted = services.VoteService.has_voter_voted(
|
|
db,
|
|
current_voter.id,
|
|
election_id
|
|
)
|
|
|
|
return {"has_voted": has_voted}
|