From c8ca79063c93462f7b92ddf8f4ddc195b5b313bb Mon Sep 17 00:00:00 2001 From: umsangdon Date: Tue, 6 Oct 2026 12:22:08 +0900 Subject: [PATCH] =?UTF-8?q?fix(B05):=2054-22=20H1=20=C2=B7=20H2=20?= =?UTF-8?q?=EA=B2=BD=EB=A1=9C=20=EB=84=98=EA=B8=B0=20=EB=A7=89=EC=9D=8C=20?= =?UTF-8?q?=E2=80=94=20=EA=B5=AC=EC=A1=B0=EB=AC=BC=20id=20=EC=95=88?= =?UTF-8?q?=EC=A0=84=20=EC=9D=B4=EB=A6=84=20=C2=B7=20=EC=A7=80=EB=A9=B4=20?= =?UTF-8?q?=ED=95=84=ED=84=B0=20=C2=B7=20=ED=91=9C=ED=98=84=20=ED=97=88?= =?UTF-8?q?=EC=9A=A9=20=EB=AA=A9=EB=A1=9D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - H1: 구조물 저장 structure_id 를 안전 이름 규칙(FOLDER_PATTERN 과 한 벌)으로 검사 · 어기면 422 - H1: 복사본 맞춤에서 한 번 더 — 규칙 밖 · base 밖 폴더는 지우기 · 쓰기 건너뜀(못만듦) - H2: filter_key · method 를 허용 목록(Literal)으로 · 파일 이름 만드는 곳(표고 sampler · 비용면 · 스켈레톤)에서 한 번 더 - 시험: test_54_22_security 26 통과 · 관 간격 시험 가짜 필터 이름을 허용 값으로 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EYJJRCfHfdCGNhmXTjQKV7 --- B05_Profile/B05_Profile_Engine_Skeleton.py | 2 + B05_Profile/B05_Profile_Engine_Solver.py | 5 + B05_Profile/B05_Profile_Schema.py | 11 +- B05_Profile/B05_Profile_Structures_Schema.py | 6 +- M02_MasterTemplete/M02_Item_Snapshot.py | 21 +++- common_util/common_util_surface_sampler.py | 13 ++- resources/tester/test_54_22_security.py | 109 ++++++++++++++++++ .../tester/test_b05_pipe_clearance_switch.py | 2 +- 8 files changed, 158 insertions(+), 11 deletions(-) create mode 100644 resources/tester/test_54_22_security.py diff --git a/B05_Profile/B05_Profile_Engine_Skeleton.py b/B05_Profile/B05_Profile_Engine_Skeleton.py index 385e31a78..947981955 100644 --- a/B05_Profile/B05_Profile_Engine_Skeleton.py +++ b/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 diff --git a/B05_Profile/B05_Profile_Engine_Solver.py b/B05_Profile/B05_Profile_Engine_Solver.py index 804d9c8de..4058c979b 100644 --- a/B05_Profile/B05_Profile_Engine_Solver.py +++ b/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" diff --git a/B05_Profile/B05_Profile_Schema.py b/B05_Profile/B05_Profile_Schema.py index d2790ec5b..e6407c988 100644 --- a/B05_Profile/B05_Profile_Schema.py +++ b/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) diff --git a/B05_Profile/B05_Profile_Structures_Schema.py b/B05_Profile/B05_Profile_Structures_Schema.py index e7c8db289..c289c3363 100644 --- a/B05_Profile/B05_Profile_Structures_Schema.py +++ b/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) diff --git a/M02_MasterTemplete/M02_Item_Snapshot.py b/M02_MasterTemplete/M02_Item_Snapshot.py index d7e67022f..9df1bb0cb 100644 --- a/M02_MasterTemplete/M02_Item_Snapshot.py +++ b/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 diff --git a/common_util/common_util_surface_sampler.py b/common_util/common_util_surface_sampler.py index 597243f2c..fddba295a 100644 --- a/common_util/common_util_surface_sampler.py +++ b/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" diff --git a/resources/tester/test_54_22_security.py b/resources/tester/test_54_22_security.py new file mode 100644 index 000000000..42cf375a6 --- /dev/null +++ b/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") diff --git a/resources/tester/test_b05_pipe_clearance_switch.py b/resources/tester/test_b05_pipe_clearance_switch.py index 7daef7d92..4787d6a9f 100644 --- a/resources/tester/test_b05_pipe_clearance_switch.py +++ b/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}, )