Merge remote-tracking branches 'origin/main_laptop_1' and 'origin/sub_laptop_7' into sub_laptop_5

B05_Profile/B05_Profile_Engine_Skeleton.py

@@ -20,6 +20,7 @@ from B05_Profile.B05_Profile_Engine_Solver import (
     _cost_surface_signature,
     _load_or_build_cost_surface,
 )
+from common_util.common_util_surface_sampler import check_surface_keys
 from config.config_system import (
     SKELETON_MAIN_RIDGE_ACC_THRESHOLD_CELLS,
     SKELETON_MAIN_VALLEY_ACC_THRESHOLD_CELLS,
@@ -277,6 +278,7 @@ def load_or_build_skeleton(
     project_root: Path, filter_key: str, method: str, smooth: bool
 ) -> dict[str, Any]:
     """스켈레톤을 캐시에서 로드하거나 새로 계산해 저장한다."""
+    check_surface_keys(filter_key, method)
     project_root = Path(project_root)
     models_dir = project_root / _MODELS_SUBDIR
     cache_dir = project_root / _ROUTE_CACHE_SUBDIR

B05_Profile/B05_Profile_Engine_Solver.py

@@ -19,6 +19,7 @@ from B05_Profile.B05_Profile_Engine_Geometry import (
     resample_polyline_2d,
     single_segment_dijkstra,
 )
+from common_util.common_util_surface_sampler import check_surface_keys
 from config.config_system import (
     FOREST_ROAD_MAX_GRADE,
     FOREST_ROAD_MIN_CURVE_R_M,
@@ -45,6 +46,7 @@ _ROUTE_CACHE_SUBDIR = Path("B05_Profile") / "route"
 
 def _load_dtm_grid(models_dir: Path, filter_key: str, smooth: bool):
     """필터의 정규 DTM 격자(x, y, z, valid_mask)를 로드한다."""
+    check_surface_keys(filter_key)
     suffix = "_smooth" if smooth else ""
     dtm_path = models_dir / f"dtm_{filter_key}{suffix}.npz"
     if not dtm_path.exists():
@@ -70,6 +72,7 @@ def _sample_surface_on_grid(
     dtm_z: np.ndarray,
 ) -> np.ndarray:
     """확정 지표면 모델의 표고를 DTM 격자에 샘플링한다(실패 시 DTM으로 폴백)."""
+    check_surface_keys(filter_key, method)
     if method == "dtm":
         return dtm_z
 
@@ -145,6 +148,7 @@ def _sample_surface_on_grid(
 
 def _source_npz_paths(models_dir: Path, filter_key: str, method: str, smooth: bool) -> list[Path]:
     """비용면이 의존하는 소스 모델 파일(존재하는 것만, 안정 순서)."""
+    check_surface_keys(filter_key, method)
     suffix = "_smooth" if smooth else ""
     candidates = [
         models_dir / f"dtm_{filter_key}{suffix}.npz",
@@ -211,6 +215,7 @@ def _load_or_build_cost_surface(
     project_root: Path, models_dir: Path, filter_key: str, method: str, smooth: bool
 ):
     """비용면을 반환한다(서명 일치 시 캐시 재사용, 아니면 재빌드·재캐시)."""
+    check_surface_keys(filter_key, method)
     cache_dir = project_root / _ROUTE_CACHE_SUBDIR
     suffix = "_smooth" if smooth else ""
     cache_path = cache_dir / f"cost_surface_{filter_key}_{method}{suffix}.npz"

B05_Profile/B05_Profile_Schema.py

@@ -4,6 +4,7 @@ from typing import Any, Literal
 
 from pydantic import BaseModel, ConfigDict, Field, model_validator
 
+from common_util.common_util_surface_sampler import SurfaceFilter, SurfaceMethod
 from config.config_system import (
     GRADE_MAIN_DIRECTIONS,
     GRADE_TERRAIN_TYPES,
@@ -63,8 +64,10 @@ class RouteSolveRequest(BaseModel):
 
     model_config = ConfigDict(extra="forbid")
 
-    filter_key: str = Field(description="지면 필터 키 (grid_min_z/csf/pmf)")
-    method: str = Field(default="dtm", description="지표면 표현 (dtm/tin/nurbs/implicit/meshfree)")
+    filter_key: SurfaceFilter = Field(description="지면 필터 키 (grid_min_z/csf/pmf)")
+    method: SurfaceMethod = Field(
+        default="dtm", description="지표면 표현 (dtm/tin/nurbs/implicit/meshfree)"
+    )
     smooth: bool = Field(default=False)
     surface_model_id: int | None = Field(default=None, description="기반 지표면 모델 id")
     algorithm: str = Field(
@@ -269,8 +272,8 @@ class RouteConfirmRequest(BaseModel):
 
     model_config = ConfigDict(extra="forbid")
 
-    filter_key: str | None = None
-    method: str | None = None
+    filter_key: SurfaceFilter | None = None
+    method: SurfaceMethod | None = None
     smooth: bool = False
     surface_model_id: int | None = None
     irregular_stations: list[IrregularStationInput] = Field(default_factory=list)

B05_Profile/B05_Profile_Structures_Schema.py

@@ -164,6 +164,10 @@ def registry_schema_version() -> int:
     return int(_read_definition(*_stamp()).get("schema_version", 1))
 
 
+#: 구조물 id 모양 — 복사본 폴더 이름(`M02_Item_Snapshot.FOLDER_PATTERN`)이 되므로 경로 글자 금지
+STRUCTURE_ID_RULE = r"[0-9A-Za-z][0-9A-Za-z_-]*"
+
+
 class StructureInstance(BaseModel):
     """배치된 구조물 1건.
 
@@ -173,7 +177,7 @@ class StructureInstance(BaseModel):
 
     model_config = ConfigDict(extra="forbid")
 
-    structure_id: str | None = None
+    structure_id: str | None = Field(default=None, pattern=rf"^{STRUCTURE_ID_RULE}$", max_length=64)
     type_id: str
     placement: Placement
     chainage_m: float | None = Field(default=None, ge=0)

M02_MasterTemplete/M02_Item_Snapshot.py

@@ -26,6 +26,7 @@ from pathlib import Path
 from typing import Any
 
 from B05_Profile.B05_Profile_Structures_Repository import load_structures
+from B05_Profile.B05_Profile_Structures_Schema import STRUCTURE_ID_RULE
 from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file
 from common_util.common_util_json import atomic_write_json
 from common_util.common_util_summary_item import GROUP_JOIN, snapshot_folder_name
@@ -51,7 +52,7 @@ LOCAL_FILES = (VALUES_FILE, PICKS_FILE)
 #: 복사본 폴더 이름 — 구조물 id(영문 · 숫자 · `_` `-`) 또는 `pipe@측점` · 옵션 그룹(PLAN 46)은
 #: 뒤에 `__{그룹 id}`
 FOLDER_PATTERN = re.compile(
-    r"^(?:pipe@\d+(?:\.\d+)?(?:__g[0-9a-z]{1,16})?|[0-9A-Za-z][0-9A-Za-z_-]*)$"
+    rf"^(?:pipe@\d+(?:\.\d+)?(?:__g[0-9a-z]{{1,16}})?|{STRUCTURE_ID_RULE})$"
 )
 
 
@@ -69,6 +70,14 @@ def folder_name(owner_id: str) -> str:
     return owner_id.replace("#", "n")
 
 
+def safe_folder(base: Path, name: str) -> Path | None:
+    """`base` 바로 아래 복사본 폴더 — 이름이 규칙 밖 · `base` 밖이면 None(지우기 · 쓰기 막음)."""
+    target = base / name
+    if not FOLDER_PATTERN.match(name) or target.resolve().parent != base.resolve():
+        return None
+    return target
+
+
 def now_text() -> str:
     return datetime.now().isoformat(timespec="seconds")
 
@@ -181,8 +190,8 @@ def migrate_items(project_root: str | Path) -> list[str]:
             continue
         users = [o for o, item in wanted.items() if snapshot_folder_name(item) == child.name]
         for owner in users:
-            target = base / folder_name(owner)
-            if (target / MANIFEST_NAME).is_file():
+            target = safe_folder(base, folder_name(owner))
+            if target is None or (target / MANIFEST_NAME).is_file():
                 continue  # 이미 새 복사본이 있음
             shutil.copytree(child, target, dirs_exist_ok=True)
             origin = manifest.get("원본") or {}
@@ -212,7 +221,11 @@ def sync_snapshots(project_root: str | Path) -> dict[str, list[str]]:
     failed: list[str] = []
     removed: list[str] = []
     for folder, (owner, item) in sorted(wanted.items()):
-        target = base / folder
+        target = safe_folder(base, folder)
+        if target is None:
+            logger.warning("복사본 폴더 이름이 규칙 밖 — 건너뜀: %r", folder)
+            failed.append(folder)
+            continue
         manifest = read_json(target / MANIFEST_NAME)
         if isinstance(manifest, dict) and same_item(manifest, item):
             continue

common_util/common_util_surface_sampler.py

@@ -13,11 +13,21 @@ common_util로 옮겼다 — 두 화면의 종단 Z가 갈라지면 세부유역
 from collections.abc import Callable
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Protocol
+from typing import Literal, Protocol, get_args
 
 import numpy as np
 from scipy.interpolate import RegularGridInterpolator
 
+#: 지면 필터 · 지표면 표현 허용 목록 — 파일 이름에 들어감(B04 `_FILTERS` · `MODEL_METHODS`)
+SurfaceFilter = Literal["classification", "grid_min_z", "csf", "pmf", "ransac"]
+SurfaceMethod = Literal["dtm", "tin", "nurbs", "implicit", "meshfree"]
+
+
+def check_surface_keys(filter_key: str, method: str = "dtm") -> None:
+    """파일 이름 만들기 전 검사 — 허용 목록 밖이면 ValueError(경로 넘기 막음)."""
+    if filter_key not in get_args(SurfaceFilter) or method not in get_args(SurfaceMethod):
+        raise ValueError(f"지면 필터 · 표현 이름이 허용 목록 밖: {filter_key!r} · {method!r}")
+
 
 class SurfaceElevationSampler(Protocol):
     """종·횡단 생성기가 의존하는 최소 표고 조회 인터페이스."""
@@ -129,6 +139,7 @@ def build_surface_sampler(
     models_dir: Path | str, source_filter: str, method: str, smooth: bool
 ) -> SurfaceElevationSampler:
     """1단계 확정 모델을 종·횡단용 일괄 XY 표고 sampler로 연다."""
+    check_surface_keys(source_filter, method)
     models_dir = Path(models_dir)
     smooth_suffix = "_smooth" if smooth and method in {"dtm", "tin"} else ""
     dtm_smooth = models_dir / f"dtm_{source_filter}_smooth.npz"

resources/tester/test_54_22_security.py

@@ -0,0 +1,109 @@
+"""54-22 보안 — H1 구조물 id 경로 넘기 · H2 지면 필터 · 표현 이름 경로 넘기.
+
+못박는 것
+  - H1: 구조물 저장 `structure_id` 는 안전 이름만(어기면 검사 실패 = 422) · 복사본 맞춤은
+    규칙 밖 · `base` 밖 폴더를 지우지도 쓰지도 않음.
+  - H2: `filter_key` · `method` 는 허용 목록만(스키마) · 파일 이름 만드는 곳에서도 한 번 더.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+from pydantic import ValidationError
+
+from B05_Profile.B05_Profile_Engine_Skeleton import load_or_build_skeleton
+from B05_Profile.B05_Profile_Engine_Solver import _load_dtm_grid, _source_npz_paths
+from B05_Profile.B05_Profile_Schema import RouteConfirmRequest, RouteSolveRequest
+from B05_Profile.B05_Profile_Structures_Schema import StructureInstance
+from common_util.common_util_surface_sampler import build_surface_sampler, check_surface_keys
+from M02_MasterTemplete import M02_Item_Snapshot as snap
+
+BAD_IDS = ["../../..", "..", "a/b", "a\\b", "C:/Windows", "/etc", ".hidden", "", "x" * 65]
+BAD_KEYS = ["../other/dtm_csf", "csf/../../x", "C:/x", "", "CSF"]
+
+
+def _structure(structure_id):
+    return StructureInstance(
+        structure_id=structure_id, type_id="x", placement="point", chainage_m=1
+    )
+
+
+@pytest.mark.parametrize("bad", BAD_IDS)
+def test_H1_악성_id_거절(bad):
+    with pytest.raises(ValidationError):
+        _structure(bad)
+
+
+@pytest.mark.parametrize("good", [None, "0c3d99cc5d8f49139c871d662537da42", "abc_1-2", "s1__g1"])
+def test_H1_정상_id_통과(good):
+    assert _structure(good).structure_id == good
+
+
+def test_H1_safe_folder(tmp_path):
+    base = tmp_path / "templates" / "structures"
+    base.mkdir(parents=True)
+    assert snap.safe_folder(base, "abc123") == base / "abc123"
+    assert snap.safe_folder(base, "pipe@40.0__gin") == base / "pipe@40.0__gin"
+    for bad in BAD_IDS[:-1] + [str(tmp_path)]:  # 길이는 스키마 몫
+        assert snap.safe_folder(base, bad) is None
+
+
+def test_H1_맞춤이_base_밖을_안_건드림(tmp_path, monkeypatch):
+    root = tmp_path / "proj"
+    base = snap.structures_dir(root)
+    base.mkdir(parents=True)
+    victim = root / "victim"  # base / "../../victim"
+    victim.mkdir()
+    (victim / "keep.txt").write_text("살아야 함", encoding="utf-8")
+    item = {"layer": "master", "owner": "", "key": "k"}
+    monkeypatch.setattr(
+        snap, "assignments", lambda _root: {"../../victim": item, str(victim): item, "ok1": item}
+    )
+
+    def fake_write(target: Path, *_args):
+        target.mkdir(parents=True, exist_ok=True)
+        return {}
+
+    monkeypatch.setattr(snap, "write_snapshot", fake_write)
+    result = snap.sync_snapshots(root)
+    assert (victim / "keep.txt").is_file()
+    assert set(result["못만듦"]) == {"../../victim", str(victim)}
+    assert result["만듦"] == ["ok1"]
+    assert result["지움"] == []
+
+
+@pytest.mark.parametrize("bad", BAD_KEYS)
+def test_H2_스키마_허용_목록(bad):
+    body = {"bp": {"x": 0, "y": 0}, "ep": {"x": 1, "y": 1}}
+    with pytest.raises(ValidationError):
+        RouteSolveRequest(filter_key=bad, **body)
+    with pytest.raises(ValidationError):
+        RouteSolveRequest(filter_key="csf", method=bad, **body)
+    with pytest.raises(ValidationError):
+        RouteConfirmRequest(filter_key=bad)
+    with pytest.raises(ValidationError):
+        RouteConfirmRequest(method=bad)
+
+
+def test_H2_스키마_정상값_통과():
+    assert RouteConfirmRequest(filter_key="grid_min_z", method="tin").method == "tin"
+    assert RouteConfirmRequest().filter_key is None
+
+
+@pytest.mark.parametrize("bad", BAD_KEYS)
+def test_H2_파일_이름_만드는_곳(bad, tmp_path):
+    with pytest.raises(ValueError):
+        check_surface_keys(bad)
+    with pytest.raises(ValueError):
+        check_surface_keys("csf", bad)
+    with pytest.raises(ValueError):
+        build_surface_sampler(tmp_path, bad, "dtm", False)
+    with pytest.raises(ValueError):
+        _load_dtm_grid(tmp_path, bad, False)
+    with pytest.raises(ValueError):
+        _source_npz_paths(tmp_path, "csf", bad, False)
+    with pytest.raises(ValueError):
+        load_or_build_skeleton(tmp_path, bad, "dtm", False)
+    check_surface_keys("classification", "meshfree")

resources/tester/test_b05_pipe_clearance_switch.py

@@ -71,7 +71,7 @@ def test_request_wins_over_stored():
 def test_schema_defaults_to_none_and_ships_the_flag():
     """스키마 기본은 None(미지정) — 저장값을 덮어쓰지 않는다."""
     request = RouteSolveRequest(
-        filter_key="f",
+        filter_key="csf",
         bp={"x": 0.0, "y": 0.0},
         ep={"x": 100.0, "y": 0.0},
     )