fix: address CRITICAL security issues - permission escalation and encoding inconsistency

backend/src/v1/inbox_messages/schemas.py

@@ -8,6 +8,31 @@ from uuid import UUID
 from pydantic import BaseModel, ConfigDict
 
 
+class PermissionBits:
+    VIEW = 1  # 001
+    INVITE = 2  # 010
+    EDIT = 4  # 100
+
+    @classmethod
+    def encode(cls, view: bool, edit: bool, invite: bool) -> int:
+        value = 0
+        if view:
+            value |= cls.VIEW
+        if edit:
+            value |= cls.EDIT
+        if invite:
+            value |= cls.INVITE
+        return value
+
+    @classmethod
+    def decode(cls, permission: int) -> dict[str, bool]:
+        return {
+            "view": bool(permission & cls.VIEW),
+            "edit": bool(permission & cls.EDIT),
+            "invite": bool(permission & cls.INVITE),
+        }
+
+
 class InboxMessageType(str, Enum):
     FRIEND_REQUEST = "friend_request"
     CALENDAR = "calendar"
@@ -41,6 +66,8 @@ class InboxMessageListRequest(BaseModel):
 
 
 class InboxMessageAcceptRequest(BaseModel):
+    model_config: ClassVar[ConfigDict] = ConfigDict(extra="forbid")
+
     permission_view: bool = True
     permission_edit: bool = False
     permission_invite: bool = False

backend/src/v1/inbox_messages/service.py

@@ -4,6 +4,8 @@ from enum import Enum
 from typing import TYPE_CHECKING
 from uuid import UUID
 
+import json
+
 from fastapi import HTTPException
 from sqlalchemy.exc import SQLAlchemyError
 
@@ -22,6 +24,7 @@ from v1.inbox_messages.schemas import (
     InboxMessageResponse,
     InboxMessageStatus,
     InboxMessageType,
+    PermissionBits,
 )
 
 if TYPE_CHECKING:
@@ -84,11 +87,23 @@ class InboxMessageService(BaseService):
                     status_code=400, detail="Message is not a calendar invitation"
                 )
 
-            permission = self._encode_permission(request)
+            invited_permission = self._parse_invited_permission(message.content)
+            requested_permission = PermissionBits.encode(
+                request.permission_view,
+                request.permission_edit,
+                request.permission_invite,
+            )
+            final_permission = requested_permission & invited_permission
+            if final_permission == 0:
+                raise HTTPException(
+                    status_code=400,
+                    detail="No valid permissions requested (must be subset of invited permissions)",
+                )
+
             subscription = ScheduleSubscription(
                 item_id=message.schedule_item_id,
                 subscriber_id=user_id,
-                permission=permission,
+                permission=final_permission,
                 status=SubscriptionStatus.ACTIVE,
                 created_by=user_id,
             )
@@ -98,7 +113,12 @@ class InboxMessageService(BaseService):
                 user_id,
                 InboxMessageStatus.ACCEPTED.value,
             )
+            if updated is None:
+                await self._session.rollback()
+                raise HTTPException(status_code=404, detail="Inbox message not found")
             await self._session.commit()
+        except HTTPException:
+            raise
         except SQLAlchemyError:
             await self._session.rollback()
             logger.exception(
@@ -110,8 +130,6 @@ class InboxMessageService(BaseService):
                 status_code=503, detail="Inbox message store unavailable"
             )
 
-        if updated is None:
-            raise HTTPException(status_code=404, detail="Inbox message not found")
         return self._to_response(updated)
 
     async def dismiss_invitation(self, message_id: UUID) -> InboxMessageResponse:
@@ -160,15 +178,14 @@ class InboxMessageService(BaseService):
             created_at=message.created_at,
         )
 
-    def _encode_permission(self, request: InboxMessageAcceptRequest) -> int:
-        permission = 0
-        if request.permission_view:
-            permission |= 1
-        if request.permission_edit:
-            permission |= 2
-        if request.permission_invite:
-            permission |= 4
-        return permission
+    def _parse_invited_permission(self, content: str | None) -> int:
+        if not content:
+            return 0
+        try:
+            data = json.loads(content)
+            return int(data.get("permission", 0))
+        except (json.JSONDecodeError, ValueError, TypeError):
+            return 0
 
     def _status_value(self, status: object) -> str:
         if isinstance(status, Enum):

backend/src/v1/schedule_items/schemas.py

@@ -5,7 +5,7 @@ from enum import Enum
 from typing import ClassVar
 from uuid import UUID
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, EmailStr, Field
 
 
 class AttachmentType(str, Enum):
@@ -99,7 +99,9 @@ class ScheduleItemListRequest(BaseModel):
 
 
 class ScheduleItemShareRequest(BaseModel):
-    email: str = Field(..., description="Email of user to share with")
+    model_config: ClassVar[ConfigDict] = ConfigDict(extra="forbid")
+
+    email: EmailStr = Field(..., description="Email of user to share with")
     permission_view: bool = Field(True, description="Grant view permission")
     permission_edit: bool = Field(False, description="Grant edit permission")
     permission_invite: bool = Field(False, description="Grant invite permission")