feat(security): 호출량 감시 — 상한 넘은 세션만 시스템 로그에 표시

계산 결과는 화면에 나가도 된다는 방침이라, 남는 위험은 입력을 바꿔가며 출력을
긁어 모으는 것임. 사람이 화면을 쓰는 속도에는 한계가 있음 — 화면 한 번 여는 데
API 가 약 180회 나가므로 한 시간에 2만 회를 넘으면 사람이 아니라고 봄.

- 막지 않음. 고객 화면을 끊을 위험이 있고 어디서 끊을지는 실제 사용 기록을 본 뒤
  정할 일임. 지금은 시스템 로그(보관 1년)에 RATE_ANOMALY 한 줄만 남김.
- 세는 값은 메모리(세션·시간대별)라 요청당 DB 비용 0. 상한을 넘은 그 순간에만
  세션 주인을 한 번 조회해 기록하고, 같은 시간대에는 다시 안 남김.
- 지난 시간대 기록은 버려 사전이 무한정 자라지 않게 함.
- 상한은 API_CALL_HOURLY_LIMIT(기본 20000, .env 로 조정).

빌드 껍데기는 이미 단단해 손대지 않음 — 실측 확인: 배포 산출물에 소스맵 0개,
식별자 뭉개짐(import{$ as e,C as t,...}), computeCrossDesign 등 원본 함수명
검색 결과 0건. 난독화 도구 추가는 얻는 것 대비 비용이 큼.

시험 tmp/tests/test_call_volume_watch.py 5건. 전체 484 통과.
실서버 확인 — 정상 사용 흐름에서 감시 로그 0건.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

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("호출량 감시 기록 실패")

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 = (

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)
+
+
 # ─────────────────────────────────────────────────────────────────────────
 # 정적 파일 서빙 (프론트엔드)
 # ─────────────────────────────────────────────────────────────────────────