diff --git a/B05_Profile/B05_Profile_Router_Terrain.py b/B05_Profile/B05_Profile_Router_Terrain.py index 27faaa26c..ee918e151 100644 --- a/B05_Profile/B05_Profile_Router_Terrain.py +++ b/B05_Profile/B05_Profile_Router_Terrain.py @@ -199,6 +199,32 @@ def _design_grid( } +async def dtm_grid_npz(connection, project_id: UUID, model_id: int, smooth: bool) -> Path: + """그 프로젝트 DTM 모델의 npz(평활 고르기 = 서버 표본기와 같음) — 없으면 LookupError(까닭). + + 경로 = 프로젝트 storage 안 DB 상대 경로만(밖으로 나가면 LookupError · 계산 창구 `/calc` 도 씀). + """ + stored = await get_project_storage_relative_path(connection, project_id) + async with connection.cursor() as cursor: + await cursor.execute( + "SELECT model_type, model_file_path FROM surface_models" + " WHERE id = %s AND project_id = %s", + (model_id, str(project_id)), + ) + row = await cursor.fetchone() + if not stored or not row or row[0] != "dtm" or not row[1]: + raise LookupError("DTM 격자가 없는 지표면 모델입니다.") + root = Path(resolve_stored_project_path(stored)) + model_path = (root / str(row[1])).resolve() + if not model_path.is_relative_to(root.resolve()): + raise LookupError("DTM 격자가 없는 지표면 모델입니다.") + smooth_path = model_path.with_name(f"{model_path.stem}_smooth.npz") + npz_path = smooth_path if smooth and smooth_path.is_file() else model_path + if not npz_path.is_file(): + raise LookupError("DTM 격자 파일이 없습니다.") + return npz_path + + @router.get("/{project_id}/surface/models/{model_id}/grid", response_model=None) async def read_surface_grid( request: Request, project_id: UUID, model_id: int, smooth: bool = False @@ -209,26 +235,9 @@ async def read_surface_grid( """ pool = get_db_pool() async with pool.acquire() as connection: - stored = await get_project_storage_relative_path(connection, project_id) - async with connection.cursor() as cursor: - await cursor.execute( - "SELECT model_type, model_file_path FROM surface_models" - " WHERE id = %s AND project_id = %s", - (model_id, str(project_id)), - ) - row = await cursor.fetchone() - if not stored or not row or row[0] != "dtm" or not row[1]: - return JSONResponse( - status_code=404, - content={"status": "error", "message": "DTM 격자가 없는 지표면 모델입니다."}, - ) - model_path = Path(resolve_stored_project_path(stored)) / str(row[1]) - smooth_path = model_path.with_name(f"{model_path.stem}_smooth.npz") - npz_path = smooth_path if smooth and smooth_path.is_file() else model_path - if not npz_path.is_file(): - return JSONResponse( - status_code=404, - content={"status": "error", "message": "DTM 격자 파일이 없습니다."}, - ) + try: + npz_path = await dtm_grid_npz(connection, project_id, model_id, smooth) + except LookupError as exc: + return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)}) grid_path = await asyncio.to_thread(dtm_grid_file, npz_path) return cached_file_response(request, grid_path, "application/octet-stream") diff --git a/B06_Section/B06_Section_Router_Calc.py b/B06_Section/B06_Section_Router_Calc.py index 91e88a4ef..0bd7c8a29 100644 --- a/B06_Section/B06_Section_Router_Calc.py +++ b/B06_Section/B06_Section_Router_Calc.py @@ -5,6 +5,7 @@ 브라우저가 계산 자리 판(`calc_sites.json`)에서 묶음을 「서버」로 둔 때만 부름 (`design_calc_server.ts`). 권한 = 프로젝트 접근(`protected_with_company`) · 호출량 = `main.py` 미들웨어 · 본문 상한 `MAX_BODY`. +파일 경로 = 서버가 프로젝트 id 로 정함(그 storage 안 DTM 격자) · 입력에 경로 칸 = 400. 실행 = 늘 떠 있는 Node 하나(`common_util_calc_runner.run_calc_served`) · 정본 저장 길은 이 창구를 안 씀. """ @@ -17,6 +18,7 @@ from fastapi import APIRouter, Request from fastapi.responses import JSONResponse from common_util.common_util_calc_runner import SERVER_ONLY_IDS, run_calc_served +from config.config_db import get_db_pool router = APIRouter(prefix="/api/projects", tags=["B06 Calc Server"]) @@ -28,6 +30,38 @@ def _error(status: int, message: str) -> JSONResponse: return JSONResponse(status_code=status, content={"status": "error", "message": message}) +async def _server_grid(project_id: UUID, inputs: dict) -> str | None: + """파일 경로는 브라우저 값을 안 씀 — 단계 격자 = `{model_id, smooth, design}` → 이 프로젝트 + storage 안 DTM 격자 경로(서버가 정함) · 경로 칸(`stages.grid.path` · `stations.grid`) = 거절.""" + stations = inputs.get("stations") + if isinstance(stations, dict) and "grid" in stations: + return "격자 경로는 서버가 정함" + stages = inputs.get("stages") + grid = stages.get("grid") if isinstance(stages, dict) else None + if grid is None: + return None + if not isinstance(grid, dict) or "path" in grid: + return "격자 경로는 서버가 정함 — {model_id, smooth, design}" + model_id, design = grid.get("model_id"), grid.get("design", "las") + if ( + not isinstance(model_id, int) + or isinstance(model_id, bool) + or design not in ("las", "sheet") + ): + return "격자 = {model_id: 정수, smooth, design: las | sheet}" + from B05_Profile.B05_Profile_Router_Terrain import dtm_grid_npz + from common_util.common_util_surface_sampler import dtm_grid_file + + async with get_db_pool().acquire() as connection: + try: + npz = await dtm_grid_npz(connection, project_id, model_id, bool(grid.get("smooth"))) + except LookupError as exc: + return str(exc) + path = await asyncio.to_thread(dtm_grid_file, npz) + stages["grid"] = {"path": str(path), "design": design} + return None + + @router.post("/{project_id}/calc", response_model=None) async def run_calc_endpoint(project_id: UUID, request: Request) -> dict | JSONResponse: declared = request.headers.get("content-length") @@ -50,6 +84,9 @@ async def run_calc_endpoint(project_id: UUID, request: Request) -> dict | JSONRe return _error(400, "inputs = 객체") if SERVER_ONLY_IDS.intersection(run): return _error(400, "서버 측점 생성은 창구 밖") + problem = await _server_grid(project_id, inputs) + if problem: + return _error(400, problem) if "stations.list" in run and isinstance(inputs.get("stations"), dict): # 심는 측점 · 카드 이름 표 = 서버 구조물 정의(서버 측점 생성과 같은 표 · 511-B B-2) from B05_Profile.B05_Profile_Engine_Sections_Node import structure_table diff --git a/resources/tester/test_511_b_calc_path_guard.py b/resources/tester/test_511_b_calc_path_guard.py new file mode 100644 index 000000000..d4658e56a --- /dev/null +++ b/resources/tester/test_511_b_calc_path_guard.py @@ -0,0 +1,157 @@ +"""PLAN 511-B (요구 320-4) — 계산 창구 `/calc` 가 브라우저 파일 경로를 안 씀. + +못박는 것 + - 단계 격자 경로 칸(`stages.grid.path`) · 측점 격자 칸(`stations.grid`) = 400 · Node 안 부름. + - 격자 = `{model_id, smooth, design}` → 서버가 그 프로젝트 storage 안 DTM 격자 경로로 갈음. + - `dtm_grid_npz` — 다른 프로젝트 모델 · storage 밖 상대 경로(../) · 절대 경로 = LookupError. + - 공유 DB 안 씀(가짜 풀 · 가짜 연결) · 실제 storage 안 씀(tmp_path). + + ./venv/Scripts/python.exe -m pytest resources/tester/test_511_b_calc_path_guard.py -q +""" + +from __future__ import annotations + +import asyncio +import sys +from pathlib import Path + +import pytest +from fastapi import Depends, FastAPI +from fastapi.testclient import TestClient + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from B05_Profile import B05_Profile_Router_Terrain as terrain # noqa: E402 +from B06_Section import B06_Section_Router_Calc as calc_router # noqa: E402 +from common_util import common_util_auth as auth # noqa: E402 +from common_util import common_util_surface_sampler as sampler # noqa: E402 + +PROJECT = "11111111-2222-3333-4444-555555555555" +OTHER = "99999999-2222-3333-4444-555555555555" + + +class _Cursor: + rows: dict = {} + + async def execute(self, sql, args=None): + self.args = args + + async def fetchone(self): + if self.args == (PROJECT,): + return (1,) # 권한 — 회사 1 + return self.rows.get(self.args) + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + +class _Conn(_Cursor): + def cursor(self): + return _Cursor() + + async def __aenter__(self): + return self + + +class _Pool: + def acquire(self): + return _Conn() + + +@pytest.fixture +def storage(tmp_path, monkeypatch): + """프로젝트 둘의 storage(tmp) — 모델 행 = (id, project) → (종류, 상대 경로).""" + roots = {PROJECT: tmp_path / "p1", OTHER: tmp_path / "p2"} + for root in roots.values(): + (root / "models").mkdir(parents=True) + (roots[PROJECT] / "models" / "dtm_las.npz").write_bytes(b"x") + (roots[OTHER] / "models" / "dtm_las.npz").write_bytes(b"x") + + async def stored(_conn, project_id): + return str(project_id) + + monkeypatch.setattr(terrain, "get_project_storage_relative_path", stored) + monkeypatch.setattr(terrain, "resolve_stored_project_path", lambda pid: str(roots[pid])) + _Cursor.rows = { + (7, PROJECT): ("dtm", "models/dtm_las.npz"), + (8, OTHER): ("dtm", "models/dtm_las.npz"), + (9, PROJECT): ("dtm", "../p2/models/dtm_las.npz"), + (10, PROJECT): ("dtm", str(roots[OTHER] / "models" / "dtm_las.npz")), + } + return roots + + +def _npz(model_id: int, project: str = PROJECT) -> Path: + from uuid import UUID + + return asyncio.run(terrain.dtm_grid_npz(_Conn(), UUID(project), model_id, False)) + + +def test_dtm_grid_npz_storage_안만(storage): + assert _npz(7) == (storage[PROJECT] / "models" / "dtm_las.npz").resolve() + for model_id in (8, 9, 10): # 다른 프로젝트 모델 · ../ · 절대 경로 + with pytest.raises(LookupError): + _npz(model_id) + + +@pytest.fixture +def client(monkeypatch, storage): + monkeypatch.setattr(auth, "get_db_pool", lambda: _Pool()) + monkeypatch.setattr(calc_router, "get_db_pool", lambda: _Pool()) + monkeypatch.setattr(sampler, "dtm_grid_file", lambda npz: Path(npz).with_suffix(".bin")) + seen: list = [] + + def fake_run(run, inputs): + seen.append(inputs) + return {"results": {}, "ms": 0} + + monkeypatch.setattr(calc_router, "run_calc_served", fake_run) + app = FastAPI() + app.include_router( + calc_router.router, + dependencies=[Depends(auth.verify_session), Depends(auth.require_project_access)], + ) + app.dependency_overrides[auth.verify_session] = lambda: { + "role": "USER", + "company_id": 1, + "user_id": 7, + } + return TestClient(app), seen + + +def _post(http, inputs, run=("stage.route",)): + return http.post(f"/api/projects/{PROJECT}/calc", json={"run": list(run), "inputs": inputs}) + + +def test_경로_칸_거절(client, storage): + http, seen = client + other = str(storage[OTHER] / "models" / "dtm_las.bin") + bad = [ + {"stages": {"grid": {"path": other, "design": "las"}}}, + {"stages": {"grid": {"path": "../../p2/models/dtm_las.bin"}}}, + {"stages": {"grid": {"model_id": 7, "path": other}}}, + {"stages": {"grid": "C:/Windows/win.ini"}}, + {"stages": {"grid": {"model_id": "7"}}}, + {"stages": {"grid": {"model_id": 7, "design": "../x"}}}, + {"stations": {"grid": other, "line": []}}, + ] + for inputs in bad: + res = _post(http, inputs, ("stations.list",) if "stations" in inputs else ("stage.route",)) + assert res.status_code == 400, inputs + for model_id in (8, 9, 10): # 다른 프로젝트 · ../ · 절대 경로 모델 = 거절 + assert _post(http, {"stages": {"grid": {"model_id": model_id}}}).status_code == 400 + assert seen == [] # Node 안 부름 + + +def test_격자_서버가_정함(client, storage): + http, seen = client + res = _post(http, {"stages": {"grid": {"model_id": 7, "smooth": False, "design": "sheet"}}}) + assert res.status_code == 200, res.text + want = (storage[PROJECT] / "models" / "dtm_las.bin").resolve() + assert seen[-1]["stages"]["grid"] == {"path": str(want), "design": "sheet"} + assert _post(http, {"stages": {"grid": None}}).status_code == 200 # 격자 없음 = 그대로