[fix] negodata: 견적 회사 스코프 적용 (목록·단건·세션 IDOR 차단) + 거짓 주석 제거

quotations 에 company_id 컬럼이 없어 견적 조회가 회사 무관하게 전부 노출되던 멀티테넌트 누수를 차단. 작성자(user_id)→users.company_id 서브쿼리로 회사 스코프를 걸어 목록(/list)과 단건·세션·협력사 엔드포인트(_fetch 가드)에서 남의 회사 견적을 NOT_FOUND 처리. 스케줄러/내부 호출은 company_id 없이 우회. 라우터/서비스의 '회사 스코핑 안 함' 거짓 주석 정정.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

negodata/backend/crud/quotation_crud.py

@@ -19,12 +19,12 @@ from common.utils.gtime import GTime
 class IQuotationCRUD(ABC):
     @abstractmethod
     async def search(
-        self, cdb: AsyncSession, owner, search, status, type_, start_from, start_to, skip, limit
+        self, cdb: AsyncSession, company_id, owner, search, status, type_, start_from, start_to, skip, limit
     ) -> Tuple[ErrorType, list, int]:
         pass
 
     @abstractmethod
-    async def get_by_id(self, cdb: AsyncSession, qt_id) -> Tuple[ErrorType, quotations]:
+    async def get_by_id(self, cdb: AsyncSession, qt_id, company_id=None) -> Tuple[ErrorType, quotations]:
         pass
 
     @abstractmethod
@@ -40,7 +40,7 @@ class IQuotationCRUD(ABC):
         pass
 
     @abstractmethod
-    async def get_last_supplier_type(self, cdb: AsyncSession, supplier_id) -> Tuple[ErrorType, Optional[tuple]]:
+    async def get_last_supplier_type(self, cdb: AsyncSession, supplier_id, company_id=None) -> Tuple[ErrorType, Optional[tuple]]:
         pass
 
     @abstractmethod
@@ -149,6 +149,7 @@ class QuotationCRUD(IQuotationCRUD):
     async def search(
         self,
         cdb: AsyncSession,
+        company_id,
         owner,
         search: Optional[str],
         status: Optional[str],
@@ -159,7 +160,11 @@ class QuotationCRUD(IQuotationCRUD):
         limit: int,
     ) -> Tuple[ErrorType, list, int]:
         try:
-            conditions = [quotations.deleted == False]  # noqa: E712
+            # 회사 스코프(멀티테넌트): quotations 엔 company_id 가 없어 작성자(user_id)→users.company_id 로 건다.
+            conditions = [
+                quotations.deleted == False,  # noqa: E712
+                quotations.user_id.in_(select(users.user_id).where(users.company_id == company_id)),
+            ]
             if owner:
                 conditions.append(quotations.user_id == owner)  # '내 견적만' — 작성자(user_id)=로그인 유저
             if search:
@@ -248,9 +253,13 @@ class QuotationCRUD(IQuotationCRUD):
             LOG.e_no_callstack(ex)
             return ErrorType.DB_RUN_FAILED, {}
 
-    async def get_by_id(self, cdb: AsyncSession, qt_id) -> Tuple[ErrorType, quotations]:
+    async def get_by_id(self, cdb: AsyncSession, qt_id, company_id=None) -> Tuple[ErrorType, quotations]:
         try:
-            query = select(quotations).where(quotations.qt_id == qt_id, quotations.deleted == False).limit(1)  # noqa: E712
+            # company_id 가 주어지면 회사 스코프(작성자 회사)로 좁힌다 — 남의 회사 견적은 '없음'으로 떨어진다.
+            conds = [quotations.qt_id == qt_id, quotations.deleted == False]  # noqa: E712
+            if company_id is not None:
+                conds.append(quotations.user_id.in_(select(users.user_id).where(users.company_id == company_id)))
+            query = select(quotations).where(*conds).limit(1)
             err_type, row_list = await DB_SESSION_MNG.execute(cdb, query)
             if err_type != ErrorType.SUCCESS:
                 return err_type, None
@@ -367,18 +376,21 @@ class QuotationCRUD(IQuotationCRUD):
             LOG.e_no_callstack(ex)
             return ErrorType.DB_RUN_FAILED, {}
 
