Enforce tenant policy on public scheduling links

src/govoplan_scheduling/backend/manifest.py

@@ -25,8 +25,12 @@ from govoplan_core.core.people import (
     CAPABILITY_ACCESS_PEOPLE_SEARCH,
     CAPABILITY_ADDRESSES_PEOPLE_SEARCH,
 )
-from govoplan_core.core.poll import CAPABILITY_POLL_SCHEDULING
+from govoplan_core.core.poll import CAPABILITY_POLL_SCHEDULING, PollCapabilityError
-from govoplan_core.core.poll_participation import CAPABILITY_POLL_PARTICIPATION_GATEWAY
+from govoplan_core.core.poll_participation import (
+    CAPABILITY_POLL_PARTICIPATION_GATEWAY,
+    PollResponseGatewayRef,
+    poll_participation_gateway_provider,
+)
 from govoplan_core.core.policy import CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY
 from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
@@ -104,7 +108,7 @@ DOCUMENTATION = (
         body=(
             "Scheduling records participant requirements, quorum and weighting constraints, response deadlines, reminders, and yes/no/maybe availability through Poll. "
             "Calendar-aware organizers can inspect conflicts and create tentative holds before deciding. After a decision, Scheduling releases unused holds, creates or links the final event, and records notification handoff state. "
-            "Signed external links expose only the bounded request information allowed by the request's participation and privacy policy."
+            "Signed external links expose only the bounded request information allowed by the request's participation and privacy policy. Their governed Poll invitation resolves the tenant before Scheduling runs, so tenant module policy can withdraw public participation without weakening token validation."
         ),
         layer="configured",
         documentation_types=("user",),
@@ -212,6 +216,47 @@ def _scheduling_router(context: ModuleContext):
     return router
 
 
+def _public_tenant_resolver(request: object, session: object) -> str | None:
+    path_params = getattr(request, "path_params", {})
+    request_id = str(path_params.get("request_id") or "").strip()
+    token = str(path_params.get("token") or "").strip()
+    path = str(getattr(getattr(request, "url", None), "path", ""))
+    if not request_id or not token or "/scheduling/public/" not in path:
+        return None
+
+    from govoplan_scheduling.backend.db.models import SchedulingRequest
+
+    app = getattr(request, "app", None)
+    registry = getattr(getattr(app, "state", None), "govoplan_registry", None)
+    provider = poll_participation_gateway_provider(registry)
+    if provider is None:
+        return None
+    gateway = PollResponseGatewayRef(
+        module_id=MODULE_ID,
+        resource_type="scheduling_request",
+        resource_id=request_id,
+    )
+    try:
+        invitation = provider.resolve_public_invitation(
+            session,
+            token=token,
+            gateway=gateway,
+        )
+    except PollCapabilityError:
+        return None
+    scheduling_request = (
+        session.query(SchedulingRequest)
+        .filter(
+            SchedulingRequest.id == request_id,
+            SchedulingRequest.tenant_id == invitation.tenant_id,
+            SchedulingRequest.poll_id == invitation.poll_id,
+            SchedulingRequest.deleted_at.is_(None),
+        )
+        .one_or_none()
+    )
+    return scheduling_request.tenant_id if scheduling_request is not None else None
+
+
 manifest = ModuleManifest(
     id=MODULE_ID,
     name=MODULE_NAME,
@@ -279,6 +324,7 @@ manifest = ModuleManifest(
         ),
     ),
     route_factory=_scheduling_router,
+    public_tenant_resolver=_public_tenant_resolver,
     tenant_summary_providers=(_tenant_summary,),
     migration_spec=MigrationSpec(
         module_id=MODULE_ID,

tests/test_manifest.py

@@ -29,6 +29,7 @@ class SchedulingManifestTests(unittest.TestCase):
         self.assertIn("addresses.people_search", manifest.optional_capabilities)
         self.assertIn("evaluation", manifest.optional_dependencies)
         self.assertIsNotNone(manifest.route_factory)
+        self.assertIsNotNone(manifest.public_tenant_resolver)
         self.assertIsNotNone(manifest.migration_spec)
         self.assertIsNotNone(manifest.frontend)
         self.assertEqual(

tests/test_response_editing.py

@@ -33,7 +33,12 @@ from govoplan_scheduling.backend.db.models import (
     SchedulingParticipant,
     SchedulingRequest,
 )
-from govoplan_scheduling.backend.manifest import ADMIN_SCOPE, RESPOND_SCOPE, WRITE_SCOPE
+from govoplan_scheduling.backend.manifest import (
+    ADMIN_SCOPE,
+    RESPOND_SCOPE,
+    WRITE_SCOPE,
+    get_manifest as get_scheduling_manifest,
+)
 from govoplan_scheduling.backend.router import (
     api_get_my_scheduling_availability,
     api_submit_scheduling_availability,
@@ -78,6 +83,7 @@ class SchedulingResponseEditingTests(unittest.TestCase):
         registry.register(get_poll_manifest())
         registry.configure_capability_context(ModuleContext(registry=registry, settings=object()))
         configure_runtime(registry=registry)
+        self.registry = registry
         self.engine = create_engine("sqlite:///:memory:")
         Base.metadata.create_all(
             self.engine,
@@ -512,6 +518,26 @@ class SchedulingResponseEditingTests(unittest.TestCase):
         )
         self.assertEqual(tokens, {})
         token = self._issue_copy(public_request, public_request.participants[0])
+        resolver = get_scheduling_manifest().public_tenant_resolver
+        self.assertIsNotNone(resolver)
+        public_http_request = SimpleNamespace(
+            app=SimpleNamespace(
+                state=SimpleNamespace(govoplan_registry=self.registry)
+            ),
+            path_params={
+                "request_id": public_request.id,
+                "token": token,
+            },
+            url=SimpleNamespace(
+                path=(
+                    f"/api/v1/scheduling/public/{public_request.id}/{token}"
+                )
+            ),
+        )
+        self.assertEqual(
+            "tenant-1",
+            resolver(public_http_request, self.session),
+        )
         invitation = self.session.query(PollInvitation).filter(
             PollInvitation.id == public_request.participants[0].poll_invitation_id
         ).one()