"""Fail-closed tenant/owner lookup helpers for file-backed resources with durable (PostgreSQL-target) read-first cutover for projects. Project lookups prefer the durable ``projects`` table when the session factory is available, then fall back to the legacy filesystem ``ProjectManager``. This keeps read paths working during the filesystem → durable migration without breaking existing routes that consume the legacy ``Project`` dataclass shape. """ from __future__ import annotations import re from typing import Optional from flask import current_app, request from ..models.project import Project, ProjectManager, ProjectStatus from ..models.task import Task, TaskManager from ..services.report_agent import Report, ReportManager from ..services.simulation_manager import SimulationManager, SimulationState from ..utils.api_errors import ApiError from .auth import current_actor from .policy import Role _SAFE_RESOURCE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$") def _valid_resource_id(value: str) -> bool: return isinstance(value, str) and bool(_SAFE_RESOURCE_ID.fullmatch(value)) def _durable_project_for_scope( project_id: str, *, organization_id: str, owner_user_id: Optional[str] ) -> Optional[Project]: """Resolve a project from the durable table and wrap it in the legacy shape. Only active when the local (durable) memory backend is selected; the legacy Zep backend keeps reading the filesystem manager exclusively. Returns None when the durable store is unavailable or no matching tenant/owner row exists. Does not raise. """ try: from ..config import Config if Config.MEMORY_BACKEND != "local": return None session_factory = current_app.extensions.get("crowdsight_session_factory") if session_factory is None: return None from ..services.product_repository import ProductRepository session = session_factory() try: row = ProductRepository(session).get_project( project_id, organization_id=organization_id ) finally: session.close() if row is None: return None if owner_user_id is not None and row.owner_user_id != owner_user_id: return None try: project_status = ProjectStatus(row.status) except ValueError: project_status = ProjectStatus.CREATED return Project( project_id=row.id, name=row.name, status=project_status, created_at=row.created_at.isoformat() if row.created_at else "", updated_at=row.updated_at.isoformat() if row.updated_at else "", organization_id=row.organization_id, owner_user_id=row.owner_user_id, files=[], total_text_length=row.total_text_length, ontology=row.ontology or None, analysis_summary=row.analysis_summary, graph_id=row.graph_id, graph_build_task_id=row.graph_build_task_id, simulation_requirement=row.simulation_requirement, error=row.error, ) except Exception: return None def scoped_project(project_id: str) -> Optional[Project]: if not _valid_resource_id(project_id): return None actor = current_actor() owner_user_id = actor.user_id if actor.role is Role.USER else None durable = _durable_project_for_scope( project_id, organization_id=actor.organization_id, owner_user_id=owner_user_id, ) if durable is not None: return durable return ProjectManager.get_project_for_scope( project_id, organization_id=actor.organization_id, owner_user_id=owner_user_id, ) def require_scoped_project(project_id: str) -> Project: project = scoped_project(project_id) if project is None: raise ApiError("resource_not_found", 404, "common.notFound") return project def scoped_graph(graph_id: str) -> Optional[Project]: if not _valid_resource_id(graph_id): return None actor = current_actor() return ProjectManager.find_project_by_graph_id( graph_id, organization_id=actor.organization_id, owner_user_id=actor.user_id if actor.role is Role.USER else None, ) def scoped_simulation(simulation_id: str) -> Optional[SimulationState]: if not _valid_resource_id(simulation_id): return None state = SimulationManager().get_simulation(simulation_id) if state is None: return None if scoped_project(state.project_id) is None: return None return state def require_scoped_simulation(simulation_id: str) -> SimulationState: state = scoped_simulation(simulation_id) if state is None: raise ApiError("resource_not_found", 404, "common.notFound") return state def scoped_task(task_id: str) -> Optional[Task]: if not _valid_resource_id(task_id): return None actor = current_actor() task = TaskManager().get_task( task_id, organization_id=actor.organization_id, owner_user_id=actor.user_id if actor.role is Role.USER else None, ) if task is None: return None metadata = task.metadata if isinstance(task.metadata, dict) else {} if metadata.get("organization_id") != actor.organization_id: return None if actor.role is Role.USER and metadata.get("owner_user_id") != actor.user_id: return None return task def require_scoped_task(task_id: str) -> Task: task = scoped_task(task_id) if task is None: raise ApiError("resource_not_found", 404, "common.notFound") return task def scoped_report(report_id: str) -> Optional[Report]: if not _valid_resource_id(report_id): return None report = ReportManager.get_report(report_id) if report is None: return None if scoped_simulation(report.simulation_id) is None: return None return report def require_scoped_report(report_id: str) -> Report: report = scoped_report(report_id) if report is None: raise ApiError("resource_not_found", 404, "common.notFound") return report def scoped_reports(simulation_id: str | None = None, limit: int = 50) -> list[Report]: if simulation_id is not None and scoped_simulation(simulation_id) is None: return [] safe_limit = min(max(int(limit), 1), 100) reports = ReportManager.list_reports(simulation_id=simulation_id, limit=safe_limit * 4) return [report for report in reports if scoped_report(report.report_id) is not None][:safe_limit] def scoped_simulations(project_id: str | None = None) -> list[SimulationState]: if project_id is not None and scoped_project(project_id) is None: return [] simulations = SimulationManager().list_simulations(project_id=project_id) return [state for state in simulations if scoped_simulation(state.simulation_id) is not None] def enforce_request_scope() -> None: """Reject IDs outside the current actor's tenant/owner scope before handlers run.""" view_args = request.view_args or {} payload = request.get_json(silent=True) payload = payload if isinstance(payload, dict) else {} project_id = view_args.get("project_id") or request.args.get("project_id") or payload.get("project_id") simulation_id = view_args.get("simulation_id") or request.args.get("simulation_id") or payload.get("simulation_id") report_id = view_args.get("report_id") or request.args.get("report_id") or payload.get("report_id") graph_id = view_args.get("graph_id") or request.args.get("graph_id") or payload.get("graph_id") task_id = view_args.get("task_id") or request.args.get("task_id") or payload.get("task_id") if project_id and scoped_project(project_id) is None: raise ApiError("resource_not_found", 404, "common.notFound") if simulation_id and scoped_simulation(simulation_id) is None: raise ApiError("resource_not_found", 404, "common.notFound") if report_id and scoped_report(report_id) is None: raise ApiError("resource_not_found", 404, "common.notFound") if graph_id and scoped_graph(graph_id) is None: raise ApiError("resource_not_found", 404, "common.notFound") if task_id and scoped_task(task_id) is None: raise ApiError("resource_not_found", 404, "common.notFound")