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

src/govoplan_poll/backend/router.py

@@ -5,7 +5,7 @@ from sqlalchemy.orm import Session
 
 from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
 from govoplan_core.db.session import get_session
-from govoplan_poll.backend.manifest import READ_SCOPE, RESPOND_SCOPE, WRITE_SCOPE
+from govoplan_poll.backend.manifest import ADMIN_SCOPE, READ_SCOPE, RESPOND_SCOPE, WRITE_SCOPE
 from govoplan_poll.backend.schemas import (
     PollCreateRequest,
     PollDecisionRequest,
@@ -28,15 +28,16 @@ from govoplan_poll.backend.service import (
     create_poll,
     decide_poll,
     get_poll_by_invitation_token,
-    get_poll,
+    get_visible_poll,
     list_poll_responses,
     list_poll_invitations,
-    list_polls,
+    list_visible_polls,
     open_poll,
     poll_invitation_response,
     poll_response,
     poll_response_item,
     poll_result_summary_by_id,
+    require_visible_poll_results,
     revoke_poll_invitation,
     submit_poll_response,
     submit_poll_response_with_token,
@@ -57,6 +58,8 @@ def _poll_http_error(exc: PollError) -> HTTPException:
         return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
     if str(exc) == "Poll invitation not found":
         return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
+    if str(exc) == "Poll results are not visible":
+        return HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc))
     return HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc))
 
 
@@ -64,6 +67,38 @@ def _poll_response(poll) -> PollResponse:
     return PollResponse.model_validate(poll_response(poll))
 
 
+def _principal_actor_ids(principal: ApiPrincipal) -> tuple[str, ...]:
+    candidates = (
+        principal.account_id,
+        principal.membership_id,
+        getattr(principal.user, "id", None),
+        principal.identity_id,
+        principal.principal.service_account_id,
+    )
+    return tuple(dict.fromkeys(str(value) for value in candidates if value))
+
+
+def _can_manage_polls(principal: ApiPrincipal) -> bool:
+    return has_scope(principal, WRITE_SCOPE) or has_scope(principal, ADMIN_SCOPE)
+
+
+def _can_view_sensitive_poll_data(principal: ApiPrincipal, poll: object) -> bool:
+    return _can_manage_polls(principal) or getattr(poll, "created_by_user_id", None) in _principal_actor_ids(principal)
+
+
+def _require_sensitive_poll_data_access(principal: ApiPrincipal, poll: object) -> None:
+    if not _can_view_sensitive_poll_data(principal, poll):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Poll respondent and invitation data is restricted to poll organizers",
+        )
+
+
+def _require_sensitive_poll_data_scope(principal: ApiPrincipal) -> None:
+    if not (has_scope(principal, READ_SCOPE) or _can_manage_polls(principal)):
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"Missing scope: {READ_SCOPE}")
+
+
 @router.get("/polls", response_model=PollListResponse)
 def api_list_polls(
     status_filter: str | None = Query(default=None, alias="status"),
@@ -72,7 +107,14 @@ def api_list_polls(
     principal: ApiPrincipal = Depends(get_api_principal),
 ) -> PollListResponse:
     _require_scope(principal, READ_SCOPE)
-    polls = list_polls(session, tenant_id=principal.tenant_id, status=status_filter, kind=kind)
+    polls = list_visible_polls(
+        session,
+        tenant_id=principal.tenant_id,
+        actor_ids=_principal_actor_ids(principal),
+        can_manage=_can_manage_polls(principal),
+        status=status_filter,
+        kind=kind,
+    )
     return PollListResponse(polls=[_poll_response(poll) for poll in polls])
 
 
@@ -87,7 +129,9 @@ def api_create_poll(
         poll = create_poll(session, tenant_id=principal.tenant_id, user_id=principal.account_id, payload=payload)
     except PollError as exc:
         raise _poll_http_error(exc) from exc
-    return _poll_response(poll)
+    response = _poll_response(poll)
+    session.commit()
+    return response
 
 
 @router.get("/polls/{poll_id}", response_model=PollResponse)
@@ -98,7 +142,15 @@ def api_get_poll(
 ) -> PollResponse:
     _require_scope(principal, READ_SCOPE)
     try:
-        return _poll_response(get_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id))
+        return _poll_response(
+            get_visible_poll(
+                session,
+                tenant_id=principal.tenant_id,
+                poll_id=poll_id,
+                actor_ids=_principal_actor_ids(principal),
+                can_manage=_can_manage_polls(principal),
+            )
+        )
     except PollError as exc:
         raise _poll_http_error(exc) from exc
 
@@ -112,9 +164,12 @@ def api_update_poll(
 ) -> PollResponse:
     _require_scope(principal, WRITE_SCOPE)
     try:
-        return _poll_response(update_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id, payload=payload))
+        poll = update_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id, payload=payload)
     except PollError as exc:
         raise _poll_http_error(exc) from exc
