Fixed: Generic errors.ForbiddenError raised when election closed

app/crud.py

@@ -46,7 +46,8 @@ def get_progress(db: Session, election_ref: str, token: str) -> schemas.Progress
         raise errors.UnauthorizedError("Wrong election ref")
 
     # Check we can update the election
-    if not payload["admin"]:
+    # Use .get() for a safe check. If "admin" key is missing, it returns None (which is falsy).
+    if not payload.get("admin"):
         raise errors.ForbiddenError("You are not allowed to manage the election")
 
     # Votes are provided for each candidate and each voter
@@ -284,11 +285,12 @@ def update_election(
     election_ref = payload["election"]
 
     # Check we can update the election
-    if not payload["admin"]:
+    # Use .get() for a safe check. If "admin" key is missing, it returns None (which is falsy).
+    if not payload.get("admin"):
         raise errors.ForbiddenError("You are not allowed to manage the election")
 
     if election_ref != election.ref:
-        raise errors.ForbiddenError("Wrong election ref")
+        raise errors.WrongElectionError("The provided admin token does not match this election.")
 
     db_election = get_election(db, election_ref)
 
@@ -297,12 +299,12 @@ def update_election(
 
     if election.date_start is not None and election.date_end is None and db_election.date_end is not None:
         if schemas.parse_date(election.date_start) > schemas.parse_date(db_election.date_end):
-            raise errors.ForbiddenError(
+            raise errors.InvalidDateError(
                 "The start date must be before the end date of the election"
             )
     elif election.date_end is not None and election.date_start is None:
         if schemas.parse_date(election.date_end) < schemas.parse_date(db_election.date_start):
-            raise errors.ForbiddenError(
+            raise errors.InvalidDateError(
                 "The end date must be after the start date of the election"
             )
 
@@ -332,20 +334,30 @@ def update_election(
         candidate_ids = {c.id for c in election.candidates}
         db_candidate_ids = {c.id for c in db_election.candidates}
         if candidate_ids != db_candidate_ids:
-            raise errors.ForbiddenError("You must have the same candidate ids")
+            raise errors.ImmutableIdsError("The set of candidate IDs cannot be changed during an update.")
 
     if election.grades is not None:
         grade_ids = {c.id for c in election.grades}
         db_grade_ids = {c.id for c in db_election.grades}
         if grade_ids != db_grade_ids:
-            raise errors.ForbiddenError("You must have the same grade ids")
+            raise errors.ImmutableIdsError("The set of grade IDs cannot be changed during an update.")
 
     # Update the candidates and grades
     if election.candidates is not None:
         update_candidates(db, election.candidates, db_election.candidates)
     if election.grades is not None:
         update_grades(db, election.grades, db_election.grades)
 
+    # Check if start_date is being changed
+    if election.date_start is not None and str(db_election.date_start) != election.date_start:
+        # If so, check if any votes have been cast
+        num_votes_cast = db.query(models.Vote).filter(
+            models.Vote.election_ref == election_ref,
+            models.Vote.grade_id.is_not(None)
+        ).count()
+        if num_votes_cast > 0:
+            raise errors.ElectionIsActiveError("Cannot change the start date of an election that already has votes.")
+
     for key in [
         "name",
         "description",
@@ -381,7 +393,7 @@ def _check_ballot_is_consistent(
         for c in election.candidates
     }
     if not all(len(votes) == 1 for votes in votes_by_candidate.values()):
-        raise errors.ForbiddenError("Unconsistent ballot")
+        raise errors.InconsistentBallotError("Inconsistent ballot: each candidate must have exactly one vote.")
 
 
 def create_ballot(db: Session, ballot: schemas.BallotCreate) -> schemas.BallotGet:
@@ -426,7 +438,7 @@ def _check_public_election(db: Session, election_ref: str):
     if db_election is None:
         raise errors.NotFoundError("elections")
     if db_election.restricted:
-        raise errors.ForbiddenError(
+        raise errors.ElectionRestrictedError(
             "The election is restricted. You can not create new votes"
         )
     return db_election
@@ -437,17 +449,17 @@ def _check_election_is_started(election: models.Election):
     If it is not, raise an error.
     """
     if election.date_start is not None and election.date_start > datetime.now():
-        raise errors.ForbiddenError("The election has not started yet. You can not create votes")
+        raise errors.ElectionNotStartedError("The election has not started yet. You can not create votes")
 
 def _check_election_is_not_ended(election: models.Election):
     """
     Check that the election is not ended.
     If it is, raise an error.
     """
     if election.date_end is not None and election.date_end < datetime.now():
-        raise errors.ForbiddenError("The election has ended. You can not create new votes")
+        raise errors.ElectionFinishedError("The election has ended. You can not create new votes")
     if election.force_close:
-        raise errors.ForbiddenError("The election is closed. You can not create or update votes")    
+        raise errors.ElectionFinishedError("The election is closed. You can not create or update votes")
 
 def _check_items_in_election(
     db: Session,
@@ -566,7 +578,7 @@ def get_results(db: Session, election_ref: str, token: t.Optional[str]) -> schem
         and (db_election.date_end is not None and db_election.date_end > datetime.now())
         and not db_election.force_close
     ):
-        raise errors.ForbiddenError("The election is not closed")
+        raise errors.ResultsHiddenError("Results are hidden until the election is closed.")
 
     query = db.query(
         models.Vote.candidate_id, models.Grade.value, func.count(models.Vote.id)

app/errors.py

@@ -2,54 +2,100 @@
 Utility to handle exceptions
 """
 
-
-class NotFoundError(Exception):
-    """
-    An item can not be found
-    """
-
-    def __init__(self, name: str):
-        self.name = name
-
-
-class InconsistentDatabaseError(Exception):
-    """
-    An inconsistent value was detected on the database
-    """
-
-    def __init__(self, name: str, details: str | None = None):
-        self.name = name
-        self.details = details
-
-
-class BadRequestError(Exception):
-    """
-    The request is made inconsistent
-    """
-
-    def __init__(self, details: str):
-        self.details = details
-
-
-class ForbiddenError(Exception):
+class CustomError(Exception):
     """
-    The request is made inconsistent
+    Base class for custom application errors.
     """
+    status_code: int = 500
+    error_code: str = "UNEXPECTED_ERROR"
+    message: str = "An unexpected error occurred."
 
-    def __init__(self, details: str = "Forbidden"):
-        self.details = details
+    def __init__(self, message: str | None = None, status_code: int | None = None, error_code: str | None = None):
+        super().__init__(message or self.message)
+        if status_code is not None:
+            self.status_code = status_code
+        if error_code is not None:
+            self.error_code = error_code
 
 
-class UnauthorizedError(Exception):
-    """
-    The verification could not be verified
-    """
+class NotFoundError(CustomError):
+    status_code = 404
+    error_code = "NOT_FOUND"
+    message = "The requested item could not be found."
 
     def __init__(self, name: str):
-        self.name = name
+        super().__init__(message=f"Oops! No {name} were found.")
 
+class InconsistentDatabaseError(CustomError):
+    status_code = 500
+    error_code = "INCONSISTENT_DATABASE"
+    message = "A serious internal error has occurred."
+
+    def __init__(self, name: str, details: str | None = None):
+        super().__init__(message=f"A serious error has occurred with {name}. {details or ''}")
+
+class BadRequestError(CustomError):
+    status_code = 400
+    error_code = "BAD_REQUEST"
+    message = "The request is invalid."
+
+class ForbiddenError(CustomError):
+    status_code = 403
+    error_code = "FORBIDDEN"
+    message = "You are not authorized to perform this action."
+
+class UnauthorizedError(CustomError):
+    status_code = 401
+    error_code = "UNAUTHORIZED"
+    message = "Authentication is required and has failed or has not yet been provided."
+
+class NoRecordedVotes(CustomError):
+    status_code = 403
+    error_code = "NO_RECORDED_VOTES"
+    message = "No votes have been recorded yet."
+
+class ElectionFinishedError(CustomError):
+    status_code = 403
+    error_code = "ELECTION_FINISHED"
+    message = "The election has finished and cannot be voted on."
+
+class InvalidDateError(CustomError):
+    status_code = 409
+    error_code = "INVALID_DATE_CONFIGURATION"
+    message = "The provided date configuration is invalid."
+
+class ElectionNotStartedError(CustomError):
+    status_code = 403
+    error_code = "ELECTION_NOT_STARTED"
+    message = "The election has not started yet."
+
+class ElectionRestrictedError(CustomError):
+    status_code = 403
+    error_code = "ELECTION_RESTRICTED"
+    message = "This is a restricted election."
+
+class InconsistentBallotError(CustomError):
+    status_code = 403
+    error_code = "INCONSISTENT_BALLOT"
+    message = "This ballot is inconsistent."
+
+class ResultsHiddenError(CustomError):
+    status_code = 403
+    error_code = "RESULTS_HIDDEN"
+    message = "Results are hidden."
+
+class WrongElectionError(CustomError):
+    status_code = 403
+    error_code = "WRONG_ELECTION"
+    message = "Wrong election."
+
+class ImmutableIdsError(CustomError):
+    status_code = 403
+    error_code = "IMMUTABLE_IDS"
+    message = "The set of IDs is immutable."
+
+class ElectionIsActiveError(CustomError):
+    status_code = 403
+    error_code = "ELECTION_IS_ACTIVE"
+    message = "This election is already active and cannot be modified."
 
-class NoRecordedVotes(Exception):
-    """
-    We can't display results if we don't have resutls
-    """

app/main.py

@@ -3,8 +3,8 @@
 from fastapi import Depends, FastAPI, HTTPException, Request, Body, Header
 from fastapi.responses import JSONResponse, PlainTextResponse
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.exceptions import RequestValidationError
 from sqlalchemy.orm import Session
-from jose import jwe, jws
 from jose.exceptions import JWEError, JWSError
 
 from . import crud, models, schemas, errors
@@ -23,75 +23,36 @@
     allow_headers=["*"],
 )
 
-
-@app.get("/")
-async def main():
-    return {"message": "Hello World"}
-
-@app.exception_handler(schemas.ArgumentsSchemaError)
-async def invalid_schema_exception_handler(
-    request: Request, exc: schemas.ArgumentsSchemaError
-):
+@app.exception_handler(RequestValidationError)
+async def validation_exception_handler(request: Request, exc: RequestValidationError):
+    # This overrides FastAPI's default 422 validation error handler
+    # to produce our standardized error format.
     return JSONResponse(
         status_code=422,
-        content={
-            "message": f"Validation Error. {exc}",
-        },
+        content={"error": "VALIDATION_ERROR", "message": str(exc)},
     )
 
-@app.exception_handler(errors.NotFoundError)
-async def not_found_exception_handler(request: Request, exc: errors.NotFoundError):
-    return JSONResponse(
-        status_code=404,
-        content={"message": f"Oops! No {exc.name} were found."},
-    )
-
-
-@app.exception_handler(errors.UnauthorizedError)
-async def unauthorized_exception_handler(request: Request, exc: errors.NotFoundError):
-    return JSONResponse(
-        status_code=401, content={"message": "Unautorized", "details": exc.name}
-    )
-
-
-@app.exception_handler(errors.ForbiddenError)
-async def forbidden_exception_handler(request: Request, exc: errors.ForbiddenError):
-    return JSONResponse(
-        status_code=403,
-        content={"message": f"Forbidden", "details": exc.details},
-    )
-
-
-@app.exception_handler(errors.BadRequestError)
-async def bad_request_exception_handler(request: Request, exc: errors.BadRequestError):
-    return JSONResponse(
-        status_code=400,
-        content={"message": f"Bad Request", "details": exc.details},
-    )
+@app.get("/")
+async def main():
+    return {"message": "Hello World"}
 
 
-@app.exception_handler(errors.NoRecordedVotes)
-async def no_recorded_votes_exception_handler(
-    request: Request, exc: errors.NoRecordedVotes
-):
+@app.exception_handler(errors.CustomError)
+async def custom_error_exception_handler(request: Request, exc: errors.CustomError):
     return JSONResponse(
-        status_code=403,
-        content={"message": f"No votes have been recorded yet"},
+        status_code=exc.status_code,
+        content={"error": exc.error_code, "message": str(exc)},
     )
 
-
-@app.exception_handler(errors.InconsistentDatabaseError)
-async def inconsistent_database_exception_handler(
-    request: Request, exc: errors.InconsistentDatabaseError
+@app.exception_handler(schemas.ArgumentsSchemaError)
+async def invalid_schema_exception_handler(
+    request: Request, exc: schemas.ArgumentsSchemaError
 ):
     return JSONResponse(
-        status_code=500,
-        content={
-            "message": f"A serious error has occured with {exc.name}. {exc.details or ''}"
-        },
+        status_code=422,
+        content={"error": "SCHEMA_VALIDATION_ERROR", "message": str(exc)},
     )
 
-
 @app.get("/liveness")
 def read_root():
     return "OK"