Merge remote-tracking branch 'origin/sub_laptop_1' into main_laptop_1

B04_PreProcess/B04_PreProcess_Router_Watershed.py

@@ -58,7 +58,7 @@ from common_util.common_util_wamis_station import (
     build_station_rainfall_table,
     is_jeju,
 )
-from config.config_db import get_db_pool
+from config.config_db import get_db_pool, run_with_connection
 from config.config_system import (
     DRAINAGE_ARROW_SPACING_M,
     DRAINAGE_DESIGN_RETURN_PERIOD_YR,
@@ -172,11 +172,13 @@ async def _prepare(project_id: UUID) -> dict[str, Any] | JSONResponse:
     잘라 프로젝트 좌표계로 돌려준다. 원본을 그대로 쓰면 유역·관이 확정 노선 밖에도
     찍히고 좌표계마저 갈린다(2026-09-01 실측: 관은 5179, 노선은 5176이었다).
     """
-    pool = get_db_pool()
-    async with pool.acquire() as connection:
-        stored_path = await get_project_storage_relative_path(connection, project_id)
-        epsg = await get_surface_crs_epsg(connection, project_id, 0)
-        surface_params = await get_surface_confirmation_params(connection, str(project_id))
+    # 셋은 서로 기다릴 이유가 없다 — DB 가 원격이라 순차로 내면 왕복 12ms 가 세 번 붙는다
+    # (2026-09-06 실측). 커넥션을 갈라 같이 보낸다.
+    stored_path, epsg, surface_params = await asyncio.gather(
+        run_with_connection(get_project_storage_relative_path, project_id),
+        run_with_connection(get_surface_crs_epsg, project_id, 0),
+        run_with_connection(get_surface_confirmation_params, str(project_id)),
+    )
 
     project_root = Path(resolve_stored_project_path(stored_path))
     route_file = find_planned_route_file(_route_input_dir(stored_path))

common_util/common_util_audit.py

@@ -14,7 +14,7 @@ from datetime import datetime, timedelta, timezone
 from typing import Any
 
 from config.config_db import get_db_pool
-from config.config_system import AUDIT_LOG_RETENTION_DAYS
+from config.config_system import API_CALL_HOURLY_LIMIT, AUDIT_LOG_RETENTION_DAYS
 
 logger = logging.getLogger(__name__)
 
@@ -82,3 +82,70 @@ async def purge_expired_audit_logs() -> int:
     except Exception:
         logger.exception("시스템 로그 정리 실패")
         return 0
+
+
+# ─────────────────────────────────────────────────────────────────────────
+# 호출량 감시 (2026-09-06 사용자 지시 — 보안)
+# ─────────────────────────────────────────────────────────────────────────
+# 계산 결과는 화면에 나가도 된다는 것이 방침이므로, 남는 위험은 **입력을 바꿔가며 출력을
+# 긁어 모으는 것**이다. 사람이 화면을 쓰는 속도에는 한계가 있다 — 화면 한 번 여는 데 API
+# 가 스무 번쯤 나가므로 한 시간에 수천 번을 넘으면 사람이 아니다.
+#
+# 막지는 않는다. 고객 화면을 끊을 위험이 있고, 어디서 끊을지는 실제 사용 기록을 본 뒤에
+# 정할 일이다. 지금은 **시스템 로그에 한 줄 남겨** 눈에 띄게만 한다(보관 1년).
+_CALL_WINDOW_SECONDS = 3600
+_CALL_COUNTS: dict[tuple[str, int], int] = {}
+_CALL_FLAGGED: set[tuple[str, int]] = set()
+
+
+def _call_bucket(now: float) -> int:
+    return int(now // _CALL_WINDOW_SECONDS)
+
+
+def note_api_call(session_id: str | None) -> bool:
+    """이 세션의 이번 시간대 호출을 하나 센다. 방금 상한을 넘었으면 True.
+
+    True 는 **한 시간대에 한 번만** 나온다 — 로그가 넘치지 않게 한다.
+    """
+    if not session_id:
+        return False
+    from time import time as _now
+
+    bucket = _call_bucket(_now())
+    key = (session_id, bucket)
+    count = _CALL_COUNTS.get(key, 0) + 1
+    _CALL_COUNTS[key] = count
+    if count < API_CALL_HOURLY_LIMIT or key in _CALL_FLAGGED:
+        return False
+    _CALL_FLAGGED.add(key)
+    # 지난 시간대 기록은 버린다 — 오래 켜 둔 서버에서 사전이 무한정 자라지 않게.
+    for old in [k for k in _CALL_COUNTS if k[1] < bucket]:
+        _CALL_COUNTS.pop(old, None)
+        _CALL_FLAGGED.discard(old)
+    return True
+
+
+async def record_call_burst(session_id: str, request: Any | None = None) -> None:
+    """상한을 넘은 세션을 시스템 로그에 남긴다 — 사람이 볼 수 있게만 하고 막지는 않는다."""
+    pool = get_db_pool()
+    try:
+        async with pool.acquire() as connection, connection.cursor() as cursor:
+            await cursor.execute("SELECT user_id FROM sessions WHERE id = %s", (session_id,))
+            row = await cursor.fetchone()
+            if not row:
+                return
+            await record_audit(
+                cursor,
+                actor_id=int(row[0]),
+                action="RATE_ANOMALY",
+                resource_type="api",
+                resource_ref=f"{API_CALL_HOURLY_LIMIT}+/h",
+                request=request,
+            )
+            await connection.commit()
+        logger.warning(
+            "호출량 감시: 한 시간에 %d회를 넘은 세션이 있어 시스템 로그에 남겼습니다.",
+            API_CALL_HOURLY_LIMIT,
+        )
+    except Exception:
+        logger.exception("호출량 감시 기록 실패")

common_util/common_util_auth.py

@@ -15,6 +15,7 @@ from config.config_system import (
     DEVICE_TOKEN_COOKIE_NAME,
     EMAIL_REVERIFY_DAYS,
     PASSWORD_BCRYPT_ROUNDS,
+    SESSION_ACTIVITY_WRITE_INTERVAL_SECONDS,
     SESSION_COOKIE_NAME,
     SESSION_COOKIE_SECURE,
     SESSION_IDLE_TIMEOUT_SECONDS,
@@ -127,6 +128,12 @@ async def verify_session(request: Request) -> dict[str, Any]:
             await connection.commit()
             raise HTTPException(status_code=401, detail="세션이 만료되었습니다.")
 
+        # `last_activity_at` 은 유휴 판정(기본 4시간)에만 쓰는 값이라 초 단위로 정확할 이유가
+        # 없다. 그런데 **요청마다** 쓰고 있어 원격 DB 왕복이 UPDATE + commit 으로 붙었다
+        # (2026-09-06 실측: verify_session 34.4ms = SELECT 10.8 + UPDATE 10.4 + commit 9.5).
+        # 인증이 걸린 모든 요청이 지나는 자리라, 화면 한 번 여는 데 API 가 180번 나가면
+        # 그만큼 곱해진다. 1분에 한 번만 쓴다 — 4시간 판정에는 영향이 없다.
+        if (now - row[4]).total_seconds() >= SESSION_ACTIVITY_WRITE_INTERVAL_SECONDS:
             await cursor.execute(
                 "UPDATE sessions SET last_activity_at = CURRENT_TIMESTAMP WHERE id = %s",
                 (session_id,),

common_util/common_util_drainage_context.py

@@ -37,7 +37,7 @@ from common_util.common_util_route_profile import Z_SOURCE_CSV, resolve_route_pr
 from common_util.common_util_storage import resolve_stored_project_path
 from common_util.common_util_surface_confirmation import get_surface_confirmation_params
 from common_util.common_util_surface_sampler import build_surface_sampler
-from config.config_db import get_db_pool
+from config.config_db import run_with_connection
 
 logger = logging.getLogger(__name__)
 
@@ -63,14 +63,8 @@ class DrainageContext:
     surface_params: dict[str, Any] = field(default_factory=dict)
 
 
-async def _query(repository_call: Callable[..., Any], *args: Any) -> Any:
-    """저장소 함수 하나를 **자기 커넥션**으로 실행한다 — 같이 보내려면 커넥션이 갈려야 한다.
-
-    풀 최대치가 20이라 여기서 서너 개를 동시에 잡아도 여유가 있다(`config_system.DB_POOL_MAX`).
-    """
-    pool = get_db_pool()
-    async with pool.acquire() as connection:
-        return await repository_call(connection, *args)
+# 「자기 커넥션으로 하나씩 돌려 `gather` 로 묶는다」는 정의는 `config_db` 한 곳에 둔다.
+_query = run_with_connection
 
 
 async def load_drainage_context(project_id: UUID) -> tuple[DrainageContext | None, str]:

config/config_db.py

@@ -5,7 +5,8 @@ config_db.py
 비동기 연결 풀 생성 및 관리.
 """
 
-from typing import Optional
+from collections.abc import Callable
+from typing import Any, Optional
 
 import aiomysql
 
@@ -58,3 +59,18 @@ def get_db_pool() -> aiomysql.Pool:
     if not db_pool:
         raise RuntimeError("DB pool not initialized. Call init_db_pool() first.")
     return db_pool
+
+
+async def run_with_connection(repository_call: Callable[..., Any], *args: Any) -> Any:
+    """저장소 함수 하나를 **자기 커넥션**으로 실행한다 — `asyncio.gather` 로 묶기 위한 것.
+
+    DB 가 원격이라 질의 하나가 곧 왕복 약 12ms 다(2026-09-06 실측). 서로 기다릴 이유가 없는
+    읽기를 한 커넥션에서 순차로 내면 그 왕복이 그대로 더해진다. 커넥션을 갈라 같이 보내면
+    가장 느린 하나의 시간만 든다. 풀 최대치는 `DB_POOL_MAX`(기본 20).
+
+    ⚠ **읽기에만 쓸 것.** 순서가 필요한 쓰기(한 트랜잭션 안의 UPDATE 들)를 이걸로 묶으면
+    커넥션이 갈려 트랜잭션이 깨진다.
+    """
+    pool = get_db_pool()
+    async with pool.acquire() as connection:
+        return await repository_call(connection, *args)

config/config_system.py

@@ -100,6 +100,11 @@ TEMP_UPLOAD_DIR_NAME = os.getenv("TEMP_UPLOAD_DIR_NAME", "tmp")
 TEMP_UPLOAD_RETENTION_DAYS = int(os.getenv("TEMP_UPLOAD_RETENTION_DAYS", "30"))
 # 시스템 로그(누가 무엇을 했나) 보관 기간 — 사고 추적에 1년 (2026-09-06 사용자 확정).
 AUDIT_LOG_RETENTION_DAYS = int(os.getenv("AUDIT_LOG_RETENTION_DAYS", "365"))
+
+# 한 세션이 한 시간에 부를 수 있는 API 횟수 — 넘으면 **막지 않고 시스템 로그에만** 남긴다
+# (2026-09-06 사용자 지시). 화면 한 번 여는 데 약 180회가 나가므로, 사람이 쉬지 않고
+# 화면을 열어도 한 시간에 수천 회다. 그 몇 배를 넘으면 사람이 아니라고 본다.
+API_CALL_HOURLY_LIMIT = int(os.getenv("API_CALL_HOURLY_LIMIT", "20000"))
 TEMP_UPLOAD_CLEANUP_INTERVAL_HOURS = int(os.getenv("TEMP_UPLOAD_CLEANUP_INTERVAL_HOURS", "6"))
 MERGE_TIMEOUT_SECONDS = int(os.getenv("MERGE_TIMEOUT_SECONDS", "3600"))
 SEND_ANALYSIS_COMPLETION_EMAIL = (
@@ -233,6 +238,12 @@ SESSION_COOKIE_NAME = os.getenv("SESSION_COOKIE_NAME", "session_id")
 DEVICE_TOKEN_COOKIE_NAME = os.getenv("DEVICE_TOKEN_COOKIE_NAME", "device_token")
 SESSION_MAX_AGE_SECONDS = int(os.getenv("SESSION_MAX_AGE_SECONDS", "43200"))
 SESSION_IDLE_TIMEOUT_SECONDS = int(os.getenv("SESSION_IDLE_TIMEOUT_SECONDS", "14400"))
+
+# `sessions.last_activity_at` 을 다시 쓰는 최소 간격(초). 유휴 판정이 4시간 단위라 초 단위로
+# 정확할 이유가 없고, 요청마다 쓰면 원격 DB 왕복이 약 20ms 씩 붙는다(2026-09-06 실측).
+SESSION_ACTIVITY_WRITE_INTERVAL_SECONDS = int(
+    os.getenv("SESSION_ACTIVITY_WRITE_INTERVAL_SECONDS", "60")
+)
 SESSION_COOKIE_SECURE = os.getenv("SESSION_COOKIE_SECURE", "True").lower() == "true"
 PASSWORD_BCRYPT_ROUNDS = int(os.getenv("PASSWORD_BCRYPT_ROUNDS", "12"))
 OTP_VALID_MINUTES = int(os.getenv("EMAIL_OTP_VALID_MINUTES", "5"))

main.py

@@ -54,6 +54,7 @@ from B06_Section.B06_Section_Router_Confirm import (
 from B07_DesignDetail.B07_DesignDetail_Router import router as b07_design_router
 from B07_DesignDetail.B07_DesignDetail_Router_Frame import router as b07_frame_router
 from B08_Quantity.B08_Quantity_Router import router as b08_quantity_router
+from common_util.common_util_audit import note_api_call, record_call_burst
 from common_util.common_util_auth import (
     require_company,
     require_project_access,
@@ -72,6 +73,7 @@ from config.config_system import (
     LOG_LEVEL,
     SERVER_HOST,
     SERVER_PORT,
+    SESSION_COOKIE_NAME,
     STATIC_DIR,
     STATIC_URL,
 )
@@ -369,6 +371,22 @@ app.add_middleware(
     compresslevel=_COMPRESS_LEVEL,
 )
 
+
+# ─────────────────────────────────────────────────────────────────────────
+# 호출량 감시 (2026-09-06 사용자 지시 — 보안)
+# ─────────────────────────────────────────────────────────────────────────
+# 계산 결과는 화면에 나가도 된다는 방침이라, 남는 위험은 입력을 바꿔가며 출력을 긁어 모으는
+# 것이다. **막지는 않고** 상한을 넘은 세션만 시스템 로그에 한 줄 남긴다 — 자세한 이유는
+# `common_util_audit.note_api_call` 머리 참조. 세는 값은 메모리에 있어 요청당 비용이 없다.
+@app.middleware("http")
+async def watch_call_volume(request, call_next):  # type: ignore[no-untyped-def]
+    if request.url.path.startswith("/api/"):
+        session_id = request.cookies.get(SESSION_COOKIE_NAME)
+        if note_api_call(session_id) and session_id:
+            asyncio.create_task(record_call_burst(session_id, request))
+    return await call_next(request)
+
+
 # ─────────────────────────────────────────────────────────────────────────
 # 정적 파일 서빙 (프론트엔드)
 # ─────────────────────────────────────────────────────────────────────────