apunkt/dograh
1
0
Fork 0
mirror of https://github.com/dograh-hq/dograh.git synced 2026-07-25 12:01:04 +02:00
Code Issues Projects Releases Packages Wiki Activity

fix: fix org scoped access for resources

Browse source
This commit is contained in:
Abhishek Kumar 2026-07-09 16:39:34 +05:30
parent f3bcf24370
commit e4b53f78e9
47 changed files with 2667 additions and 400 deletions
Download patch file Download diff file Expand all files Collapse all files

2
api/services/campaign/campaign_call_dispatcher.py
View file

@ -303,6 +303,7 @@ class CampaignCallDispatcher:
initial_context=initial_context,
campaign_id=campaign.id,
queued_run_id=queued_run.id, # Link to queued run for retry tracking
organization_id=campaign.organization_id,
)
# Store slot_id mapping in Redis for cleanup later
@ -342,6 +343,7 @@ class CampaignCallDispatcher:
quota_result = await authorize_workflow_run_start(
workflow_id=campaign.workflow_id,
organization_id=campaign.organization_id,
workflow_run_id=workflow_run.id,
)
if not quota_result.has_quota:

35
api/services/configuration/ai_model_configuration.py
View file

@ -55,9 +55,9 @@ class WorkflowAIModelConfigurationMigrationResult:
async def get_resolved_ai_model_configuration(
*,
user_id: int | None,
organization_id: int | None,
) -> ResolvedAIModelConfiguration:
"""Resolve the effective model configuration for an organization."""
organization_configuration = await get_organization_ai_model_configuration_v2(
organization_id
)
@ -68,22 +68,14 @@ async def get_resolved_ai_model_configuration(
organization_configuration=organization_configuration,
)
if user_id is None:
return ResolvedAIModelConfiguration(
effective=EffectiveAIModelConfiguration(),
source="empty",
)
legacy = await db_client.get_user_configurations(user_id)
return ResolvedAIModelConfiguration(
effective=legacy,
source="legacy_user_v1" if _has_model_services(legacy) else "empty",
effective=EffectiveAIModelConfiguration(),
source="empty",
)
async def get_effective_ai_model_configuration_for_workflow(
*,
user_id: int | None,
organization_id: int | None,
workflow_configurations: dict | None,
) -> EffectiveAIModelConfiguration:
@ -97,7 +89,6 @@ async def get_effective_ai_model_configuration_for_workflow(
)
resolved_config = await get_resolved_ai_model_configuration(
user_id=user_id,
organization_id=organization_id,
)
return resolve_effective_config(
@ -145,19 +136,12 @@ async def migrate_workflow_model_configurations_to_v2(
fallback_user_config: EffectiveAIModelConfiguration,
) -> WorkflowAIModelConfigurationMigrationResult:
workflows = await _list_workflows_for_model_configuration_migration(organization_id)
owner_configs: dict[int, EffectiveAIModelConfiguration] = {}
workflow_updates: list[tuple[int, dict]] = []
definition_updates: list[tuple[int, dict]] = []
migrated_workflow_ids: set[int] = set()
for workflow in workflows:
base_config = fallback_user_config
if workflow.user_id is not None:
if workflow.user_id not in owner_configs:
owner_configs[
workflow.user_id
] = await db_client.get_user_configurations(workflow.user_id)
base_config = owner_configs[workflow.user_id]
workflow_configs, workflow_changed = (
migrate_workflow_configuration_model_override_to_v2(
@ -420,19 +404,6 @@ def _mask_secret_value(value):
return mask_key(value)
def _has_model_services(configuration: EffectiveAIModelConfiguration) -> bool:
return any(
service is not None
for service in (
configuration.llm,
configuration.tts,
configuration.stt,
configuration.embeddings,
configuration.realtime,
)
)
def _convert_any_dograh_legacy_configuration(
configuration: EffectiveAIModelConfiguration,
dograh_key: str,

1
api/services/organization_context.py
View file

@ -31,7 +31,6 @@ async def get_organization_context(user: UserModel) -> OrganizationContextRespon
)
resolved = await get_resolved_ai_model_configuration(
user_id=user.id,
organization_id=organization_id,
)
managed_service_version = resolved.effective.managed_service_version

4
api/services/pipecat/in_memory_buffers.py
View file

@ -233,7 +233,9 @@ class InMemoryLogsBuffer:
if self._user_speech_start_timestamp:
payload_with_timestamps["timestamp"] = self._user_speech_start_timestamp
if self._user_speech_end_timestamp:
payload_with_timestamps["end_timestamp"] = self._user_speech_end_timestamp
payload_with_timestamps["end_timestamp"] = (
self._user_speech_end_timestamp
)
elif event_type == RealtimeFeedbackType.BOT_TEXT.value:
bot_interval_is_active = self._bot_speech_end_timestamp is None
if bot_interval_is_active and self._bot_speech_start_timestamp:

4
api/services/pipecat/realtime_feedback_observer.py
View file

@ -53,10 +53,10 @@ from pipecat.frames.frames import (
TranscriptionFrame,
TTSSpeakFrame,
TTSTextFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
UserMuteStartedFrame,
UserMuteStoppedFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
VADUserStartedSpeakingFrame,
VADUserStoppedSpeakingFrame,
)

59
api/services/pipecat/run_pipeline.py
View file

@ -323,7 +323,7 @@ async def _run_pipeline_telephony_impl(
"telephony_configuration_id"
)
# Resolve effective user config here so the transport can tune its
# Resolve effective org config here so the transport can tune its
# bot-stopped-speaking fallback based on is_realtime; pass the resolved
# values into _run_pipeline so it doesn't fetch them again.
from api.services.configuration.ai_model_configuration import (
@ -334,7 +334,6 @@ async def _run_pipeline_telephony_impl(
(workflow_run.definition.workflow_configurations or {}) if workflow_run else {}
)
user_config = await get_effective_ai_model_configuration_for_workflow(
user_id=user_id,
organization_id=workflow.organization_id if workflow else None,
workflow_configurations=run_configs,
)
@ -379,6 +378,7 @@ async def run_pipeline_smallwebrtc(
user_id: int,
call_context_vars: dict = {},
user_provider_id: str | None = None,
organization_id: int | None = None,
) -> None:
"""Run pipeline for WebRTC connections."""
# Register before any async setup so deploy drains see calls that are still
@ -392,6 +392,7 @@ async def run_pipeline_smallwebrtc(
user_id,
call_context_vars=call_context_vars,
user_provider_id=user_provider_id,
organization_id=organization_id,
)
finally:
unregister_active_call(workflow_run_id)
@ -404,6 +405,7 @@ async def _run_pipeline_smallwebrtc_impl(
user_id: int,
call_context_vars: dict = {},
user_provider_id: str | None = None,
organization_id: int | None = None,
) -> None:
"""Run pipeline for WebRTC connections"""
logger.debug(
@ -411,8 +413,14 @@ async def _run_pipeline_smallwebrtc_impl(
)
set_current_run_id(workflow_run_id)
# Get workflow to extract all pipeline configurations
workflow = await db_client.get_workflow(workflow_id, user_id)
workflow_scope = (
{"organization_id": organization_id}
if organization_id is not None
else {"user_id": user_id}
)
# Get workflow to extract all pipeline configurations.
workflow = await db_client.get_workflow(workflow_id, **workflow_scope)
# Set org context early so tasks created by the transport inherit it
if workflow:
@ -428,19 +436,26 @@ async def _run_pipeline_smallwebrtc_impl(
# Create audio configuration for WebRTC
audio_config = create_audio_config(WorkflowRunMode.SMALLWEBRTC.value)
# Resolve workflow_run + effective user_config here so the transport can
# Resolve workflow_run + effective org config here so the transport can
# tune its bot-stopped-speaking fallback based on is_realtime. _run_pipeline
# reuses these via kwargs so we don't fetch twice.
from api.services.configuration.ai_model_configuration import (
get_effective_ai_model_configuration_for_workflow,
)
workflow_run = await db_client.get_workflow_run(workflow_run_id, user_id)
workflow_run = await db_client.get_workflow_run(workflow_run_id, **workflow_scope)
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",
)
run_configs = (
(workflow_run.definition.workflow_configurations or {}) if workflow_run else {}
)
user_config = await get_effective_ai_model_configuration_for_workflow(
user_id=user_id,
organization_id=workflow.organization_id if workflow else None,
workflow_configurations=run_configs,
)
@ -463,6 +478,7 @@ async def _run_pipeline_smallwebrtc_impl(
user_provider_id=user_provider_id,
workflow_run=workflow_run,
resolved_user_config=user_config,
organization_id=organization_id,
)
@ -476,6 +492,7 @@ async def _run_pipeline(
user_provider_id: str | None = None,
workflow_run=None,
resolved_user_config=None,
organization_id: int | None = None,
) -> None:
"""Run the pipeline with active-call drain accounting."""
register_active_call(workflow_run_id)
@ -490,6 +507,7 @@ async def _run_pipeline(
user_provider_id=user_provider_id,
workflow_run=workflow_run,
resolved_user_config=resolved_user_config,
organization_id=organization_id,
)
finally:
unregister_active_call(workflow_run_id)
@ -505,6 +523,7 @@ async def _run_pipeline_impl(
user_provider_id: str | None = None,
workflow_run=None,
resolved_user_config=None,
organization_id: int | None = None,
) -> None:
"""
Run the pipeline with the given transport and configuration
@ -515,11 +534,26 @@ async def _run_pipeline_impl(
workflow_run_id: The ID of the workflow run
user_id: The ID of the user
workflow_run: Pre-fetched workflow run row. Fetched here if None.
resolved_user_config: User configuration with model_overrides already
applied. Fetched and resolved here if None.
resolved_user_config: Organization model configuration with workflow
model_overrides already applied. Fetched and resolved here if None.
"""
workflow_scope = (
{"organization_id": organization_id}
if organization_id is not None
else {"user_id": user_id}
)
if workflow_run is None:
workflow_run = await db_client.get_workflow_run(workflow_run_id, user_id)
workflow_run = await db_client.get_workflow_run(
workflow_run_id, **workflow_scope
)
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",
)
# If the workflow run is already completed, we don't need to run it again
if workflow_run.is_completed:
@ -532,7 +566,7 @@ async def _run_pipeline_impl(
merged_call_context_vars = {**merged_call_context_vars, **call_context_vars}
# Get workflow for metadata (name, organization_id, call_disposition_codes)
workflow = await db_client.get_workflow(workflow_id, user_id)
workflow = await db_client.get_workflow(workflow_id, **workflow_scope)
if not workflow:
raise HTTPException(status_code=404, detail="Workflow not found")
@ -564,7 +598,7 @@ async def _run_pipeline_impl(
term.strip() for term in dictionary.split(",") if term.strip()
]
# Resolve model overrides from the version onto global user config (skip
# Resolve model overrides from the version onto global org config (skip
# when the caller already resolved it).
if resolved_user_config is None:
from api.services.configuration.ai_model_configuration import (
@ -572,7 +606,6 @@ async def _run_pipeline_impl(
)
user_config = await get_effective_ai_model_configuration_for_workflow(
user_id=user_id,
organization_id=workflow.organization_id,
workflow_configurations=run_configs,
)

67
api/services/quota_service.py
View file

@ -337,6 +337,7 @@ async def _authorize_oss_managed_v2_correlation(
async def authorize_workflow_run_start(
*,
workflow_id: int,
organization_id: int,
workflow_run_id: int | None = None,
actor_user: UserModel | None = None,
) -> QuotaCheckResult:
@ -344,30 +345,75 @@ async def authorize_workflow_run_start(
The workflow organization is the billing subject for hosted deployments.
OSS deployments are billed per service key instead. The workflow owner is
used only to resolve the effective model configuration.
used only as billing metadata.
"""
if organization_id is None:
logger.warning(
"Workflow start authorization denied: missing organization scope for workflow {}",
workflow_id,
)
return QuotaCheckResult(
has_quota=False,
error_code="workflow_not_found",
error_message="Workflow not found",
)
try:
workflow = await db_client.get_workflow_by_id(workflow_id)
if not workflow:
workflow = await db_client.get_workflow(
workflow_id,
organization_id=organization_id,
)
except Exception as e:
logger.error(
"Workflow start authorization denied: failed to load workflow {} for org {}: {}",
workflow_id,
organization_id,
e,
)
return QuotaCheckResult(
has_quota=False,
error_code="workflow_not_found",
error_message="Workflow not found",
)
if not workflow:
logger.warning(
"Workflow start authorization denied: workflow {} not found for org {}",
workflow_id,
organization_id,
)
return QuotaCheckResult(
has_quota=False,
error_code="workflow_not_found",
error_message="Workflow not found",
)
try:
actor_id = getattr(actor_user, "id", None) if actor_user is not None else None
if actor_user is not None and actor_id is None:
logger.warning(
"Workflow start authorization denied: actor is missing id for workflow {} org {}",
workflow_id,
organization_id,
)
return QuotaCheckResult(
has_quota=False,
error_code="workflow_not_found",
error_message="Workflow not found",
)
actor_id = getattr(actor_user, "id", None)
if actor_id is not None and workflow.organization_id is not None:
if actor_id is not None:
try:
is_member = await db_client.is_user_member_of_organization(
user_id=actor_id,
organization_id=workflow.organization_id,
organization_id=organization_id,
)
except Exception as e:
logger.error(
"Workflow start authorization denied: failed to validate actor {} membership for workflow {} org {}: {}",
actor_id,
workflow_id,
workflow.organization_id,
organization_id,
e,
)
return QuotaCheckResult(
@ -380,7 +426,7 @@ async def authorize_workflow_run_start(
"Workflow start authorization denied: actor {} is not a member of workflow {} org {}",
actor_id,
workflow_id,
workflow.organization_id,
organization_id,
)
return QuotaCheckResult(
has_quota=False,
@ -397,15 +443,14 @@ async def authorize_workflow_run_start(
)
user_config = await get_effective_ai_model_configuration_for_workflow(
user_id=workflow_owner.id,
organization_id=workflow.organization_id,
organization_id=organization_id,
workflow_configurations=workflow.workflow_configurations,
)
if DEPLOYMENT_MODE != "oss":
return await _authorize_hosted_workflow_run_start(
workflow_owner=workflow_owner,
organization_id=workflow.organization_id,
organization_id=organization_id,
workflow_id=workflow.id,
workflow_run_id=workflow_run_id,
user_config=user_config,

2
api/services/telephony/ari_manager.py
View file

@ -591,6 +591,7 @@ class ARIConnection:
gathered_context={
"call_id": call_id,
},
organization_id=self.organization_id,
)
logger.info(
@ -603,6 +604,7 @@ class ARIConnection:
# store the MPS correlation id before the pipeline starts.
quota_result = await authorize_workflow_run_start(
workflow_id=inbound_workflow_id,
organization_id=self.organization_id,
workflow_run_id=workflow_run.id,
)
if not quota_result.has_quota:

13
api/services/workflow/qa/llm_config.py
View file

@ -12,7 +12,7 @@ async def resolve_llm_config(
"""Resolve the LLM provider, model, API key, and extra kwargs for QA analysis.
If the QA node has its own LLM configuration (qa_use_workflow_llm=False),
use those settings directly. Otherwise, fall back to the user's configured LLM.
use those settings directly. Otherwise, fall back to the workflow/org LLM.
Returns:
(provider, model, api_key, service_kwargs) tuple — service_kwargs can be
@ -30,7 +30,7 @@ async def resolve_llm_config(
kwargs,
)
# Fall back to user's configured LLM
# Fall back to the workflow/org configured LLM
provider, model, api_key, kwargs = await resolve_user_llm_config(workflow_run)
if qa_data.qa_model and qa_data.qa_model != "default":
@ -42,17 +42,13 @@ async def resolve_llm_config(
async def resolve_user_llm_config(
workflow_run: WorkflowRunModel,
) -> tuple[str, str, str, dict]:
"""Resolve the user's configured LLM (from EffectiveAIModelConfiguration).
"""Resolve the workflow/org configured LLM.
Returns:
(provider, model, api_key, service_kwargs) tuple
"""
user_id = None
if workflow_run.workflow and workflow_run.workflow.user:
user_id = workflow_run.workflow.user.id
llm_config: dict = {}
if user_id:
if workflow_run.workflow:
from api.services.configuration.ai_model_configuration import (
get_effective_ai_model_configuration_for_workflow,
)
@ -68,7 +64,6 @@ async def resolve_user_llm_config(
)
user_configuration = await get_effective_ai_model_configuration_for_workflow(
user_id=user_id,
organization_id=workflow_run.workflow.organization_id
if workflow_run.workflow
else None,

1
api/services/workflow/text_chat_runner.py
View file

@ -458,7 +458,6 @@ async def execute_text_chat_pending_turn(
)
user_config = await get_effective_ai_model_configuration_for_workflow(
user_id=workflow_run.workflow.user.id,
organization_id=workflow.organization_id,
workflow_configurations=run_configs,
)