Aislo/common_util/common_util_audit.py

"""시스템 로그 기록·정리 한 곳 (2026-09-06 사용자 확정).

같은 INSERT 문이 여섯 자리에 흩어져 있었고, 대상(무엇에 한 일인가)과 접속 정보(어디서
했는가)는 칸만 있고 값이 비어 있었다. 기록은 이 함수 하나로 모은다.

보관 기간은 `AUDIT_LOG_RETENTION_DAYS`(기본 365일) — 사고 추적에 1년이면 충분하다는
사용자 판단(2026-09-06). 임시 보관함 정리 루프가 돌 때 함께 지운다.
"""

from __future__ import annotations

import logging
from datetime import datetime, timedelta, timezone
from typing import Any

from config.config_db import get_db_pool
from config.config_system import API_CALL_HOURLY_LIMIT, AUDIT_LOG_RETENTION_DAYS

logger = logging.getLogger(__name__)

# 브라우저 문자열은 길다 — 표 칸은 TEXT 지만 화면·로그가 감당할 만큼만 자른다.
_USER_AGENT_MAX = 300


def request_origin(request: Any | None) -> tuple[str | None, str | None]:
    """요청에서 접속 주소와 브라우저 문자열을 꺼낸다. 없으면 (None, None).

    프록시 뒤에서는 `X-Forwarded-For` 의 **첫 주소**가 실제 사용자다.
    """
    if request is None:
        return None, None
    try:
        forwarded = request.headers.get("x-forwarded-for")
        address = forwarded.split(",")[0].strip() if forwarded else None
        if not address and request.client is not None:
            address = request.client.host
        agent = (request.headers.get("user-agent") or "")[:_USER_AGENT_MAX] or None
        return address, agent
    except Exception:  # 기록이 본 작업을 막으면 안 된다.
        return None, None


async def record_audit(
    cursor: Any,
    *,
    actor_id: int,
    action: str,
    resource_type: str | None = None,
    resource_ref: str | int | None = None,
    request: Any | None = None,
) -> None:
    """시스템 로그 한 줄을 적는다 — 호출부의 트랜잭션(cursor)에 얹는다.

    `resource_ref` 는 프로젝트 UUID 처럼 문자열이어도 되고 숫자여도 된다. 숫자면 옛
    `resource_id` 칸에도 같이 넣어 예전 기록과 같은 모양을 지킨다.
    """
    reference = None if resource_ref is None else str(resource_ref)
    numeric = int(resource_ref) if isinstance(resource_ref, int) else None
    address, agent = request_origin(request)
    await cursor.execute(
        """INSERT INTO system_audit_logs
               (user_id, action, resource_type, resource_id, resource_ref, ip_address, user_agent)
           VALUES (%s, %s, %s, %s, %s, %s, %s)""",
        (actor_id, action, resource_type, numeric, reference, address, agent),
    )


async def purge_expired_audit_logs() -> int:
    """보관 기간이 지난 시스템 로그를 지운다. 지운 줄 수를 돌려준다."""
    cutoff = datetime.now(timezone.utc) - timedelta(days=max(1, AUDIT_LOG_RETENTION_DAYS))
    pool = get_db_pool()
    try:
        async with pool.acquire() as connection, connection.cursor() as cursor:
            await cursor.execute("DELETE FROM system_audit_logs WHERE timestamp < %s", (cutoff,))
            removed = cursor.rowcount
            if removed:
                await connection.commit()
                logger.info(
                    "시스템 로그 정리: %d건 삭제 (보관 %d일)", removed, AUDIT_LOG_RETENTION_DAYS
                )
            return removed
    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("호출량 감시 기록 실패")