+    response = _poll_response(poll)
+    session.commit()
+    return response
 
 
 @router.post("/polls/{poll_id}/open", response_model=PollStatusResponse)
@@ -128,7 +183,9 @@ def api_open_poll(
         poll = open_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id)
     except PollError as exc:
         raise _poll_http_error(exc) from exc
-    return PollStatusResponse(poll=_poll_response(poll))
+    response = PollStatusResponse(poll=_poll_response(poll))
+    session.commit()
+    return response
 
 
 @router.post("/polls/{poll_id}/close", response_model=PollStatusResponse)
@@ -142,7 +199,9 @@ def api_close_poll(
         poll = close_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id)
     except PollError as exc:
         raise _poll_http_error(exc) from exc
-    return PollStatusResponse(poll=_poll_response(poll))
+    response = PollStatusResponse(poll=_poll_response(poll))
+    session.commit()
+    return response
 
 
 @router.post("/polls/{poll_id}/decide", response_model=PollStatusResponse)
@@ -157,7 +216,9 @@ def api_decide_poll(
         poll = decide_poll(session, tenant_id=principal.tenant_id, poll_id=poll_id, payload=payload)
     except PollError as exc:
         raise _poll_http_error(exc) from exc
-    return PollStatusResponse(poll=_poll_response(poll))
+    response = PollStatusResponse(poll=_poll_response(poll))
+    session.commit()
+    return response
 
 
 @router.post("/polls/{poll_id}/responses", response_model=PollResponseItem, status_code=status.HTTP_201_CREATED)
@@ -169,10 +230,31 @@ def api_submit_poll_response(
 ) -> PollResponseItem:
     _require_scope(principal, RESPOND_SCOPE)
     try:
-        response = submit_poll_response(session, tenant_id=principal.tenant_id, poll_id=poll_id, payload=payload)
+        get_visible_poll(
+            session,
+            tenant_id=principal.tenant_id,
+            poll_id=poll_id,
+            actor_ids=_principal_actor_ids(principal),
+            can_manage=_can_manage_polls(principal),
+        )
+        actor_payload = payload.model_copy(
+            update={
+                "respondent_id": principal.account_id,
+                "respondent_label": principal.display_name or principal.email,
+                "metadata": {key: value for key, value in payload.metadata.items() if key != "invitation_id"},
+            }
+        )
+        response = submit_poll_response(
+            session,
+            tenant_id=principal.tenant_id,
+            poll_id=poll_id,
+            payload=actor_payload,
+        )
     except PollError as exc:
         raise _poll_http_error(exc) from exc
-    return PollResponseItem.model_validate(poll_response_item(response))
+    result = PollResponseItem.model_validate(poll_response_item(response))
+    session.commit()
+    return result
 
 
 @router.get("/polls/{poll_id}/responses", response_model=PollResponseListResponse)
@@ -181,8 +263,22 @@ def api_list_poll_responses(
     session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(get_api_principal),
 ) -> PollResponseListResponse:
-    _require_scope(principal, READ_SCOPE)
+    _require_sensitive_poll_data_scope(principal)
     try:
+        poll = get_visible_poll(
+            session,
+            tenant_id=principal.tenant_id,
+            poll_id=poll_id,
+            actor_ids=_principal_actor_ids(principal),
+            can_manage=_can_manage_polls(principal),
+        )
+        _require_sensitive_poll_data_access(principal, poll)
+        require_visible_poll_results(
+            session,
+            poll=poll,
+            actor_ids=_principal_actor_ids(principal),
+            can_manage=_can_manage_polls(principal),
+        )
         responses = list_poll_responses(session, tenant_id=principal.tenant_id, poll_id=poll_id)
     except PollError as exc:
         raise _poll_http_error(exc) from exc
@@ -197,6 +293,19 @@ def api_poll_summary(
 ) -> PollResultSummaryResponse:
     _require_scope(principal, READ_SCOPE)
     try:
+        poll = get_visible_poll(
+            session,
+            tenant_id=principal.tenant_id,
+            poll_id=poll_id,
+            actor_ids=_principal_actor_ids(principal),
+            can_manage=_can_manage_polls(principal),
+        )
+        require_visible_poll_results(
+            session,
+            poll=poll,
+            actor_ids=_principal_actor_ids(principal),
+            can_manage=_can_manage_polls(principal),
+        )
         return PollResultSummaryResponse.model_validate(poll_result_summary_by_id(session, tenant_id=principal.tenant_id, poll_id=poll_id))
     except PollError as exc:
         raise _poll_http_error(exc) from exc
@@ -214,7 +323,9 @@ def api_create_poll_invitation(
         invitation, token = create_poll_invitation(session, tenant_id=principal.tenant_id, poll_id=poll_id, payload=payload)
     except PollError as exc:
         raise _poll_http_error(exc) from exc
-    return PollInvitationResponse.model_validate(poll_invitation_response(invitation, token=token))
+    response = PollInvitationResponse.model_validate(poll_invitation_response(invitation, token=token))
+    session.commit()
+    return response
 
 
 @router.get("/polls/{poll_id}/invitations", response_model=PollInvitationListResponse)
@@ -223,8 +334,16 @@ def api_list_poll_invitations(
     session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(get_api_principal),
 ) -> PollInvitationListResponse:
-    _require_scope(principal, READ_SCOPE)
+    _require_sensitive_poll_data_scope(principal)
     try:
+        poll = get_visible_poll(
+            session,
+            tenant_id=principal.tenant_id,
+            poll_id=poll_id,
+            actor_ids=_principal_actor_ids(principal),
+            can_manage=_can_manage_polls(principal),
+        )
+        _require_sensitive_poll_data_access(principal, poll)
         invitations = list_poll_invitations(session, tenant_id=principal.tenant_id, poll_id=poll_id)
     except PollError as exc:
         raise _poll_http_error(exc) from exc
@@ -250,7 +369,9 @@ def api_revoke_poll_invitation(
         )
     except PollError as exc:
         raise _poll_http_error(exc) from exc
-    return PollInvitationResponse.model_validate(poll_invitation_response(invitation))
+    response = PollInvitationResponse.model_validate(poll_invitation_response(invitation))
+    session.commit()
+    return response
 
 
 @router.get("/public/{token}", response_model=PollResponse)
@@ -274,4 +395,6 @@ def api_submit_public_poll_response(
         response = submit_poll_response_with_token(session, token=token, payload=payload)
     except PollError as exc:
         raise _poll_http_error(exc) from exc
-    return PollResponseItem.model_validate(poll_response_item(response))
+    result = PollResponseItem.model_validate(poll_response_item(response))
+    session.commit()
+    return result

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,

tests/test_authorization.py

@@ -0,0 +1,294 @@
+from __future__ import annotations
+
+import unittest
+from types import SimpleNamespace
+
+from fastapi import HTTPException
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.db.base import Base
+from govoplan_poll.backend.db.models import Poll, PollInvitation, PollOption, PollResponse
+from govoplan_poll.backend.router import (
+    api_create_poll,
+    api_get_poll,
+    api_list_poll_invitations,
+    api_list_poll_responses,
+    api_list_polls,
+    api_poll_summary,
+    api_submit_poll_response,
+)
+from govoplan_poll.backend.schemas import (
+    PollAnswerInput,
+    PollCreateRequest,
+    PollInvitationCreateRequest,
+    PollSubmitResponseRequest,
+)
+from govoplan_poll.backend.service import close_poll, create_poll, create_poll_invitation, open_poll
+
+
+READ_SCOPE = "poll:poll:read"
+WRITE_SCOPE = "poll:poll:write"
+RESPOND_SCOPE = "poll:response:write"
+
+
+class PollAuthorizationTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(
+            self.engine,
+            tables=[Poll.__table__, PollOption.__table__, PollResponse.__table__, PollInvitation.__table__],
+        )
+        self.Session = sessionmaker(bind=self.engine)
+        self.session: Session = self.Session()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        Base.metadata.drop_all(
+            self.engine,
+            tables=[PollInvitation.__table__, PollResponse.__table__, PollOption.__table__, Poll.__table__],
+        )
+        self.engine.dispose()
+
+    @staticmethod
+    def _principal(
+        account_id: str,
+        *,
+        tenant_id: str = "tenant-1",
+        user_id: str | None = None,
+        scopes: set[str] | None = None,
+    ) -> ApiPrincipal:
+        membership_id = user_id or f"membership-{account_id}"
+        display_name = f"Actor {account_id}"
+        return ApiPrincipal(
+            principal=PrincipalRef(
+                account_id=account_id,
+                membership_id=membership_id,
+                tenant_id=tenant_id,
+                scopes=frozenset(scopes or {READ_SCOPE, RESPOND_SCOPE}),
+                display_name=display_name,
+            ),
+            account=SimpleNamespace(id=account_id),
+            user=SimpleNamespace(id=membership_id),
+        )
+
+    def _poll(
+        self,
+        *,
+        owner_id: str = "owner",
+        visibility: str = "tenant",
+        result_visibility: str = "after_close",
+        title: str = "Decision",
+    ) -> Poll:
+        poll = create_poll(
+            self.session,
+            tenant_id="tenant-1",
+            user_id=owner_id,
+            payload=PollCreateRequest(
+                title=title,
+                kind="yes_no",
+                visibility=visibility,
+                result_visibility=result_visibility,
+            ),
+        )
+        open_poll(self.session, tenant_id="tenant-1", poll_id=poll.id)
+        return poll
+
+    @staticmethod
+    def _answer(*, claimed_respondent: str = "victim") -> PollSubmitResponseRequest:
+        return PollSubmitResponseRequest(
+            respondent_id=claimed_respondent,
+            respondent_label="Claimed victim",
+            answers=[PollAnswerInput(option_key="yes")],
+        )
+
+    def test_authenticated_response_identity_is_bound_to_actor(self) -> None:
+        poll = self._poll(result_visibility="public")
+        first_actor = self._principal("account-1")
+        second_actor = self._principal("account-2")
+
+        first = api_submit_poll_response(
+            poll.id,
+            self._answer(),
+            session=self.session,
+            principal=first_actor,
+        )
+        second = api_submit_poll_response(
+            poll.id,
+            self._answer(),
+            session=self.session,
+            principal=second_actor,
+        )
+        updated_first = api_submit_poll_response(
+            poll.id,
+            PollSubmitResponseRequest(
+                respondent_id="account-2",
+                answers=[PollAnswerInput(option_key="no")],
+            ),
+            session=self.session,
+            principal=first_actor,
+        )
+
+        self.assertEqual(first.respondent_id, "account-1")
+        self.assertEqual(first.respondent_label, "Actor account-1")
+        self.assertEqual(second.respondent_id, "account-2")
+        self.assertNotEqual(first.id, second.id)
+        self.assertEqual(updated_first.id, first.id)
+        self.assertEqual(updated_first.answers[0]["option_key"], "no")
+
+        cross_tenant_actor = self._principal("account-1", tenant_id="tenant-2")
+        with self.assertRaises(HTTPException) as cross_tenant:
+            api_submit_poll_response(
+                poll.id,
+                self._answer(),
+                session=self.session,
+                principal=cross_tenant_actor,
+            )
+        self.assertEqual(cross_tenant.exception.status_code, 404)
+
+    def test_create_route_persists_after_request_session_closes(self) -> None:
+        created = api_create_poll(
+            PollCreateRequest(title="Persisted poll", kind="yes_no", visibility="tenant"),
+            session=self.session,
+            principal=self._principal("owner", scopes={WRITE_SCOPE}),
+        )
+
+        self.session.close()
+        self.session = self.Session()
+        persisted = self.session.query(Poll).filter(Poll.id == created.id).one_or_none()
+
+        self.assertIsNotNone(persisted)
+        self.assertEqual(persisted.title, "Persisted poll")
+
+    def test_authenticated_response_cannot_claim_an_invitation(self) -> None:
+        poll = self._poll(result_visibility="public")
+
+        response = api_submit_poll_response(
+            poll.id,
+            self._answer().model_copy(
+                update={"metadata": {"invitation_id": "forged-invitation", "client_note": "kept"}}
+            ),
+            session=self.session,
+            principal=self._principal("account-1"),
+        )
+
+        self.assertEqual(response.metadata, {"client_note": "kept"})
+
+    def test_private_poll_requires_ownership_management_or_active_assignment(self) -> None:
+        poll = self._poll(visibility="private")
+        reader = self._principal("reader")
+
+        listed = api_list_polls(
+            status_filter=None,
+            kind=None,
+            session=self.session,
+            principal=reader,
+        )
+        self.assertEqual(listed.polls, [])
+        with self.assertRaises(HTTPException) as hidden:
+            api_get_poll(poll.id, session=self.session, principal=reader)
+        self.assertEqual(hidden.exception.status_code, 404)
+
+        invitation, _token = create_poll_invitation(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            payload=PollInvitationCreateRequest(respondent_id=reader.membership_id),
+        )
+        visible = api_get_poll(poll.id, session=self.session, principal=reader)
+        self.assertEqual(visible.id, poll.id)
+
+        invitation.revoked_at = invitation.created_at
+        self.session.flush()
+        with self.assertRaises(HTTPException) as revoked:
+            api_get_poll(poll.id, session=self.session, principal=reader)
+        self.assertEqual(revoked.exception.status_code, 404)
+
+        manager = self._principal("manager", scopes={READ_SCOPE, WRITE_SCOPE})
+        self.assertEqual(api_get_poll(poll.id, session=self.session, principal=manager).id, poll.id)
+
+        other_tenant = self._principal("owner", tenant_id="tenant-2", scopes={READ_SCOPE, WRITE_SCOPE})
+        with self.assertRaises(HTTPException) as cross_tenant:
+            api_get_poll(poll.id, session=self.session, principal=other_tenant)
+        self.assertEqual(cross_tenant.exception.status_code, 404)
+
+    def test_after_response_results_do_not_leak_to_other_actors(self) -> None:
+        poll = self._poll(result_visibility="after_response")
+        respondent = self._principal("respondent")
+        bystander = self._principal("bystander")
+
+        with self.assertRaises(HTTPException) as before_response:
+            api_poll_summary(poll.id, session=self.session, principal=respondent)
+        self.assertEqual(before_response.exception.status_code, 403)
+
+        api_submit_poll_response(poll.id, self._answer(), session=self.session, principal=respondent)
+        self.assertEqual(api_poll_summary(poll.id, session=self.session, principal=respondent).response_count, 1)
+        with self.assertRaises(HTTPException) as own_raw_responses:
+            api_list_poll_responses(poll.id, session=self.session, principal=respondent)
+        self.assertEqual(own_raw_responses.exception.status_code, 403)
+
+        with self.assertRaises(HTTPException) as other_actor_summary:
+            api_poll_summary(poll.id, session=self.session, principal=bystander)
+        self.assertEqual(other_actor_summary.exception.status_code, 403)
+        with self.assertRaises(HTTPException) as other_actor_responses:
+            api_list_poll_responses(poll.id, session=self.session, principal=bystander)
+        self.assertEqual(other_actor_responses.exception.status_code, 403)
+
+    def test_participants_can_see_aggregate_but_not_raw_responses_or_invitation_roster(self) -> None:
+        poll = self._poll(result_visibility="public")
+        first_actor = self._principal("account-1")
+        second_actor = self._principal("account-2")
+        create_poll_invitation(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            payload=PollInvitationCreateRequest(respondent_id=first_actor.membership_id, email="one@example.test"),
+        )
+        create_poll_invitation(
+            self.session,
+            tenant_id="tenant-1",
+            poll_id=poll.id,
+            payload=PollInvitationCreateRequest(respondent_id=second_actor.membership_id, email="two@example.test"),
+        )
+        api_submit_poll_response(poll.id, self._answer(), session=self.session, principal=first_actor)
+        api_submit_poll_response(poll.id, self._answer(), session=self.session, principal=second_actor)
+
+        self.assertEqual(api_poll_summary(poll.id, session=self.session, principal=first_actor).response_count, 2)
+        with self.assertRaises(HTTPException) as raw_responses:
+            api_list_poll_responses(poll.id, session=self.session, principal=first_actor)
+        self.assertEqual(raw_responses.exception.status_code, 403)
+        with self.assertRaises(HTTPException) as invitation_roster:
+            api_list_poll_invitations(poll.id, session=self.session, principal=first_actor)
+        self.assertEqual(invitation_roster.exception.status_code, 403)
+
+        organizer = self._principal("owner", scopes={READ_SCOPE})
+        self.assertEqual(len(api_list_poll_responses(poll.id, session=self.session, principal=organizer).responses), 2)
+        self.assertEqual(len(api_list_poll_invitations(poll.id, session=self.session, principal=organizer).invitations), 2)
+        manager = self._principal("manager", scopes={WRITE_SCOPE})
+        self.assertEqual(len(api_list_poll_responses(poll.id, session=self.session, principal=manager).responses), 2)
+        self.assertEqual(len(api_list_poll_invitations(poll.id, session=self.session, principal=manager).invitations), 2)
+
+    def test_after_close_and_organizer_result_visibility_are_enforced(self) -> None:
+        after_close = self._poll(result_visibility="after_close", title="After close")
+        reader = self._principal("reader")
+
+        with self.assertRaises(HTTPException) as still_open:
+            api_poll_summary(after_close.id, session=self.session, principal=reader)
+        self.assertEqual(still_open.exception.status_code, 403)
+        close_poll(self.session, tenant_id="tenant-1", poll_id=after_close.id)
+        self.assertEqual(api_poll_summary(after_close.id, session=self.session, principal=reader).response_count, 0)
+
+        organizer_only = self._poll(result_visibility="organizer", title="Organizer only")
+        api_submit_poll_response(organizer_only.id, self._answer(), session=self.session, principal=reader)
+        with self.assertRaises(HTTPException) as participant_denied:
+            api_poll_summary(organizer_only.id, session=self.session, principal=reader)
+        self.assertEqual(participant_denied.exception.status_code, 403)
+
+        owner = self._principal("owner", scopes={READ_SCOPE})
+        self.assertEqual(api_poll_summary(organizer_only.id, session=self.session, principal=owner).response_count, 1)
+
+
+if __name__ == "__main__":
+    unittest.main()