fix(poll): enforce actor-bound access and persistence

src/govoplan_poll/backend/service.py

@@ -211,6 +211,126 @@ def get_poll(session: Session, *, tenant_id: str, poll_id: str) -> Poll:
     return poll
 
 
+def _actor_ids(values: tuple[str, ...]) -> tuple[str, ...]:
+    return tuple(dict.fromkeys(value for value in values if value))
+
+
+def _has_active_invitation(session: Session, *, poll: Poll, actor_ids: tuple[str, ...]) -> bool:
+    ids = _actor_ids(actor_ids)
+    if not ids:
+        return False
+    invitations = (
+        session.query(PollInvitation)
+        .filter(
+            PollInvitation.tenant_id == poll.tenant_id,
+            PollInvitation.poll_id == poll.id,
+            PollInvitation.respondent_id.in_(ids),
+            PollInvitation.revoked_at.is_(None),
+        )
+        .all()
+    )
+    now = _now()
+    return any(
+        invitation.expires_at is None or response_datetime(invitation.expires_at) > now
+        for invitation in invitations
+    )
+
+
+def poll_is_visible(
+    session: Session,
+    *,
+    poll: Poll,
+    actor_ids: tuple[str, ...],
+    can_manage: bool = False,
+) -> bool:
+    ids = _actor_ids(actor_ids)
+    if can_manage or (poll.created_by_user_id is not None and poll.created_by_user_id in ids):
+        return True
+    if poll.visibility in {"tenant", "public"}:
+        return True
+    if poll.visibility in {"private", "unlisted"}:
+        return _has_active_invitation(session, poll=poll, actor_ids=ids)
+    return False
+
+
+def list_visible_polls(
+    session: Session,
+    *,
+    tenant_id: str,
+    actor_ids: tuple[str, ...],
+    can_manage: bool = False,
+    status: str | None = None,
+    kind: str | None = None,
+) -> list[Poll]:
+    return [
+        poll
+        for poll in list_polls(session, tenant_id=tenant_id, status=status, kind=kind)
+        if poll_is_visible(session, poll=poll, actor_ids=actor_ids, can_manage=can_manage)
+    ]
+
+
+def get_visible_poll(
+    session: Session,
+    *,
+    tenant_id: str,
+    poll_id: str,
+    actor_ids: tuple[str, ...],
+    can_manage: bool = False,
+) -> Poll:
+    poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
+    if not poll_is_visible(session, poll=poll, actor_ids=actor_ids, can_manage=can_manage):
+        # Do not disclose the existence of a private or unlisted poll.
+        raise PollError("Poll not found")
+    return poll
+
+
+def poll_results_are_visible(
+    session: Session,
+    *,
+    poll: Poll,
+    actor_ids: tuple[str, ...],
+    can_manage: bool = False,
+) -> bool:
+    ids = _actor_ids(actor_ids)
+    if can_manage or (poll.created_by_user_id is not None and poll.created_by_user_id in ids):
+        return True
+    if poll.result_visibility == "public":
+        return True
+    if poll.result_visibility == "after_close":
+        if poll.status in {"closed", "decided", "archived"}:
+            return True
+        return poll.closes_at is not None and response_datetime(poll.closes_at) <= _now()
+    if poll.result_visibility == "after_response" and ids:
+        return (
+            session.query(PollResponse)
+            .filter(
+                PollResponse.tenant_id == poll.tenant_id,
+                PollResponse.poll_id == poll.id,
+                PollResponse.respondent_id.in_(ids),
+                PollResponse.deleted_at.is_(None),
+            )
+            .first()
+            is not None
+        )
+    return False
+
+
+def require_visible_poll_results(
+    session: Session,
+    *,
+    poll: Poll,
+    actor_ids: tuple[str, ...],
+    can_manage: bool = False,
+) -> None:
+    if not poll_results_are_visible(
+        session,
+        poll=poll,
+        actor_ids=actor_ids,
+        can_manage=can_manage,
+    ):
+        raise PollError("Poll results are not visible")
+
+
 def update_poll(session: Session, *, tenant_id: str, poll_id: str, payload: PollUpdateRequest) -> Poll:
     poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
     if poll.status in {"closed", "decided", "archived"}:
@@ -379,8 +499,31 @@ def _existing_response(session: Session, *, poll: Poll, respondent_id: str | Non
     )
 
 
+def _lock_poll_for_response(session: Session, *, tenant_id: str, poll_id: str) -> Poll:
+    """Serialize the read-or-create path for identified respondents."""
+
+    poll = (
+        session.query(Poll)
+        .filter(
+            Poll.tenant_id == tenant_id,
+            Poll.id == poll_id,
+            Poll.deleted_at.is_(None),
+        )
+        .populate_existing()
+        .with_for_update()
+        .first()
+    )
+    if poll is None:
+        raise PollError("Poll not found")
+    return poll
+
+
 def submit_poll_response(session: Session, *, tenant_id: str, poll_id: str, payload: PollSubmitResponseRequest) -> PollResponse:
-    poll = get_poll(session, tenant_id=tenant_id, poll_id=poll_id)
+    poll = _lock_poll_for_response(
+        session,
+        tenant_id=tenant_id,
+        poll_id=poll_id,
+    )
     _assert_poll_accepts_responses(poll)
     if payload.respondent_id is None and not poll.allow_anonymous:
         raise PollError("Anonymous responses are not allowed for this poll")
@@ -561,7 +704,7 @@ def get_poll_by_invitation_token(session: Session, *, token: str) -> Poll:
 def submit_poll_response_with_token(session: Session, *, token: str, payload: PollSubmitResponseRequest) -> PollResponse:
     invitation = get_poll_invitation_by_token(session, token=token)
     metadata = dict(payload.metadata)
-    metadata.setdefault("invitation_id", invitation.id)
+    metadata["invitation_id"] = invitation.id
     response_payload = PollSubmitResponseRequest(
         respondent_id=invitation.respondent_id or f"invitation:{invitation.id}",
         respondent_label=payload.respondent_label or invitation.respondent_label or invitation.email,