From e7f3b1d7bf84fdbeb5c091457651cc8dead47c18 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Mon, 20 Jul 2026 17:33:56 +0200 Subject: [PATCH] fix(poll): enforce actor-bound access and persistence --- src/govoplan_poll/backend/router.py | 157 ++++++++++++-- src/govoplan_poll/backend/service.py | 147 +++++++++++++- tests/test_authorization.py | 294 +++++++++++++++++++++++++++ 3 files changed, 579 insertions(+), 19 deletions(-) create mode 100644 tests/test_authorization.py diff --git a/src/govoplan_poll/backend/router.py b/src/govoplan_poll/backend/router.py index 797f5ec..1b537cc 100644 --- a/src/govoplan_poll/backend/router.py +++ b/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 diff --git a/src/govoplan_poll/backend/service.py b/src/govoplan_poll/backend/service.py index d7747cc..e180178 100644 --- a/src/govoplan_poll/backend/service.py +++ b/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, diff --git a/tests/test_authorization.py b/tests/test_authorization.py new file mode 100644 index 0000000..59eea82 --- /dev/null +++ b/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()