-    async def get_last_supplier_type(self, cdb: AsyncSession, supplier_id) -> Tuple[ErrorType, Optional[tuple]]:
+    async def get_last_supplier_type(self, cdb: AsyncSession, supplier_id, company_id=None) -> Tuple[ErrorType, Optional[tuple]]:
         """협력사의 직전 견적 supplier_type. (supplier_type, qt_number) | None.
         sessions(supplier_id) ⨝ quotations 에서 supplier_type 가 있는 최신 견적 1건."""
         try:
+            conds = [
+                sessions.supplier_id == supplier_id,
+                quotations.supplier_type.isnot(None),
+                quotations.deleted == False,  # noqa: E712
+            ]
+            if company_id is not None:
+                conds.append(quotations.user_id.in_(select(users.user_id).where(users.company_id == company_id)))
             query = (
                 select(quotations.supplier_type, quotations.number)
                 .join(sessions, sessions.quotation_id == quotations.qt_id)
-                .where(
+                .where(*conds)
-                    sessions.supplier_id == supplier_id,
-                    quotations.supplier_type.isnot(None),
-                    quotations.deleted == False,  # noqa: E712
-                )
                 .order_by(quotations.created_at.desc())
                 .limit(1)
             )

negodata/backend/router/v1/quotation/quotation.py

@@ -23,8 +23,7 @@ from .protocol import (
     Res_TargetBreakdown,
 )
 
-# 라우터(컨트롤러). 인증(Depends(IsValidAccessToken))으로 UserInfo 를 받는다.
+# 라우터(컨트롤러). 인증(Depends(IsValidAccessToken))으로 UserInfo 를 받아 company_id 로 스코프(작성자 회사 기준).
-# quotations 테이블에 company_id 가 없어 회사 스코핑은 하지 않는다(토큰 검증만).
 # 라우팅 주의: 정적/하위 경로(/list, /create, /{qt_id}/status ...)를 /{qt_id} 보다 먼저 선언해야
 # /{qt_id} 가 /list 등을 가로채지 않는다.
 router = APIRouter(prefix="/v1/quotation", tags=["Quotation"], responses={404: {"description": "Not found"}})
@@ -44,7 +43,7 @@ async def list_quotations(
     pg: PageParams = Depends(),
 ):
     owner = user_info.user_id if mine else None
-    return RemoveNoneResponse(await service.list_quotations(owner, search, status, type, start_from, start_to, pg))
+    return RemoveNoneResponse(await service.list_quotations(user_info.company_id, owner, search, status, type, start_from, start_to, pg))
 
 
 @router.post(path="/create", response_model=Res_CreateQuotation, summary="견적 생성")
