diff --git a/app/api/documents.py b/app/api/documents.py index d3af888..8160f32 100644 --- a/app/api/documents.py +++ b/app/api/documents.py @@ -28,7 +28,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from starlette.requests import ClientDisconnect from ai.client import AIClient, _load_prompt, parse_json_response -from core.auth import get_current_user +from core.auth import get_current_user, get_egress_class from core.config import settings from core.database import async_session, get_session from core.utils import file_hash @@ -742,11 +742,31 @@ async def get_document( doc_id: int, user: Annotated[User, Depends(get_current_user)], session: Annotated[AsyncSession, Depends(get_session)], + egress_class: Annotated[str, Depends(get_egress_class)], ): - """문서 단건 조회. 본문(extracted_text)·canonical markdown 동봉.""" + """문서 단건 조회. 본문(extracted_text)·canonical markdown 동봉. + + cloud egress(갭2): egress=cloud 토큰(예: Claude/MCP)은 search 와 동일한 cloud-eligibility + 게이트를 통과한 문서만 열람 가능 — id 직접 fetch 로 비공개/인프라/개인/restricted 문서를 + 우회 열람하는 경로를 차단한다. 부적격은 404(존재 자체 비노출). local 토큰=게이트 미발동(무회귀). + """ + from sqlalchemy import text as sql_text + from services.search.retrieval_service import cloud_eligible_doc_sql + doc = await session.get(Document, doc_id) if not doc or doc.deleted_at is not None: raise HTTPException(status_code=404, detail="문서를 찾을 수 없습니다") + if egress_class == "cloud": + eligible = ( + await session.execute( + sql_text( + "SELECT 1 FROM documents WHERE id = :doc_id AND deleted_at IS NULL" + + cloud_eligible_doc_sql("") + ).bindparams(doc_id=doc_id) + ) + ).first() + if eligible is None: + raise HTTPException(status_code=404, detail="문서를 찾을 수 없습니다") return DocumentDetailResponse.model_validate(doc) diff --git a/app/services/search/retrieval_service.py b/app/services/search/retrieval_service.py index 7a22e76..cf93602 100644 --- a/app/services/search/retrieval_service.py +++ b/app/services/search/retrieval_service.py @@ -145,6 +145,18 @@ def _license_sql(alias: str) -> str: return " AND " + restricted_exclude_sql(alias) +def cloud_eligible_doc_sql(alias: str = "") -> str: + """단일 문서가 cloud 소비자(예: Claude/MCP)에게 노출 가능한가 = search retrieval 과 + 동일한 egress allowlist(갭2) + license 제한(B-4) 결합 술어. fetch_document(cloud) 가 + search 와 byte-동일 게이트를 공유하도록 단일 source([[feedback_structural_integrity_over_path_discipline]]). + + cloud_egress·license leg 모두 bind 파라미터 없는 리터럴 술어라 호출측 추가 params 불요. + 주의: _license_sql 은 소유자 단건 다운로드엔 미적용(a안)이지만, cloud 노출은 구매 전자책 + verbatim 누출을 막아야 하므로 여기선 항상 적용 = search 와 동일(local 토큰은 이 게이트 미발동). + 반환 ' AND (egress allowlist) AND (license)' (alias='' = 컬럼 직접 참조). default-deny.""" + return _axis_sql(alias, AxisFilter(cloud_egress=True), {}) + _license_sql(alias) + + # 2단계 gate (R2-B1) — SQL string interpolation 직전 final allowlist. _VALID_DOCS_TABLE = re.compile(r"^(documents|documents_cand_[a-z0-9_]+)$") # corpus_chunks = document_chunks WHERE in_corpus=true 뷰 (Hier-Decomp-1 c2 choke point).