From 43737c67dcd53b024c3ce43f678403a51decec04 Mon Sep 17 00:00:00 2001 From: Abhishek Kumar Date: Fri, 10 Jul 2026 16:52:43 +0530 Subject: [PATCH] fix: scope workflow run to org rather than user --- api/routes/agent_stream.py | 1 - api/routes/public_agent.py | 7 +- api/routes/telephony.py | 67 +++++++++-------- .../campaign/campaign_call_dispatcher.py | 3 +- api/services/pipecat/run_pipeline.py | 42 +++++++---- api/services/telephony/README.md | 2 +- api/services/telephony/ari_manager.py | 15 ++-- api/services/telephony/base.py | 14 ++-- .../telephony/providers/ari/provider.py | 11 +-- .../telephony/providers/cloudonix/provider.py | 16 +++-- .../telephony/providers/plivo/provider.py | 8 +-- .../telephony/providers/plivo/routes.py | 3 +- .../telephony/providers/telnyx/provider.py | 10 +-- .../telephony/providers/twilio/provider.py | 8 +-- .../telephony/providers/twilio/routes.py | 3 +- .../telephony/providers/vobiz/provider.py | 8 +-- .../telephony/providers/vobiz/routes.py | 6 +- .../telephony/providers/vonage/provider.py | 17 +++-- .../telephony/providers/vonage/routes.py | 10 +-- api/tests/telephony/cloudonix/test_routes.py | 2 - api/tests/telephony/plivo/test_routes.py | 4 +- api/tests/telephony/twilio/test_routes.py | 6 +- api/tests/test_active_calls.py | 2 +- api/tests/test_public_agent_routes.py | 3 +- api/tests/test_telephony_routes.py | 14 ++-- docs/integrations/telephony/asterisk-ari.mdx | 2 +- docs/integrations/telephony/custom.mdx | 14 +++- docs/integrations/telephony/webhooks.mdx | 71 ++++++++++++++----- ui/src/client/types.gen.ts | 7 +- 29 files changed, 218 insertions(+), 158 deletions(-) diff --git a/api/routes/agent_stream.py b/api/routes/agent_stream.py index f5de8f38..a14ed603 100644 --- a/api/routes/agent_stream.py +++ b/api/routes/agent_stream.py @@ -117,7 +117,6 @@ async def agent_stream_websocket( websocket, organization_id=workflow.organization_id, workflow_id=workflow.id, - user_id=workflow.user_id, workflow_run_id=workflow_run.id, params=params, ) diff --git a/api/routes/public_agent.py b/api/routes/public_agent.py index b1a338a3..2b4e95c9 100644 --- a/api/routes/public_agent.py +++ b/api/routes/public_agent.py @@ -303,13 +303,12 @@ async def _execute_resolved_target( webhook_url = ( f"{backend_endpoint}/api/v1/telephony/{webhook_endpoint}" f"?workflow_id={target.workflow.id}" - f"&user_id={execution_user_id}" f"&workflow_run_id={workflow_run.id}" f"&organization_id={target.organization_id}" ) - # 10. Initiate call via telephony provider. workflow_id and user_id are - # required by providers that build the media WebSocket URL at dial time + # 10. Initiate call via telephony provider. workflow_id and organization_id + # are required by providers that build the media WebSocket URL at dial time # (e.g. Telnyx, Cloudonix); without them the URL contains "None/None" and # the stream connection fails. try: @@ -318,7 +317,7 @@ async def _execute_resolved_target( webhook_url=webhook_url, workflow_run_id=workflow_run.id, workflow_id=target.workflow.id, - user_id=execution_user_id, + organization_id=target.organization_id, ) except Exception as e: logger.warning( diff --git a/api/routes/telephony.py b/api/routes/telephony.py index 8caa209f..915d3911 100644 --- a/api/routes/telephony.py +++ b/api/routes/telephony.py @@ -240,12 +240,14 @@ async def initiate_call( webhook_url = ( f"{backend_endpoint}/api/v1/telephony/{webhook_endpoint}" f"?workflow_id={workflow.id}" - f"&user_id={execution_user_id}" f"&workflow_run_id={workflow_run_id}" f"&organization_id={user.selected_organization_id}" ) - keywords = {"workflow_id": workflow.id, "user_id": execution_user_id} + keywords = { + "workflow_id": workflow.id, + "organization_id": user.selected_organization_id, + } # Initiate call via provider result = await provider.initiate_call( @@ -543,13 +545,13 @@ async def websocket_ari_endpoint(websocket: WebSocket): query params (appended by the v() dial string option in externalMedia). """ workflow_id = websocket.query_params.get("workflow_id") - user_id = websocket.query_params.get("user_id") + organization_id = websocket.query_params.get("organization_id") workflow_run_id = websocket.query_params.get("workflow_run_id") - if not workflow_id or not user_id or not workflow_run_id: + if not workflow_id or not organization_id or not workflow_run_id: logger.error( - f"ARI WebSocket missing query params: " - f"workflow_id={workflow_id}, user_id={user_id}, workflow_run_id={workflow_run_id}" + f"ARI WebSocket missing query params: workflow_id={workflow_id}, " + f"organization_id={organization_id}, workflow_run_id={workflow_run_id}" ) await websocket.close(code=4400, reason="Missing required query params") return @@ -559,39 +561,55 @@ async def websocket_ari_endpoint(websocket: WebSocket): await websocket.accept(subprotocol="media") await _handle_telephony_websocket( - websocket, int(workflow_id), int(user_id), int(workflow_run_id) + websocket, int(workflow_id), int(organization_id), int(workflow_run_id) ) -@router.websocket("/ws/{workflow_id}/{user_id}/{workflow_run_id}") +@router.websocket("/ws/{workflow_id}/{organization_id}/{workflow_run_id}") async def websocket_endpoint( - websocket: WebSocket, workflow_id: int, user_id: int, workflow_run_id: int + websocket: WebSocket, workflow_id: int, organization_id: int, workflow_run_id: int ): """WebSocket endpoint for real-time call handling - routes to provider-specific handlers.""" await websocket.accept() - await _handle_telephony_websocket(websocket, workflow_id, user_id, workflow_run_id) + await _handle_telephony_websocket( + websocket, workflow_id, organization_id, workflow_run_id + ) async def _handle_telephony_websocket( - websocket: WebSocket, workflow_id: int, user_id: int, workflow_run_id: int + websocket: WebSocket, workflow_id: int, organization_id: int, workflow_run_id: int ): - """Shared WebSocket handler logic (connection already accepted).""" + """Shared WebSocket handler logic (connection already accepted). + + TODO(security): ``organization_id`` arrives in the URL the provider dials + back, so it is caller-supplied and unauthenticated — this socket has no + signature check, and the id triple is a guessable bearer capability. + Scoping the lookups below by it prevents an accidental cross-org mismatch, + not a deliberate one. The real fix is a one-shot capability token minted at + run creation and redeemed here through an atomic + ``initialized -> running`` compare-and-swap, which would also close the + read-then-write race on the state check further down. + """ try: # Set the run context set_current_run_id(workflow_run_id) # Get workflow run to determine provider type - workflow_run = await db_client.get_workflow_run(workflow_run_id) + workflow_run = await db_client.get_workflow_run( + workflow_run_id, organization_id=organization_id + ) if not workflow_run: - logger.error(f"Workflow run {workflow_run_id} not found") + logger.error( + f"Workflow run {workflow_run_id} not found for org {organization_id}" + ) await websocket.close(code=4404, reason="Workflow run not found") return - # Get workflow for organization info. System lookup keyed only on the - # workflow_id (org is derived below) — use the explicit unscoped variant. - workflow = await db_client.get_workflow_by_id(workflow_id) + workflow = await db_client.get_workflow( + workflow_id, organization_id=organization_id + ) if not workflow: - logger.error(f"Workflow {workflow_id} not found") + logger.error(f"Workflow {workflow_id} not found for org {organization_id}") await websocket.close(code=4404, reason="Workflow not found") return if workflow_run.workflow_id != workflow.id: @@ -601,13 +619,6 @@ async def _handle_telephony_websocket( ) await websocket.close(code=4400, reason="workflow_run_workflow_mismatch") return - if workflow.user_id != user_id: - logger.error( - f"Telephony websocket user mismatch for workflow {workflow.id}: " - f"got {user_id}, expected {workflow.user_id}" - ) - await websocket.close(code=4400, reason="workflow_user_mismatch") - return # Check workflow run state - only allow 'initialized' state if workflow_run.state != WorkflowRunState.INITIALIZED.value: @@ -689,7 +700,7 @@ async def _handle_telephony_websocket( # Delegate to provider-specific handler await provider.handle_websocket( - websocket, workflow_id, user_id, workflow_run_id + websocket, workflow_id, organization_id, workflow_run_id ) except WebSocketDisconnect as e: @@ -854,7 +865,7 @@ async def handle_inbound_run(request: Request): backend_endpoint, wss_backend_endpoint = await get_backend_endpoints() websocket_url = ( f"{wss_backend_endpoint}/api/v1/telephony/ws/" - f"{workflow_id}/{user_id}/{workflow_run_id}" + f"{workflow_id}/{config.organization_id}/{workflow_run_id}" ) return await provider_instance.start_inbound_stream( @@ -1022,7 +1033,7 @@ async def handle_inbound_telephony( backend_endpoint, wss_backend_endpoint = await get_backend_endpoints() websocket_url = ( f"{wss_backend_endpoint}/api/v1/telephony/ws/" - f"{workflow_id}/{workflow_context['user_id']}/{workflow_run_id}" + f"{workflow_id}/{organization_id}/{workflow_run_id}" ) response = await provider_instance.start_inbound_stream( diff --git a/api/services/campaign/campaign_call_dispatcher.py b/api/services/campaign/campaign_call_dispatcher.py index b8ab1e79..61411f8d 100644 --- a/api/services/campaign/campaign_call_dispatcher.py +++ b/api/services/campaign/campaign_call_dispatcher.py @@ -354,7 +354,6 @@ class CampaignCallDispatcher: webhook_url = ( f"{backend_endpoint}/api/v1/telephony/{webhook_endpoint}" f"?workflow_id={campaign.workflow_id}" - f"&user_id={campaign.created_by}" f"&workflow_run_id={workflow_run.id}" f"&organization_id={campaign.organization_id}" ) @@ -365,7 +364,7 @@ class CampaignCallDispatcher: workflow_run_id=workflow_run.id, from_number=from_number, workflow_id=campaign.workflow_id, - user_id=campaign.created_by, + organization_id=campaign.organization_id, ) # Store provider type and metadata in gathered_context diff --git a/api/services/pipecat/run_pipeline.py b/api/services/pipecat/run_pipeline.py index 86852554..81aa53e4 100644 --- a/api/services/pipecat/run_pipeline.py +++ b/api/services/pipecat/run_pipeline.py @@ -254,7 +254,7 @@ async def run_pipeline_telephony( provider_name: str, workflow_id: int, workflow_run_id: int, - user_id: int, + organization_id: int, call_id: str, transport_kwargs: dict, ) -> None: @@ -268,7 +268,7 @@ async def run_pipeline_telephony( provider_name=provider_name, workflow_id=workflow_id, workflow_run_id=workflow_run_id, - user_id=user_id, + organization_id=organization_id, call_id=call_id, transport_kwargs=transport_kwargs, ) @@ -285,7 +285,7 @@ async def _run_pipeline_telephony_impl( provider_name: str, workflow_id: int, workflow_run_id: int, - user_id: int, + organization_id: int, call_id: str, transport_kwargs: dict, ) -> None: @@ -300,7 +300,9 @@ async def _run_pipeline_telephony_impl( provider_name: Stable identifier of the provider (registry key). workflow_id: Workflow being executed. workflow_run_id: Workflow run row. - user_id: Owner of the workflow. + organization_id: Tenant owning the workflow and the run. Every lookup + below is scoped by it; the workflow owner is read off the workflow + row and used only for attribution. call_id: Provider call identifier. transport_kwargs: Provider-specific kwargs forwarded to the transport factory (e.g. stream_sid + call_sid for Twilio). @@ -308,12 +310,15 @@ async def _run_pipeline_telephony_impl( logger.debug(f"Running {provider_name} pipeline for workflow_run {workflow_run_id}") set_current_run_id(workflow_run_id) - workflow = await db_client.get_workflow(workflow_id, user_id) - if workflow: - set_current_org_id(workflow.organization_id) + workflow = await db_client.get_workflow( + workflow_id, organization_id=organization_id + ) + if not workflow: + raise HTTPException(status_code=404, detail="Workflow not found") + set_current_org_id(workflow.organization_id) ambient_noise_config = None - if workflow and workflow.workflow_configurations: + if workflow.workflow_configurations: ambient_noise_config = workflow.workflow_configurations.get( "ambient_noise_configuration" ) @@ -322,9 +327,16 @@ async def _run_pipeline_telephony_impl( # (test call, campaign dispatch, inbound). Transports use it to load creds # from the right config row. Falls back to None for legacy runs (transports # then resolve the org's default config). - workflow_run = await db_client.get_workflow_run(workflow_run_id) + workflow_run = await db_client.get_workflow_run( + workflow_run_id, organization_id=organization_id + ) + if not workflow_run: + raise HTTPException(status_code=404, detail="Workflow run not found") + if workflow_run.workflow_id != workflow_id: + raise HTTPException(status_code=400, detail="workflow_run_workflow_mismatch") + telephony_configuration_id = None - if workflow_run and workflow_run.initial_context: + if workflow_run.initial_context: telephony_configuration_id = workflow_run.initial_context.get( "telephony_configuration_id" ) @@ -336,11 +348,9 @@ async def _run_pipeline_telephony_impl( get_effective_ai_model_configuration_for_workflow, ) - run_configs = ( - (workflow_run.definition.workflow_configurations or {}) if workflow_run else {} - ) + run_configs = workflow_run.definition.workflow_configurations or {} user_config = await get_effective_ai_model_configuration_for_workflow( - organization_id=workflow.organization_id if workflow else None, + organization_id=workflow.organization_id, workflow_configurations=run_configs, ) is_realtime = bool(user_config.is_realtime and user_config.realtime is not None) @@ -364,10 +374,12 @@ async def _run_pipeline_telephony_impl( transport, workflow_id, workflow_run_id, - user_id, + # Attribution only — scoping is driven by organization_id below. + workflow.user_id, audio_config=audio_config, workflow_run=workflow_run, resolved_user_config=user_config, + organization_id=organization_id, ) except Exception as e: logger.error( diff --git a/api/services/telephony/README.md b/api/services/telephony/README.md index 37c113be..714702e8 100644 --- a/api/services/telephony/README.md +++ b/api/services/telephony/README.md @@ -77,7 +77,7 @@ class TelephonyProvider(ABC): async def verify_webhook_signature(self, url: str, params: Dict[str, Any], signature: str) -> bool @abstractmethod - async def get_webhook_response(self, workflow_id: int, user_id: int, workflow_run_id: int) -> str + async def get_webhook_response(self, workflow_id: int, organization_id: int, workflow_run_id: int) -> str ``` ## Configuration Loading diff --git a/api/services/telephony/ari_manager.py b/api/services/telephony/ari_manager.py index cc6a2fae..cd2e93d3 100644 --- a/api/services/telephony/ari_manager.py +++ b/api/services/telephony/ari_manager.py @@ -353,19 +353,18 @@ class ARIConnection: workflow_run_id = args_dict.get("workflow_run_id") workflow_id = args_dict.get("workflow_id") - user_id = args_dict.get("user_id") - if not workflow_run_id or not workflow_id or not user_id: + if not workflow_run_id or not workflow_id: logger.warning( f"[ARI org={self.organization_id}] StasisStart missing required args: " - f"workflow_run_id={workflow_run_id}, workflow_id={workflow_id}, user_id={user_id}" + f"workflow_run_id={workflow_run_id}, workflow_id={workflow_id}" ) return # Start pipeline connection in background task asyncio.create_task( self._handle_stasis_start( - channel_id, channel_state, workflow_run_id, workflow_id, user_id + channel_id, channel_state, workflow_run_id, workflow_id ) ) @@ -448,7 +447,6 @@ class ARIConnection: async def _create_external_media( self, workflow_id: str, - user_id: str, workflow_run_id: str, channel_id: Optional[str] = None, ) -> str: @@ -464,10 +462,10 @@ class ARIConnection: the POST and avoid racing against the StasisStart event. """ # v() appends URI query params to the websocket_client.conf URL - # e.g. wss://api.dograh.com/ws/ari?workflow_id=1&user_id=2&workflow_run_id=3 + # e.g. wss://api.dograh.com/ws/ari?workflow_id=1&organization_id=2&workflow_run_id=3 transport_data = ( f"v(workflow_id={workflow_id}," - f"user_id={user_id}," + f"organization_id={self.organization_id}," f"workflow_run_id={workflow_run_id})" ) @@ -646,7 +644,6 @@ class ARIConnection: channel_state, str(workflow_run.id), str(inbound_workflow_id), - str(user_id), ) except Exception as e: if workflow_run: @@ -668,7 +665,6 @@ class ARIConnection: channel_state: str, workflow_run_id: str, workflow_id: str, - user_id: str, ): """Set up external media for a caller channel that has entered Stasis. @@ -712,7 +708,6 @@ class ARIConnection: # 3. Create the ext media channel with the id we just registered. created_id = await self._create_external_media( workflow_id, - user_id, workflow_run_id, channel_id=ext_channel_id, ) diff --git a/api/services/telephony/base.py b/api/services/telephony/base.py index 3824864b..9acecc32 100644 --- a/api/services/telephony/base.py +++ b/api/services/telephony/base.py @@ -150,14 +150,15 @@ class TelephonyProvider(ABC): @abstractmethod async def get_webhook_response( - self, workflow_id: int, user_id: int, workflow_run_id: int + self, workflow_id: int, organization_id: int, workflow_run_id: int ) -> str: """ Generate the initial webhook response for starting a call session. Args: workflow_id: The workflow ID - user_id: The user ID + organization_id: The organization owning the workflow; providers + embed it in the media websocket URL they hand back workflow_run_id: The workflow run ID Returns: @@ -223,7 +224,7 @@ class TelephonyProvider(ABC): self, websocket: "WebSocket", workflow_id: int, - user_id: int, + organization_id: int, workflow_run_id: int, ) -> None: """ @@ -232,10 +233,14 @@ class TelephonyProvider(ABC): This method encapsulates all provider-specific WebSocket handshake and message routing logic, keeping the main websocket endpoint clean. + ``organization_id`` is the tenant that every workflow/run lookup must be + scoped by. The workflow owner is deliberately not passed in — derive it + from the workflow row where it's needed for attribution. + Args: websocket: The WebSocket connection workflow_id: The workflow ID - user_id: The user ID + organization_id: The organization owning the workflow and run workflow_run_id: The workflow run ID """ pass @@ -355,7 +360,6 @@ class TelephonyProvider(ABC): *, organization_id: int, workflow_id: int, - user_id: int, workflow_run_id: int, params: Dict[str, str], ) -> None: diff --git a/api/services/telephony/providers/ari/provider.py b/api/services/telephony/providers/ari/provider.py index 8c811f9a..bd855178 100644 --- a/api/services/telephony/providers/ari/provider.py +++ b/api/services/telephony/providers/ari/provider.py @@ -99,7 +99,6 @@ class ARIProvider(TelephonyProvider): [ f"workflow_run_id={workflow_run_id}", f"workflow_id={kwargs.get('workflow_id', '')}", - f"user_id={kwargs.get('user_id', '')}", ], ) ), @@ -179,7 +178,7 @@ class ARIProvider(TelephonyProvider): return True async def get_webhook_response( - self, workflow_id: int, user_id: int, workflow_run_id: int + self, workflow_id: int, organization_id: int, workflow_run_id: int ) -> str: """ARI does not use webhook responses - call control is via REST API.""" logger.warning( @@ -241,7 +240,7 @@ class ARIProvider(TelephonyProvider): self, websocket: "WebSocket", workflow_id: int, - user_id: int, + organization_id: int, workflow_run_id: int, ) -> None: """ @@ -253,7 +252,9 @@ class ARIProvider(TelephonyProvider): from api.services.pipecat.run_pipeline import run_pipeline_telephony # Get channel_id from workflow run context - workflow_run = await db_client.get_workflow_run(workflow_run_id, user_id) + workflow_run = await db_client.get_workflow_run( + workflow_run_id, organization_id=organization_id + ) channel_id = "" if workflow_run and workflow_run.gathered_context: channel_id = workflow_run.gathered_context.get("call_id", "") @@ -267,7 +268,7 @@ class ARIProvider(TelephonyProvider): provider_name=self.PROVIDER_NAME, workflow_id=workflow_id, workflow_run_id=workflow_run_id, - user_id=user_id, + organization_id=organization_id, call_id=channel_id, transport_kwargs={"channel_id": channel_id}, ) diff --git a/api/services/telephony/providers/cloudonix/provider.py b/api/services/telephony/providers/cloudonix/provider.py index 99b7d380..d3e97466 100644 --- a/api/services/telephony/providers/cloudonix/provider.py +++ b/api/services/telephony/providers/cloudonix/provider.py @@ -120,7 +120,10 @@ class CloudonixProvider(TelephonyProvider): logger.info( f"Selected phone number {from_number} for outbound call to {to_number}" ) - workflow_id, user_id = kwargs["workflow_id"], kwargs["user_id"] + workflow_id, organization_id = ( + kwargs["workflow_id"], + kwargs["organization_id"], + ) # Prepare call data using Cloudonix callObject schema # Note: 'caller-id' is REQUIRED by Cloudonix API @@ -130,7 +133,7 @@ class CloudonixProvider(TelephonyProvider): "cxml": f""" - + """, @@ -408,7 +411,7 @@ class CloudonixProvider(TelephonyProvider): } async def get_webhook_response( - self, workflow_id: int, user_id: int, workflow_run_id: int + self, workflow_id: int, organization_id: int, workflow_run_id: int ) -> str: """ Dummy implementation - Cloudonix doesn't use webhook responses. @@ -430,7 +433,7 @@ class CloudonixProvider(TelephonyProvider): self, websocket: "WebSocket", workflow_id: int, - user_id: int, + organization_id: int, workflow_run_id: int, ) -> None: """ @@ -502,7 +505,7 @@ class CloudonixProvider(TelephonyProvider): provider_name=self.PROVIDER_NAME, workflow_id=workflow_id, workflow_run_id=workflow_run_id, - user_id=user_id, + organization_id=organization_id, call_id=call_id, transport_kwargs={"call_id": call_id, "stream_sid": stream_sid}, ) @@ -517,7 +520,6 @@ class CloudonixProvider(TelephonyProvider): *, organization_id: int, workflow_id: int, - user_id: int, workflow_run_id: int, params: Dict[str, str], ) -> None: @@ -671,7 +673,7 @@ class CloudonixProvider(TelephonyProvider): provider_name=self.PROVIDER_NAME, workflow_id=workflow_id, workflow_run_id=workflow_run_id, - user_id=user_id, + organization_id=organization_id, call_id=call_session, transport_kwargs={ "call_id": call_session, diff --git a/api/services/telephony/providers/plivo/provider.py b/api/services/telephony/providers/plivo/provider.py index 494552bc..f089403c 100644 --- a/api/services/telephony/providers/plivo/provider.py +++ b/api/services/telephony/providers/plivo/provider.py @@ -238,13 +238,13 @@ class PlivoProvider(TelephonyProvider): return any(hmac.compare_digest(computed, candidate) for candidate in candidates) async def get_webhook_response( - self, workflow_id: int, user_id: int, workflow_run_id: int + self, workflow_id: int, organization_id: int, workflow_run_id: int ) -> str: _, wss_backend_endpoint = await get_backend_endpoints() return f""" - {wss_backend_endpoint}/api/v1/telephony/ws/{workflow_id}/{user_id}/{workflow_run_id} + {wss_backend_endpoint}/api/v1/telephony/ws/{workflow_id}/{organization_id}/{workflow_run_id} """ async def get_call_cost(self, call_id: str) -> Dict[str, Any]: @@ -309,7 +309,7 @@ class PlivoProvider(TelephonyProvider): self, websocket: "WebSocket", workflow_id: int, - user_id: int, + organization_id: int, workflow_run_id: int, ) -> None: from api.services.pipecat.run_pipeline import run_pipeline_telephony @@ -348,7 +348,7 @@ class PlivoProvider(TelephonyProvider): provider_name=self.PROVIDER_NAME, workflow_id=workflow_id, workflow_run_id=workflow_run_id, - user_id=user_id, + organization_id=organization_id, call_id=call_id, transport_kwargs={"stream_id": stream_id, "call_id": call_id}, ) diff --git a/api/services/telephony/providers/plivo/routes.py b/api/services/telephony/providers/plivo/routes.py index ff64b379..70855379 100644 --- a/api/services/telephony/providers/plivo/routes.py +++ b/api/services/telephony/providers/plivo/routes.py @@ -74,7 +74,6 @@ async def _handle_plivo_status_callback( @router.post("/plivo-xml", include_in_schema=False) async def handle_plivo_xml_webhook( workflow_id: int, - user_id: int, workflow_run_id: int, organization_id: int, request: Request, @@ -110,7 +109,7 @@ async def handle_plivo_xml_webhook( ) response_content = await provider.get_webhook_response( - workflow_id, user_id, workflow_run_id + workflow_id, organization_id, workflow_run_id ) return HTMLResponse(content=response_content, media_type="application/xml") diff --git a/api/services/telephony/providers/telnyx/provider.py b/api/services/telephony/providers/telnyx/provider.py index 3adaef52..ccf4039b 100644 --- a/api/services/telephony/providers/telnyx/provider.py +++ b/api/services/telephony/providers/telnyx/provider.py @@ -101,10 +101,10 @@ class TelnyxProvider(TelephonyProvider): # Build the WebSocket stream URL for inline audio streaming workflow_id = kwargs.get("workflow_id") - user_id = kwargs.get("user_id") + organization_id = kwargs.get("organization_id") stream_url = ( f"{wss_backend_endpoint}/api/v1/telephony/ws" - f"/{workflow_id}/{user_id}/{workflow_run_id}" + f"/{workflow_id}/{organization_id}/{workflow_run_id}" ) # Build the webhook URL for status callbacks @@ -267,7 +267,7 @@ class TelnyxProvider(TelephonyProvider): return False async def get_webhook_response( - self, workflow_id: int, user_id: int, workflow_run_id: int + self, workflow_id: int, organization_id: int, workflow_run_id: int ) -> str: """Not used for Telnyx — streaming is inline with the dial request.""" return "" @@ -338,7 +338,7 @@ class TelnyxProvider(TelephonyProvider): self, websocket: "WebSocket", workflow_id: int, - user_id: int, + organization_id: int, workflow_run_id: int, ) -> None: """Handle Telnyx WebSocket connection for real-time audio. @@ -406,7 +406,7 @@ class TelnyxProvider(TelephonyProvider): provider_name=self.PROVIDER_NAME, workflow_id=workflow_id, workflow_run_id=workflow_run_id, - user_id=user_id, + organization_id=organization_id, call_id=call_control_id, transport_kwargs={ "stream_id": stream_id, diff --git a/api/services/telephony/providers/twilio/provider.py b/api/services/telephony/providers/twilio/provider.py index e9225039..423a6a1c 100644 --- a/api/services/telephony/providers/twilio/provider.py +++ b/api/services/telephony/providers/twilio/provider.py @@ -166,7 +166,7 @@ class TwilioProvider(TelephonyProvider): return validator.validate(url, params, signature) async def get_webhook_response( - self, workflow_id: int, user_id: int, workflow_run_id: int + self, workflow_id: int, organization_id: int, workflow_run_id: int ) -> str: """ Generate TwiML response for starting a call session. @@ -176,7 +176,7 @@ class TwilioProvider(TelephonyProvider): twiml_content = f""" - + """ @@ -274,7 +274,7 @@ class TwilioProvider(TelephonyProvider): self, websocket: "WebSocket", workflow_id: int, - user_id: int, + organization_id: int, workflow_run_id: int, ) -> None: """ @@ -325,7 +325,7 @@ class TwilioProvider(TelephonyProvider): provider_name=self.PROVIDER_NAME, workflow_id=workflow_id, workflow_run_id=workflow_run_id, - user_id=user_id, + organization_id=organization_id, call_id=call_sid, transport_kwargs={"stream_sid": stream_sid, "call_sid": call_sid}, ) diff --git a/api/services/telephony/providers/twilio/routes.py b/api/services/telephony/providers/twilio/routes.py index eb90b015..89fe15e7 100644 --- a/api/services/telephony/providers/twilio/routes.py +++ b/api/services/telephony/providers/twilio/routes.py @@ -47,7 +47,6 @@ async def _persist_amd_result_if_present( @router.post("/twiml", include_in_schema=False) async def handle_twiml_webhook( workflow_id: int, - user_id: int, workflow_run_id: int, organization_id: int, request: Request, @@ -79,7 +78,7 @@ async def handle_twiml_webhook( ) response_content = await provider.get_webhook_response( - workflow_id, user_id, workflow_run_id + workflow_id, organization_id, workflow_run_id ) return HTMLResponse(content=response_content, media_type="application/xml") diff --git a/api/services/telephony/providers/vobiz/provider.py b/api/services/telephony/providers/vobiz/provider.py index 0641ceb6..02c19cd8 100644 --- a/api/services/telephony/providers/vobiz/provider.py +++ b/api/services/telephony/providers/vobiz/provider.py @@ -255,7 +255,7 @@ class VobizProvider(TelephonyProvider): return is_valid async def get_webhook_response( - self, workflow_id: int, user_id: int, workflow_run_id: int + self, workflow_id: int, organization_id: int, workflow_run_id: int ) -> str: """ Generate Vobiz XML response for starting a call session. @@ -269,7 +269,7 @@ class VobizProvider(TelephonyProvider): vobiz_xml = f""" - {wss_backend_endpoint}/api/v1/telephony/ws/{workflow_id}/{user_id}/{workflow_run_id} + {wss_backend_endpoint}/api/v1/telephony/ws/{workflow_id}/{organization_id}/{workflow_run_id} """ return vobiz_xml @@ -350,7 +350,7 @@ class VobizProvider(TelephonyProvider): self, websocket: "WebSocket", workflow_id: int, - user_id: int, + organization_id: int, workflow_run_id: int, ) -> None: """ @@ -394,7 +394,7 @@ class VobizProvider(TelephonyProvider): provider_name=self.PROVIDER_NAME, workflow_id=workflow_id, workflow_run_id=workflow_run_id, - user_id=user_id, + organization_id=organization_id, call_id=call_id, transport_kwargs={"stream_id": stream_id, "call_id": call_id}, ) diff --git a/api/services/telephony/providers/vobiz/routes.py b/api/services/telephony/providers/vobiz/routes.py index 6e8e1317..4569c264 100644 --- a/api/services/telephony/providers/vobiz/routes.py +++ b/api/services/telephony/providers/vobiz/routes.py @@ -54,7 +54,7 @@ async def _verify_vobiz_callback( @router.post("/vobiz-xml", include_in_schema=False) async def handle_vobiz_xml_webhook( - workflow_id: int, user_id: int, workflow_run_id: int, organization_id: int + workflow_id: int, workflow_run_id: int, organization_id: int ): """ Handle initial webhook from Vobiz when call is answered. @@ -65,7 +65,7 @@ async def handle_vobiz_xml_webhook( set_current_run_id(workflow_run_id) logger.info( f"[run {workflow_run_id}] Vobiz XML webhook called - " - f"workflow_id={workflow_id}, user_id={user_id}, org_id={organization_id}" + f"workflow_id={workflow_id}, org_id={organization_id}" ) workflow_run = await db_client.get_workflow_run_by_id(workflow_run_id) @@ -74,7 +74,7 @@ async def handle_vobiz_xml_webhook( logger.debug(f"[run {workflow_run_id}] Using provider: {provider.PROVIDER_NAME}") response_content = await provider.get_webhook_response( - workflow_id, user_id, workflow_run_id + workflow_id, organization_id, workflow_run_id ) logger.debug( diff --git a/api/services/telephony/providers/vonage/provider.py b/api/services/telephony/providers/vonage/provider.py index 146a3394..51af251b 100644 --- a/api/services/telephony/providers/vonage/provider.py +++ b/api/services/telephony/providers/vonage/provider.py @@ -207,7 +207,7 @@ class VonageProvider(TelephonyProvider): return False async def get_webhook_response( - self, workflow_id: int, user_id: int, workflow_run_id: int + self, workflow_id: int, organization_id: int, workflow_run_id: int ) -> str: """ Generate NCCO response for starting a call session. @@ -222,7 +222,7 @@ class VonageProvider(TelephonyProvider): "endpoint": [ { "type": "websocket", - "uri": f"{wss_backend_endpoint}/api/v1/telephony/ws/{workflow_id}/{user_id}/{workflow_run_id}", + "uri": f"{wss_backend_endpoint}/api/v1/telephony/ws/{workflow_id}/{organization_id}/{workflow_run_id}", "content-type": "audio/l16;rate=16000", # 16kHz Linear PCM "headers": {}, } @@ -323,7 +323,7 @@ class VonageProvider(TelephonyProvider): self, websocket: "WebSocket", workflow_id: int, - user_id: int, + organization_id: int, workflow_run_id: int, ) -> None: """ @@ -338,14 +338,17 @@ class VonageProvider(TelephonyProvider): try: # Get workflow run to extract call UUID - workflow_run = await db_client.get_workflow_run(workflow_run_id) + workflow_run = await db_client.get_workflow_run( + workflow_run_id, organization_id=organization_id + ) if not workflow_run: logger.error(f"Workflow run {workflow_run_id} not found") await websocket.close(code=4404, reason="Workflow run not found") return - # Get workflow for organization info - workflow = await db_client.get_workflow(workflow_id, user_id) + workflow = await db_client.get_workflow( + workflow_id, organization_id=organization_id + ) if not workflow: logger.error(f"Workflow {workflow_id} not found") await websocket.close(code=4404, reason="Workflow not found") @@ -392,7 +395,7 @@ class VonageProvider(TelephonyProvider): provider_name=self.PROVIDER_NAME, workflow_id=workflow_id, workflow_run_id=workflow_run_id, - user_id=user_id, + organization_id=organization_id, call_id=call_uuid, transport_kwargs={"call_uuid": call_uuid}, ) diff --git a/api/services/telephony/providers/vonage/routes.py b/api/services/telephony/providers/vonage/routes.py index 7a8156bf..f3239249 100644 --- a/api/services/telephony/providers/vonage/routes.py +++ b/api/services/telephony/providers/vonage/routes.py @@ -5,7 +5,6 @@ provider registry — see ProviderSpec.router. """ import json -from typing import Optional from fastapi import APIRouter, HTTPException, Request from loguru import logger @@ -20,9 +19,8 @@ router = APIRouter() @router.get("/ncco", include_in_schema=False) async def handle_ncco_webhook( workflow_id: int, - user_id: int, workflow_run_id: int, - organization_id: Optional[int] = None, + organization_id: int, ): """Handle NCCO (Nexmo Call Control Objects) webhook for Vonage. @@ -30,12 +28,10 @@ async def handle_ncco_webhook( """ workflow_run = await db_client.get_workflow_run_by_id(workflow_run_id) - provider = await get_telephony_provider_for_run( - workflow_run, organization_id or user_id - ) + provider = await get_telephony_provider_for_run(workflow_run, organization_id) response_content = await provider.get_webhook_response( - workflow_id, user_id, workflow_run_id + workflow_id, organization_id, workflow_run_id ) return json.loads(response_content) diff --git a/api/tests/telephony/cloudonix/test_routes.py b/api/tests/telephony/cloudonix/test_routes.py index 33c75e77..5fb31622 100644 --- a/api/tests/telephony/cloudonix/test_routes.py +++ b/api/tests/telephony/cloudonix/test_routes.py @@ -94,7 +94,6 @@ async def test_agent_stream_reads_call_metadata_from_start_message(): websocket, organization_id=44, workflow_id=101, - user_id=202, workflow_run_id=303, params={}, ) @@ -267,7 +266,6 @@ async def test_agent_stream_handshake_timeout_closes_socket(monkeypatch): websocket, organization_id=1, workflow_id=2, - user_id=3, workflow_run_id=4, params={}, ), diff --git a/api/tests/telephony/plivo/test_routes.py b/api/tests/telephony/plivo/test_routes.py index e3a2b06e..9b9f892a 100644 --- a/api/tests/telephony/plivo/test_routes.py +++ b/api/tests/telephony/plivo/test_routes.py @@ -89,7 +89,6 @@ async def test_plivo_xml_route_accepts_valid_signature_with_extra_query_param(): provider = _provider() query = { "workflow_id": 7, - "user_id": 8, "workflow_run_id": 123, "campaign_id": 42, "organization_id": 11, @@ -138,14 +137,13 @@ async def test_plivo_xml_route_accepts_valid_signature_with_extra_query_param(): response = await handle_plivo_xml_webhook( workflow_id=7, - user_id=8, workflow_run_id=123, organization_id=11, request=request, ) assert response.body == b"" - get_webhook_response.assert_awaited_once_with(7, 8, 123) + get_webhook_response.assert_awaited_once_with(7, 11, 123) db_client.update_workflow_run.assert_awaited_once() diff --git a/api/tests/telephony/twilio/test_routes.py b/api/tests/telephony/twilio/test_routes.py index acbb9df9..d40e1ab1 100644 --- a/api/tests/telephony/twilio/test_routes.py +++ b/api/tests/telephony/twilio/test_routes.py @@ -109,7 +109,6 @@ async def test_twiml_route_accepts_valid_signature_with_extra_query_param(): provider = _provider() query = { "workflow_id": 7, - "user_id": 8, "workflow_run_id": 123, "campaign_id": 42, "organization_id": 11, @@ -149,14 +148,13 @@ async def test_twiml_route_accepts_valid_signature_with_extra_query_param(): response = await handle_twiml_webhook( workflow_id=7, - user_id=8, workflow_run_id=123, organization_id=11, request=request, ) assert response.body == b"" - get_webhook_response.assert_awaited_once_with(7, 8, 123) + get_webhook_response.assert_awaited_once_with(7, 11, 123) @pytest.mark.asyncio @@ -166,7 +164,6 @@ async def test_twiml_route_rejects_missing_signature(): path="/api/v1/telephony/twiml", query={ "workflow_id": 7, - "user_id": 8, "workflow_run_id": 123, "organization_id": 11, }, @@ -188,7 +185,6 @@ async def test_twiml_route_rejects_missing_signature(): with pytest.raises(HTTPException) as exc_info: await handle_twiml_webhook( workflow_id=7, - user_id=8, workflow_run_id=123, organization_id=11, request=request, diff --git a/api/tests/test_active_calls.py b/api/tests/test_active_calls.py index 575eb046..ccd761e8 100644 --- a/api/tests/test_active_calls.py +++ b/api/tests/test_active_calls.py @@ -183,7 +183,7 @@ async def test_telephony_entrypoint_counts_call_during_setup(monkeypatch): provider_name="twilio", workflow_id=1, workflow_run_id=44, - user_id=7, + organization_id=7, call_id="call-1", transport_kwargs={}, ) diff --git a/api/tests/test_public_agent_routes.py b/api/tests/test_public_agent_routes.py index 4c2a1793..3a00ef57 100644 --- a/api/tests/test_public_agent_routes.py +++ b/api/tests/test_public_agent_routes.py @@ -126,7 +126,8 @@ def test_trigger_route_executes_as_workflow_owner(): initiate_kwargs = provider.initiate_call.await_args.kwargs assert initiate_kwargs["workflow_id"] == workflow.id - assert initiate_kwargs["user_id"] == workflow.user_id + # The media websocket URL is keyed on the org, not the workflow owner. + assert initiate_kwargs["organization_id"] == workflow.organization_id def test_workflow_uuid_route_uses_scoped_lookup_and_shared_execution(): diff --git a/api/tests/test_telephony_routes.py b/api/tests/test_telephony_routes.py index 2aeb5588..b244c968 100644 --- a/api/tests/test_telephony_routes.py +++ b/api/tests/test_telephony_routes.py @@ -123,8 +123,12 @@ def test_initiate_call_executes_as_workflow_owner_for_shared_org_workflow(): initiate_kwargs = provider.initiate_call.await_args.kwargs assert initiate_kwargs["workflow_id"] == workflow.id - assert initiate_kwargs["user_id"] == workflow.user_id - assert "user_id=99" in initiate_kwargs["webhook_url"] + # The media websocket URL is keyed on the org, not the workflow owner. + assert initiate_kwargs["organization_id"] == workflow.organization_id + webhook_url = initiate_kwargs["webhook_url"] + assert f"organization_id={workflow.organization_id}" in webhook_url + # The answer URL carries no workflow owner: nothing downstream scopes on it. + assert "user_id=" not in webhook_url mock_db.get_user_configurations.assert_not_called() @@ -380,11 +384,13 @@ async def test_smallwebrtc_run_reaching_telephony_websocket_closes_without_runni ): mock_concurrency.unregister_active_call = AsyncMock() mock_db.get_workflow_run = AsyncMock(return_value=workflow_run) - mock_db.get_workflow_by_id = AsyncMock(return_value=workflow) + mock_db.get_workflow = AsyncMock(return_value=workflow) mock_db.update_workflow_run = AsyncMock() - await _handle_telephony_websocket(websocket, 33, 99, 501) + await _handle_telephony_websocket(websocket, 33, 11, 501) + mock_db.get_workflow_run.assert_awaited_once_with(501, organization_id=11) + mock_db.get_workflow.assert_awaited_once_with(33, organization_id=11) websocket.close.assert_awaited_once_with( code=4400, reason=( diff --git a/docs/integrations/telephony/asterisk-ari.mdx b/docs/integrations/telephony/asterisk-ari.mdx index cffb21e6..ae6eb2c1 100644 --- a/docs/integrations/telephony/asterisk-ari.mdx +++ b/docs/integrations/telephony/asterisk-ari.mdx @@ -107,7 +107,7 @@ The section name (e.g., `dograh`) is the **WebSocket Client Name** you'll enter -Configure the `uri` as a base URL only, without a query string. During each call, Dograh asks Asterisk to create an `externalMedia` channel and Asterisk appends `workflow_id`, `user_id`, and `workflow_run_id` through the `v()` transport data for that call. Opening `/api/v1/telephony/ws/ari` directly in a browser or with `wscat` can return HTTP 403 because those routing parameters are missing; that is expected and does not indicate a `websocket_client.conf` misconfiguration. +Configure the `uri` as a base URL only, without a query string. During each call, Dograh asks Asterisk to create an `externalMedia` channel and Asterisk appends `workflow_id`, `organization_id`, and `workflow_run_id` through the `v()` transport data for that call. Opening `/api/v1/telephony/ws/ari` directly in a browser or with `wscat` can return HTTP 403 because those routing parameters are missing; that is expected and does not indicate a `websocket_client.conf` misconfiguration. diff --git a/docs/integrations/telephony/custom.mdx b/docs/integrations/telephony/custom.mdx index ffdb9dc5..069fd7a2 100644 --- a/docs/integrations/telephony/custom.mdx +++ b/docs/integrations/telephony/custom.mdx @@ -71,11 +71,11 @@ class YourProvider(TelephonyProvider): # ---------- webhooks ---------- async def verify_webhook_signature(self, url, params, signature) -> bool: ... - async def get_webhook_response(self, workflow_id, user_id, workflow_run_id) -> str: ... + async def get_webhook_response(self, workflow_id, organization_id, workflow_run_id) -> str: ... def parse_status_callback(self, data: dict) -> dict: ... # ---------- websocket ---------- - async def handle_websocket(self, websocket, workflow_id, user_id, workflow_run_id): ... + async def handle_websocket(self, websocket, workflow_id, organization_id, workflow_run_id): ... # ---------- inbound ---------- @classmethod @@ -112,6 +112,14 @@ class YourProvider(TelephonyProvider): See `api/services/telephony/base.py` for the full docstrings on each method. + + `get_webhook_response` and `handle_websocket` receive an `organization_id`, not a + user id. It is the tenant that every workflow and workflow-run lookup must be + scoped by — pass it straight through to `run_pipeline_telephony`. The workflow + owner is deliberately not handed to providers; read it off the workflow row if you + need it for attribution. + + ## Implementation Guide ### 1. Configuration schemas @@ -385,7 +393,7 @@ For end-to-end testing, save your provider through the telephony-configurations 1. **Trust the registry** — never import another provider's class directly; resolve through the factory helpers (`get_default_telephony_provider`, `get_telephony_provider_by_id`, etc.). 2. **Sensitive fields** — mark every credential field `sensitive=True` in `ProviderUIMetadata`. The save endpoint masks these on read and preserves the original when the client re-submits a masked value. -3. **Inbound signature verification** — always validate inbound webhook signatures in `verify_inbound_signature`. Returning `True` when no signature header is present is acceptable; return `False` when a signature *is* present but invalid. +3. **Inbound signature verification** — always validate inbound webhook signatures in `verify_inbound_signature`, and fail closed. If your provider signs every webhook (most do), a *missing* signature header means the request did not come from the provider: return `False`, exactly as `providers/twilio/` and `providers/plivo/` do. Never `return True` as a placeholder — these handlers are reachable by anyone who can guess a `workflow_run_id`. 4. **Transports load credentials lazily** — call `load_credentials_for_transport` with the `telephony_configuration_id` from the workflow run. Don't read the org's default config from `transport.py`. 5. **Logging** — use `loguru.logger`. diff --git a/docs/integrations/telephony/webhooks.mdx b/docs/integrations/telephony/webhooks.mdx index f20c740a..8c0f0f10 100644 --- a/docs/integrations/telephony/webhooks.mdx +++ b/docs/integrations/telephony/webhooks.mdx @@ -5,17 +5,28 @@ description: "How Dograh AI handles telephony webhooks and audio streaming" ## Overview -Dograh AI uses webhooks to communicate with telephony providers for call events and audio streaming. Webhooks are automatically configured when you initiate calls. +Dograh AI uses webhooks to communicate with telephony providers for call events and audio streaming. For outbound calls, Dograh builds these URLs and hands them to the provider when it dials — you never configure them by hand. For inbound calls, you point your provider at a single dispatcher URL; see [Inbound Calls](/integrations/telephony/inbound). + +Every webhook path lives under `/api/v1/telephony`. ## Webhook Types -### 1. Initial Call Webhook +### 1. Answer Webhook -When a call is initiated, the telephony provider requests instructions. +When an outbound call connects, the provider requests instructions. The path is provider-specific, and Dograh appends the routing parameters as a query string: -**Endpoint**: `/api/v1/telephony/webhook/{workflow_id}/{user_id}/{workflow_run_id}` +``` +?workflow_id={workflow_id}&workflow_run_id={workflow_run_id}&organization_id={organization_id} +``` -**Purpose**: Returns provider-specific instructions +| Provider | Method | Path | +| --- | --- | --- | +| Twilio, Cloudonix | `POST` | `/twiml` | +| Plivo | `POST` | `/plivo-xml` | +| Vobiz | `POST` | `/vobiz-xml` | +| Vonage | `GET` | `/ncco` | + +Telnyx and Asterisk ARI have no answer webhook. Telnyx is call-control style — Dograh POSTs the stream and event URLs to Telnyx's API instead of returning markup. ARI streams over a WebSocket only. @@ -23,7 +34,7 @@ When a call is initiated, the telephony provider requests instructions. - + ``` @@ -35,7 +46,7 @@ When a call is initiated, the telephony provider requests instructions. "action": "connect", "endpoint": [{ "type": "websocket", - "uri": "wss://your-domain/api/v1/telephony/ws/123/456/789", + "uri": "wss://your-domain/api/v1/telephony/ws/123/11/789", "content-type": "audio/l16;rate=16000" }] } @@ -44,30 +55,50 @@ When a call is initiated, the telephony provider requests instructions. -### 2. Status Callback +Here `123` is the workflow, `11` the organization, and `789` the workflow run. -Receives call lifecycle events. +### 2. Status Callbacks -**Endpoint**: `/api/v1/telephony/status-callback/{workflow_run_id}` +Receive call lifecycle events. Each is keyed on the workflow run: + +| Provider | Path | +| --- | --- | +| Twilio | `/twilio/status-callback/{workflow_run_id}` | +| Plivo | `/plivo/hangup-callback/{workflow_run_id}`, `/plivo/ring-callback/{workflow_run_id}` | +| Vobiz | `/vobiz/hangup-callback/{workflow_run_id}`, `/vobiz/ring-callback/{workflow_run_id}` | +| Vonage | `/vonage/events/{workflow_run_id}` | +| Telnyx | `/telnyx/events/{workflow_run_id}` | +| Cloudonix | `/cloudonix/status-callback/{workflow_run_id}`, `/cloudonix/cdr` | + +Providers report their own vocabulary; Dograh normalizes it into a common set of states: -**Events Tracked**: - `initiated` - Call request received - `ringing` - Call is ringing +- `in-progress` - Call is connected and streaming - `answered` - Call was answered - `completed` - Call ended normally - `busy` - Line was busy - `no-answer` - Call not answered +- `canceled` - Call was canceled before connecting - `failed` - Call failed +- `error` - Provider reported an error + +A status Dograh does not recognize is passed through unchanged rather than dropped. ### 3. WebSocket Audio Stream Real-time audio streaming for voice interaction. -**Endpoint**: `/api/v1/telephony/ws/{workflow_id}/{user_id}/{workflow_run_id}` +**Endpoint**: `/api/v1/telephony/ws/{workflow_id}/{organization_id}/{workflow_run_id}` + +Asterisk ARI instead connects to `/api/v1/telephony/ws/ari` and passes the same three values as query parameters — see [Asterisk ARI](/integrations/telephony/asterisk-ari). + +The `organization_id` segment is the tenant that owns the workflow. Dograh scopes every workflow and workflow-run lookup by it, so a run belonging to one organization can never be served under another's id. **Audio Formats**: -- **Twilio**: 8kHz μ-law (MULAW), Base64-encoded in JSON messages +- **Twilio / Plivo / Vobiz**: 8kHz μ-law (MULAW), Base64-encoded in JSON messages - **Vonage**: 16kHz Linear PCM, Binary frames +- **Asterisk ARI**: 8kHz Linear PCM via externalMedia ## How It Works @@ -76,11 +107,14 @@ Dograh AI automatically: 2. Passes them to the telephony provider when initiating calls 3. Verifies webhook signatures for security: - **Twilio**: HMAC-SHA1 signature validation + - **Plivo / Vobiz**: HMAC-SHA256 signature validation - **Vonage**: JWT token verification 4. Processes status updates to track call lifecycle 5. Manages WebSocket connections for audio streaming 6. Handles provider-specific audio formats and protocols +Signature verification is implemented per provider in `verify_inbound_signature`. If you are adding a provider, see [Custom Telephony Provider](/integrations/telephony/custom). + ## Local Development For local development, use the built-in Cloudflare tunnel: @@ -102,16 +136,21 @@ The tunnel URL is automatically detected and used for webhooks. - Check firewall rules allow incoming HTTPS traffic - Test with `curl` from external network - + + + - Providers sign the full URL, query string included — a proxy that rewrites or reorders query parameters will invalidate the signature + - Confirm the backend's public URL matches what the provider was given, including scheme and port + + - Check WebSocket upgrade headers are preserved - Verify no timeout on load balancer/proxy - Monitor for memory/CPU constraints - + - Verify workflow_run_id is included in URL - Check provider console for webhook errors - Review webhook retry logs - \ No newline at end of file + diff --git a/ui/src/client/types.gen.ts b/ui/src/client/types.gen.ts index e5f2704e..d9fd6890 100644 --- a/ui/src/client/types.gen.ts +++ b/ui/src/client/types.gen.ts @@ -6121,12 +6121,7 @@ export type UpdateWorkflowRequest = { template_context_variables?: { [key: string]: unknown; } | null; - /** - * Workflow Configurations - */ - workflow_configurations?: { - [key: string]: unknown; - } | null; + workflow_configurations?: WorkflowConfigurationDefaults | null; }; /**