P0+P1 fixes: RCE sandbox, SQL injection, RLS tenant isolation, DB roles, test syntax, attachment, permission registry, membership check

app/services/attachment_service.py

@@ -47,8 +47,18 @@ async def save_attachment(
     filename: str,
     file_content: bytes,
     mime_type: str,
+    is_system_admin: bool = False,
 ) -> dict[str, Any]:
     """Save a file to storage and create an Attachment record."""
+    # Check access on parent entity
+    if not is_system_admin:
+        from app.core.visibility import check_single_entity_access
+        has_access = await check_single_entity_access(
+            db, entity_type, entity_id, user_id, tenant_id, "write", is_system_admin
+        )
+        if not has_access:
+            raise PermissionError(f"No write access to {entity_type} {entity_id}")
+
     # Generate unique filename and relative storage path
     unique_filename = _generate_unique_filename(filename)
     file_path = f"{entity_type}/{entity_id}/{unique_filename}"

app/services/entity_permission_service.py

@@ -27,9 +27,70 @@ from app.models.entity_permission import EntityPermission
 from app.models.group import Group, UserGroup
 from app.models.role import Role
 from app.models.user import User, UserTenant
+from app.models.contact import Contact
+from app.models.address import Address
+from app.models.attachment import Attachment
+from app.models.bank_account import BankAccount
+from app.models.workflow import Workflow
+from app.models.sequence import Sequence
+from app.models.saved_filter import SavedFilter
+from app.models.saved_view import SavedView
+from app.models.webhook import Webhook
+from app.models.notification import Notification
+from app.models.custom_field_definition import CustomFieldDefinition
+from app.models.contact_folder import ContactFolder
 
 logger = logging.getLogger(__name__)
 
+# ── Entity Model Registry ────────────────────────────────────────────────────
+# Maps entity_type string to SQLAlchemy model class.
+# This replaces insecure text(f"SELECT ... FROM {entity_type}s") queries
+# with safe SQLAlchemy model-based queries (prevents SQL injection).
+ENTITY_MODELS: dict[str, type] = {
+    "contact": Contact,
+    "address": Address,
+    "attachment": Attachment,
+    "bank_account": BankAccount,
+    "workflow": Workflow,
+    "sequence": Sequence,
+    "saved_filter": SavedFilter,
+    "saved_view": SavedView,
+    "webhook": Webhook,
+    "notification": Notification,
+    "custom_field_definition": CustomFieldDefinition,
+    "contact_folder": ContactFolder,
+}
+
+# Try to add plugin models if available
+try:
+    from app.plugins.builtins.dms.models import DmsFile
+    ENTITY_MODELS["dms_file"] = DmsFile
+except ImportError:
+    pass
+try:
+    from app.plugins.builtins.calendar.models import CalendarEvent
+    ENTITY_MODELS["calendar_event"] = CalendarEvent
+except ImportError:
+    pass
+try:
+    from app.plugins.builtins.tasks.models import Task
+    ENTITY_MODELS["task"] = Task
+except ImportError:
+    pass
+try:
+    from app.plugins.builtins.mail.models import Mailbox
+    ENTITY_MODELS["mailbox"] = Mailbox
+except ImportError:
+    pass
+
+
+def _get_entity_model(entity_type: str) -> type:
+    """Get SQLAlchemy model class for entity_type, or raise ValueError."""
+    model = ENTITY_MODELS.get(entity_type)
+    if model is None:
+        raise ValueError(f"Unknown entity type: {entity_type}")
+    return model
+
 # Permission hierarchy: higher = more access
 _PERM_RANK = {"none": 0, "read": 1, "write": 2, "admin": 3, "delete": 4, "owner": 5}
 