@@ -56,68 +55,68 @@ async def create_quotation(
 
 @router.post(path="/stop/{qt_id}", response_model=Res_Quotation, summary="견적 마감")
 async def stop_quotation(qt_id: UUID, service: QuotationService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
-    return RemoveNoneResponse(await service.stop_quotation(str(qt_id)))
+    return RemoveNoneResponse(await service.stop_quotation(str(qt_id), user_info.company_id))
 
 
 @router.post(path="/regenerate/{qt_id}", response_model=Res_CreateQuotation, summary="견적 재생성(다음 라운드)")
 async def regenerate_quotation(
     qt_id: UUID, req: Req_RegenerateQuotation, service: QuotationService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)
 ):
-    return RemoveNoneResponse(await service.regenerate_quotation(str(qt_id), req.supplier_ids))
+    return RemoveNoneResponse(await service.regenerate_quotation(str(qt_id), user_info.company_id, req.supplier_ids))
 
 
 # ----- 견적 상세 (FK로 연결된 하위 데이터 / 일부는 모델 미존재로 스텁) -----
 @router.get(path="/{qt_id}/status", response_model=Res_QuotationStatus, summary="견적 상태 조회")
 async def get_quotation_status(qt_id: UUID, service: QuotationService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
-    return RemoveNoneResponse(await service.get_status(str(qt_id)))
+    return RemoveNoneResponse(await service.get_status(str(qt_id), user_info.company_id))
 
 
 @router.get(path="/{qt_id}/sessions", response_model=Res_QuotationSessions, summary="참여현황")
 async def get_quotation_sessions(qt_id: UUID, service: QuotationService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
-    return RemoveNoneResponse(await service.list_sessions(str(qt_id)))
+    return RemoveNoneResponse(await service.list_sessions(str(qt_id), user_info.company_id))
 
 
 @router.post(path="/{qt_id}/notify", response_model=Res_NotifySessions, summary="협상 초청 메일 발송")
 async def notify_quotation(qt_id: UUID, service: QuotationService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
-    return RemoveNoneResponse(await service.notify_sessions(str(qt_id)))
+    return RemoveNoneResponse(await service.notify_sessions(str(qt_id), user_info.company_id))
 
 
 @router.get(path="/session/{session_id}/chat", response_model=Res_SessionChat, summary="채팅 상세")
 async def get_session_chat(session_id: UUID, service: QuotationService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
-    return RemoveNoneResponse(await service.list_chats(str(session_id)))
+    return RemoveNoneResponse(await service.list_chats(str(session_id), user_info.company_id))
 
 
 @router.get(path="/session/{session_id}/target-breakdown", response_model=Res_TargetBreakdown, summary="세션 목표가 산정내역")
 async def get_target_breakdown(session_id: UUID, service: QuotationService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
-    return RemoveNoneResponse(await service.get_target_breakdown(str(session_id)))
+    return RemoveNoneResponse(await service.get_target_breakdown(str(session_id), user_info.company_id))
 
 
 @router.post(path="/session/{session_id}/notify", response_model=Res_NotifySessions, summary="세션 초청 메일 재발송")
 async def notify_session(session_id: UUID, service: QuotationService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
-    return RemoveNoneResponse(await service.notify_session(str(session_id)))
+    return RemoveNoneResponse(await service.notify_session(str(session_id), user_info.company_id))
 
 
 @router.get(path="/{qt_id}/result", response_model=Res_QuotationResult, summary="낙찰 결과")
 async def get_quotation_result(qt_id: UUID, service: QuotationService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
-    return RemoveNoneResponse(await service.get_result(str(qt_id)))
+    return RemoveNoneResponse(await service.get_result(str(qt_id), user_info.company_id))
 
 
 @router.get(path="/{qt_id}/cards", response_model=Res_QuotationCards, summary="견적 사용 카드")
 async def get_quotation_cards(qt_id: UUID, service: QuotationService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
-    return RemoveNoneResponse(await service.list_cards(str(qt_id)))
+    return RemoveNoneResponse(await service.list_cards(str(qt_id), user_info.company_id))
 
 
 @router.delete(path="/delete/{qt_id}", response_model=Res_DeleteQuotation, summary="견적 삭제")
 async def delete_quotation(qt_id: UUID, service: QuotationService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
-    return RemoveNoneResponse(await service.delete_quotation(str(qt_id)))
+    return RemoveNoneResponse(await service.delete_quotation(str(qt_id), user_info.company_id))
 
 
 @router.get(path="/supplier/{supplier_id}/last-type", response_model=Res_LastSupplierType, summary="협력사 직전 견적 유형")
 async def get_supplier_last_type(supplier_id: UUID, service: QuotationService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
-    return RemoveNoneResponse(await service.get_last_supplier_type(str(supplier_id)))
+    return RemoveNoneResponse(await service.get_last_supplier_type(str(supplier_id), user_info.company_id))
 
 
 # ----- 단건 조회 (정적/하위 경로 뒤에 선언) -----
 @router.get(path="/{qt_id}", response_model=Res_Quotation, summary="견적 조회")
 async def get_quotation(qt_id: UUID, service: QuotationService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
-    return RemoveNoneResponse(await service.get_quotation(str(qt_id)))
+    return RemoveNoneResponse(await service.get_quotation(str(qt_id), user_info.company_id))

negodata/backend/services/quotation_service.py

@@ -39,8 +39,8 @@ from services.email import EmailUnavailable, build_invite_email, send_email
 class QuotationService:
     """견적 비즈니스 로직.
 
-    quotations 테이블에는 company_id 가 없어 회사 스코핑은 하지 않는다(토큰 검증만).
+    회사 스코프(멀티테넌트)는 작성자(user_id)→users.company_id 조인으로 건다(quotations 에 company_id 컬럼이 없음).
-    user_id 는 생성 시 소유자로만 기록한다(조회/변경 시 소유권 필터 없음).
+    목록(list_quotations)은 회사 스코프로 제한한다. user_id 는 '내 견적만' 추가 필터로도 쓴다.
     """
 
     # 기본 전략 버전(card.versions 시드). 견적 생성 시 version_id 미지정이면 이 값으로 채운다.
@@ -115,18 +115,19 @@ class QuotationService:
         now = GTime.UTC()
         return f"EST-{now:%Y%m}-{uuid.uuid4().hex[:4].upper()}"
 
-    async def _fetch(self, qt_id: uuid.UUID):
+    async def _fetch(self, qt_id: uuid.UUID, company_id=None):
-        """견적 단건 조회. (ErrorType, quotation|None) 반환. (회사 스코프 없음)"""
+        """견적 단건 조회. (ErrorType, quotation|None) 반환.
+        company_id 가 주어지면 회사 스코프(작성자 회사) 가드 — 남의 회사 견적은 NOT_FOUND. 내부/스케줄러 호출은 None."""
         err_type, quotation = await DB_SESSION_MNG.execute_lambda(
             quotations.DBType(),
             DBWRType.DB_READ.value,
-            lambda s: self.quotation_crud.get_by_id(s, qt_id),
+            lambda s: self.quotation_crud.get_by_id(s, qt_id, company_id),
         )
         if err_type != ErrorType.SUCCESS or quotation is None:
             return ErrorType.QUOTATION_NOT_FOUND, None
         return ErrorType.SUCCESS, quotation
 
-    async def get_target_breakdown(self, session_id: str) -> Res_TargetBreakdown:
+    async def get_target_breakdown(self, session_id: str, company_id=None) -> Res_TargetBreakdown:
         """세션 목표가 산정내역(후보·채택). 저장된 target_price/anchoring 은 그대로 표기하고,
         후보값은 생성과 동일한 _candidates 로직으로 계산해 내려준다(프론트 재계산 제거 → 항상 일치).
         상속분(재생성 라운드)은 현재 후보와 무관하므로 is_inherited=True, 채택 표시는 비운다."""
@@ -140,7 +141,7 @@ class QuotationService:
             res.result.SetResult(err_type if err_type != ErrorType.SUCCESS else ErrorType.QUOTATION_NOT_FOUND)
             return res
         sess = got[0]
-        err_type, quotation = await self._fetch(sess.quotation_id)
+        err_type, quotation = await self._fetch(sess.quotation_id, company_id)
         if err_type != ErrorType.SUCCESS:
             res.result.SetResult(err_type)
             return res
@@ -185,15 +186,16 @@ class QuotationService:
         res.target_anchoring_price = sess.target_anchoring_price
         return res
 
-    async def list_quotations(self, owner, search, status, type_, start_from, start_to, pg: PageParams) -> Res_QuotationList:
+    async def list_quotations(self, company_id, owner, search, status, type_, start_from, start_to, pg: PageParams) -> Res_QuotationList:
-        """견적 목록. owner(user_id) 가 주어지면 '내 견적만'(작성자=로그인 유저)으로 필터한다."""
+        """견적 목록. 회사(company_id) 스코프로 제한하고, owner(user_id) 가 주어지면 '내 견적만'으로 더 좁힌다."""
         res = Res_QuotationList(page=pg.page, size=pg.size)
 
+        company_uuid = uuid.UUID(company_id)
         owner_uuid = uuid.UUID(owner) if owner else None
         err_type, rows, total = await DB_SESSION_MNG.execute_lambda(
             quotations.DBType(),
             DBWRType.DB_READ.value,
-            lambda s: self.quotation_crud.search(s, owner_uuid, search, status, type_, start_from, start_to, pg.skip, pg.size),
+            lambda s: self.quotation_crud.search(s, company_uuid, owner_uuid, search, status, type_, start_from, start_to, pg.skip, pg.size),
         )
         if err_type != ErrorType.SUCCESS:
             res.result.SetResult(err_type)
@@ -239,22 +241,22 @@ class QuotationService:
         res.total = total
         return res
 
-    async def get_quotation(self, qt_id: str) -> Res_Quotation:
+    async def get_quotation(self, qt_id: str, company_id=None) -> Res_Quotation:
         res = Res_Quotation()
-        err_type, quotation = await self._fetch(uuid.UUID(qt_id))
+        err_type, quotation = await self._fetch(uuid.UUID(qt_id), company_id)
         if err_type != ErrorType.SUCCESS:
             res.result.SetResult(err_type)
             return res
         res.quotation = QuotationData.model_validate(quotation)
         return res
 
-    async def get_last_supplier_type(self, supplier_id: str) -> Res_LastSupplierType:
+    async def get_last_supplier_type(self, supplier_id: str, company_id=None) -> Res_LastSupplierType:
         """협력사의 직전 견적 supplier_type(견적생성 모달 프리필용). 이력 없으면 비워서 반환."""
         res = Res_LastSupplierType()
         err_type, got = await DB_SESSION_MNG.execute_lambda(
             quotations.DBType(),
             DBWRType.DB_READ.value,
-            lambda s: self.quotation_crud.get_last_supplier_type(s, uuid.UUID(supplier_id)),
+            lambda s: self.quotation_crud.get_last_supplier_type(s, uuid.UUID(supplier_id), company_id),
         )
         if err_type != ErrorType.SUCCESS:
             res.result.SetResult(err_type)
@@ -675,14 +677,14 @@ class QuotationService:
         no_part = sum(1 for pref, eq in flags if pref is False and eq is False)
         return no_part, equal
 
-    async def regenerate_quotation(self, qt_id: str, supplier_ids: list) -> Res_CreateQuotation:
+    async def regenerate_quotation(self, qt_id: str, company_id, supplier_ids: list) -> Res_CreateQuotation:
         """[프론트] 마감된 견적을 골라 수동으로 다음 라운드를 생성한다.
         크론/수동마감의 자동 재생성과 달리 사유·체인 한도 판정 없이, 프론트가 고른 공급사로 바로 만든다.
         상품·기간·견적번호·카드버전은 원 견적에서 이어받는다(regenerate_next_round)."""
         res = Res_CreateQuotation()
         qt_uuid = uuid.UUID(qt_id)
 
-        err_type, original = await self._fetch(qt_uuid)
+        err_type, original = await self._fetch(qt_uuid, company_id)
         if err_type != ErrorType.SUCCESS or original is None:
             res.result.SetResult(err_type)
             return res
@@ -708,26 +710,26 @@ class QuotationService:
 
         return await self.regenerate_next_round(qt_uuid, supplier_ids)
 
-    async def stop_quotation(self, qt_id: str) -> Res_Quotation:
+    async def stop_quotation(self, qt_id: str, company_id=None) -> Res_Quotation:
         """[프론트] 수동 견적마감. 크론과 똑같은 마감 판정(close_and_decide)을 탄다
         (단독낙찰 확정 / 동가·미참여면 다음 라운드 재생성 / 거부·한도면 그냥 마감)."""
         res = Res_Quotation()
         qt_uuid = uuid.UUID(qt_id)
 
-        # 존재 확인
+        # 존재 확인(+회사 가드)
-        err_type, _ = await self._fetch(qt_uuid)
+        err_type, _ = await self._fetch(qt_uuid, company_id)
         if err_type != ErrorType.SUCCESS:
             res.result.SetResult(err_type)
             return res
 
         await self.close_and_decide(qt_uuid)
-        return await self.get_quotation(qt_id)
+        return await self.get_quotation(qt_id, company_id)
 
-    async def delete_quotation(self, qt_id: str) -> Res_DeleteQuotation:
+    async def delete_quotation(self, qt_id: str, company_id=None) -> Res_DeleteQuotation:
         res = Res_DeleteQuotation()
         qt_uuid = uuid.UUID(qt_id)
 
-        err_type, _ = await self._fetch(qt_uuid)
+        err_type, _ = await self._fetch(qt_uuid, company_id)
         if err_type != ErrorType.SUCCESS:
             res.result.SetResult(err_type)
             return res
@@ -740,9 +742,9 @@ class QuotationService:
             res.result.SetResult(err_type)
         return res
 
-    async def get_status(self, qt_id: str) -> Res_QuotationStatus:
+    async def get_status(self, qt_id: str, company_id=None) -> Res_QuotationStatus:
         res = Res_QuotationStatus()
-        err_type, quotation = await self._fetch(uuid.UUID(qt_id))
+        err_type, quotation = await self._fetch(uuid.UUID(qt_id), company_id)
         if err_type != ErrorType.SUCCESS:
             res.result.SetResult(err_type)
             return res
@@ -751,9 +753,9 @@ class QuotationService:
         res.message = "ok"
         return res
 
-    async def get_result(self, qt_id: str) -> Res_QuotationResult:
+    async def get_result(self, qt_id: str, company_id=None) -> Res_QuotationResult:
         res = Res_QuotationResult()
-        err_type, quotation = await self._fetch(uuid.UUID(qt_id))
+        err_type, quotation = await self._fetch(uuid.UUID(qt_id), company_id)
         if err_type != ErrorType.SUCCESS:
             res.result.SetResult(err_type)
             return res
@@ -766,10 +768,10 @@ class QuotationService:
         res.result_count = 0
         return res
 
-    async def list_sessions(self, qt_id: str) -> Res_QuotationSessions:
+    async def list_sessions(self, qt_id: str, company_id=None) -> Res_QuotationSessions:
         res = Res_QuotationSessions()
         qt_uuid = uuid.UUID(qt_id)
-        err_type, quotation = await self._fetch(qt_uuid)
+        err_type, quotation = await self._fetch(qt_uuid, company_id)
         if err_type != ErrorType.SUCCESS:
             res.result.SetResult(err_type)
             return res
@@ -812,12 +814,12 @@ class QuotationService:
         return res
 
     # ----- 협상 초청 메일 (수동 발송)
-    async def notify_sessions(self, qt_id: str) -> Res_NotifySessions:
+    async def notify_sessions(self, qt_id: str, company_id=None) -> Res_NotifySessions:
         """[수동 발송] 견적의 '미발송' 세션(공급사 담당자)에게 협상 초청 메일을 일괄 발송한다.
         대상 = email_sent_at IS NULL + 담당자 이메일 보유."""
         res = Res_NotifySessions()
         qt_uuid = uuid.UUID(qt_id)
-        err_type, quotation = await self._fetch(qt_uuid)
+        err_type, quotation = await self._fetch(qt_uuid, company_id)
         if err_type != ErrorType.SUCCESS:
             res.result.SetResult(err_type)
             return res
@@ -848,7 +850,7 @@ class QuotationService:
             await self._mark_emailed(sent_ids)
         return res
 
-    async def notify_session(self, session_id: str) -> Res_NotifySessions:
+    async def notify_session(self, session_id: str, company_id=None) -> Res_NotifySessions:
         """[수동 재발송] 단일 세션(공급사)에 초청 메일 발송(이미 보냈어도 강제 재발송)."""
         res = Res_NotifySessions()
         sess_uuid = uuid.UUID(session_id)
@@ -862,7 +864,7 @@ class QuotationService:
             return res
         sess, sp_name, email = got[0], got[1], got[2]
         res.total = 1
-        err_type, quotation = await self._fetch(sess.quotation_id)
+        err_type, quotation = await self._fetch(sess.quotation_id, company_id)
         if err_type != ErrorType.SUCCESS:
             res.result.SetResult(err_type)
             return res
@@ -911,11 +913,26 @@ class QuotationService:
             [lambda s: self.quotation_crud.mark_sessions_emailed(s, session_ids, now)],
         )
 
-    async def list_chats(self, session_id: str) -> Res_SessionChat:
+    async def list_chats(self, session_id: str, company_id=None) -> Res_SessionChat:
         res = Res_SessionChat()
         sess_uuid = uuid.UUID(session_id)
         res.session_id = sess_uuid
 
+        # 회사 가드: chats 는 session 키라 세션→견적→회사로 확인한다(남의 회사 세션이면 NOT_FOUND).
+        if company_id is not None:
+            g_err, got = await DB_SESSION_MNG.execute_lambda(
+                sessions.DBType(),
+                DBWRType.DB_READ.value,
+                lambda s: self.quotation_crud.get_session_with_supplier(s, sess_uuid),
+            )
+            if g_err != ErrorType.SUCCESS or got is None:
+                res.result.SetResult(g_err if g_err != ErrorType.SUCCESS else ErrorType.QUOTATION_NOT_FOUND)
+                return res
+            guard_err, _ = await self._fetch(got[0].quotation_id, company_id)
+            if guard_err != ErrorType.SUCCESS:
+                res.result.SetResult(guard_err)
+                return res
+
         err_type, rows = await DB_SESSION_MNG.execute_lambda(
             chats.DBType(),
             DBWRType.DB_READ.value,
@@ -945,10 +962,10 @@ class QuotationService:
         ]
         return res
 
-    async def list_cards(self, qt_id: str) -> Res_QuotationCards:
+    async def list_cards(self, qt_id: str, company_id=None) -> Res_QuotationCards:
         res = Res_QuotationCards()
         qt_uuid = uuid.UUID(qt_id)
-        err_type, quotation = await self._fetch(qt_uuid)
+        err_type, quotation = await self._fetch(qt_uuid, company_id)
         if err_type != ErrorType.SUCCESS:
             res.result.SetResult(err_type)
             return res