diff --git a/B04_PreProcess/B04_PreProcess_Router_Watershed.py b/B04_PreProcess/B04_PreProcess_Router_Watershed.py index 1ac02733..aec83a12 100644 --- a/B04_PreProcess/B04_PreProcess_Router_Watershed.py +++ b/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)) diff --git a/common_util/common_util_audit.py b/common_util/common_util_audit.py index e0dfd3a5..d1214696 100644 --- a/common_util/common_util_audit.py +++ b/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("호출량 감시 기록 실패") diff --git a/common_util/common_util_auth.py b/common_util/common_util_auth.py index e04d7d7c..0129110b 100644 --- a/common_util/common_util_auth.py +++ b/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,11 +128,17 @@ async def verify_session(request: Request) -> dict[str, Any]: await connection.commit() raise HTTPException(status_code=401, detail="세션이 만료되었습니다.") - await cursor.execute( - "UPDATE sessions SET last_activity_at = CURRENT_TIMESTAMP WHERE id = %s", - (session_id,), - ) - await connection.commit() + # `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,), + ) + await connection.commit() return { "session_id": row[0], "user_id": row[1], diff --git a/common_util/common_util_drainage_context.py b/common_util/common_util_drainage_context.py index a9bda0c5..662be1ed 100644 --- a/common_util/common_util_drainage_context.py +++ b/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]: diff --git a/config/config_db.py b/config/config_db.py index 1fad1da8..f570c4da 100644 --- a/config/config_db.py +++ b/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) diff --git a/config/config_system.py b/config/config_system.py index 1d29b57c..4a60494d 100644 --- a/config/config_system.py +++ b/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")) diff --git a/main.py b/main.py index ce531e7b..9436ea6c 100644 --- a/main.py +++ b/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) + + # ───────────────────────────────────────────────────────────────────────── # 정적 파일 서빙 (프론트엔드) # ─────────────────────────────────────────────────────────────────────────