@@ -350,11 +411,10 @@ async def get_effective_access(
     if user_q.scalar():
         return "delete"
 
-    # Check ownership — load the entity's owner_id
-    # We use raw SQL to avoid importing every model
+    # Check ownership — load the entity's owner_id via SQLAlchemy model (safe from SQL injection)
+    model = _get_entity_model(entity_type)
     owner_q = await db.execute(
-        text(f"SELECT owner_id FROM {entity_type}s WHERE id = :eid AND tenant_id = :tid"),
-        {"eid": entity_id, "tid": tenant_id},
+        select(model.owner_id).where(model.id == entity_id).where(model.tenant_id == tenant_id)
     )
     owner_row = owner_q.first()
     if not owner_row:
@@ -460,10 +520,11 @@ async def get_visible_ids(
     )
     if user_q.scalar():
         # Return all entity IDs
-        all_q = await db.execute(
-            text(f"SELECT id FROM {entity_type}s WHERE tenant_id = :tid AND deleted_at IS NULL"),
-            {"tid": tenant_id},
-        )
+        model = _get_entity_model(entity_type)
+        admin_q = select(model.id).where(model.tenant_id == tenant_id)
+        if hasattr(model, 'deleted_at'):
+            admin_q = admin_q.where(model.deleted_at.is_(None))
+        all_q = await db.execute(admin_q)
         all_ids = {row[0] for row in all_q}
         return all_ids, {eid: "delete" for eid in all_ids}
 
@@ -483,10 +544,11 @@ async def get_visible_ids(
     role_id = role_q.scalar_one_or_none()
 
     # 1. Owned entities
-    owned_q = await db.execute(
-        text(f"SELECT id FROM {entity_type}s WHERE tenant_id = :tid AND owner_id = :uid AND deleted_at IS NULL"),
-        {"tid": tenant_id, "uid": user_id},
-    )
+    model = _get_entity_model(entity_type)
+    owned_q_builder = select(model.id).where(model.tenant_id == tenant_id).where(model.owner_id == user_id)
+    if hasattr(model, 'deleted_at'):
+        owned_q_builder = owned_q_builder.where(model.deleted_at.is_(None))
+    owned_q = await db.execute(owned_q_builder)
     visible: set[uuid.UUID] = set()
     access_map: dict[uuid.UUID, str] = {}
     for (eid,) in owned_q:
@@ -494,10 +556,10 @@ async def get_visible_ids(
         access_map[eid] = "owner"
 
     # 2. Tenant-owned entities (owner_id IS NULL)
-    tenant_owned_q = await db.execute(
-        text(f"SELECT id FROM {entity_type}s WHERE tenant_id = :tid AND owner_id IS NULL AND deleted_at IS NULL"),
-        {"tid": tenant_id},
-    )
+    tenant_q_builder = select(model.id).where(model.tenant_id == tenant_id).where(model.owner_id.is_(None))
+    if hasattr(model, 'deleted_at'):
+        tenant_q_builder = tenant_q_builder.where(model.deleted_at.is_(None))
+    tenant_owned_q = await db.execute(tenant_q_builder)
     for (eid,) in tenant_owned_q:
         if eid not in visible:
             visible.add(eid)
@@ -591,16 +653,17 @@ async def batch_get_effective_access(
 
     result: dict[uuid.UUID, str] = {}
 
-    # 1. Check ownership
-    owner_q = await db.execute(
-        text(f"SELECT id, owner_id FROM {entity_type}s WHERE id = ANY(:ids) AND tenant_id = :tid AND deleted_at IS NULL"),
-        {"ids": [str(eid) for eid in entity_ids], "tid": tenant_id},
-    )
+    # 1. Check ownership via SQLAlchemy model (safe from SQL injection)
+    model = _get_entity_model(entity_type)
+    batch_q = select(model.id, model.owner_id).where(model.id.in_(entity_ids)).where(model.tenant_id == tenant_id)
+    if hasattr(model, 'deleted_at'):
+        batch_q = batch_q.where(model.deleted_at.is_(None))
+    owner_q = await db.execute(batch_q)
     for eid, owner_id in owner_q:
         if owner_id == user_id:
-            result[uuid.UUID(str(eid))] = "owner"
+            result[eid] = "owner"
         elif owner_id is None:
-            result[uuid.UUID(str(eid))] = "read"
+            result[eid] = "read"
 
     # 2. Check permissions
     now = datetime.now(UTC)