Aislo/common_util/common_util_auth.py

"""비밀번호, OTP, 세션 쿠키와 권한 검증 공통 기능."""

import hashlib
import re
import secrets
from datetime import datetime, timedelta
from typing import Any

import bcrypt
from fastapi import Depends, HTTPException, Request
from fastapi.responses import Response

from config.config_db import get_db_pool
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,
    SESSION_MAX_AGE_SECONDS,
)


def hash_password(value: str) -> str:
    return bcrypt.hashpw(value.encode(), bcrypt.gensalt(rounds=PASSWORD_BCRYPT_ROUNDS)).decode()


def verify_password(value: str, hashed: str) -> bool:
    return bcrypt.checkpw(value.encode(), hashed.encode())


def generate_otp() -> str:
    return f"{secrets.randbelow(1_000_000):06d}"


def hash_user_agent(user_agent: str) -> str:
    # 브라우저 버전 업데이트(예: Chrome/138 → 139)로 신뢰 브라우저가 풀리지 않도록
    # 버전 숫자를 제거한 뒤 해시한다.
    normalized = re.sub(r"[\d.]+", "", user_agent)
    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()


def generate_device_token() -> str:
    return secrets.token_urlsafe(32)


def hash_device_token(token: str) -> str:
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def get_device_token_cookie(request: Request) -> str | None:
    return request.cookies.get(DEVICE_TOKEN_COOKIE_NAME)


def set_device_token_cookie(response: Response, token: str) -> None:
    response.set_cookie(
        key=DEVICE_TOKEN_COOKIE_NAME,
        value=token,
        max_age=EMAIL_REVERIFY_DAYS * 24 * 60 * 60,
        secure=SESSION_COOKIE_SECURE,
        httponly=True,
        samesite="lax",
        path="/",
    )


def set_session_cookie(response: Response, session_id: str) -> None:
    response.set_cookie(
        key=SESSION_COOKIE_NAME,
        value=session_id,
        max_age=SESSION_MAX_AGE_SECONDS,
        secure=SESSION_COOKIE_SECURE,
        httponly=True,
        samesite="lax",
        path="/",
    )


def delete_session_cookie(response: Response) -> None:
    response.delete_cookie(
        key=SESSION_COOKIE_NAME,
        secure=SESSION_COOKIE_SECURE,
        httponly=True,
        samesite="lax",
        path="/",
    )


async def create_session(user_id: int, user_agent: str) -> str:
    session_id = secrets.token_urlsafe(32)
    expires_at = datetime.utcnow() + timedelta(seconds=SESSION_MAX_AGE_SECONDS)
    pool = get_db_pool()
    async with pool.acquire() as connection, connection.cursor() as cursor:
        await cursor.execute(
            """INSERT INTO sessions (id, user_id, user_agent_hash, expires_at)
               VALUES (%s, %s, %s, %s)""",
            (session_id, user_id, hash_user_agent(user_agent), expires_at),
        )
        await connection.commit()
    return session_id


async def verify_session(request: Request) -> dict[str, Any]:
    session_id = request.cookies.get(SESSION_COOKIE_NAME)
    if not session_id:
        raise HTTPException(status_code=401, detail="로그인이 필요합니다.")

    pool = get_db_pool()
    async with pool.acquire() as connection, connection.cursor() as cursor:
        await cursor.execute(
            """SELECT s.id, s.user_id, s.created_at, s.expires_at, s.last_activity_at,
                      u.email, u.name, u.company_id, u.role, u.is_master, u.status
               FROM sessions s JOIN users u ON u.id = s.user_id WHERE s.id = %s""",
            (session_id,),
        )
        row = await cursor.fetchone()
        if not row:
            raise HTTPException(status_code=401, detail="유효하지 않은 세션입니다.")

        now = datetime.utcnow()
        expired = now > row[3] or (now - row[4]).total_seconds() > SESSION_IDLE_TIMEOUT_SECONDS
        # NO_COMPANY(이메일 인증 완료·회사 미연결)도 로그인 유지 대상. 회사 연결 강제는
        # require_company 의존성에서 별도 처리한다.
        if expired or row[10] not in ("ACTIVE", "NO_COMPANY", "PENDING"):
            await cursor.execute("DELETE FROM sessions WHERE id = %s", (session_id,))
            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,),
            )
            await connection.commit()
    return {
        "session_id": row[0],
        "user_id": row[1],
        "email": row[5],
        "name": row[6],
        "company_id": row[7],
        "role": row[8],
        "is_master": bool(row[9]),
    }


async def get_optional_session(request: Request) -> dict[str, Any] | None:
    """세션이 유효하면 세션 정보를, 없거나 만료되면 None을 반환한다.

    로그인 여부와 무관하게 접근 가능한 엔드포인트(예: 기술지원 문의)에서
    로그인 사용자면 추가 정보를 붙이기 위해 사용한다. 401을 던지지 않는다.
    """
    try:
        return await verify_session(request)
    except HTTPException:
        return None


async def require_master(session: dict[str, Any] = Depends(verify_session)) -> dict[str, Any]:
    if not session["is_master"] and session["role"] != "SYSTEM_ADMIN":
        raise HTTPException(status_code=403, detail="마스터 권한이 필요합니다.")
    return session


async def require_system_admin(
    session: dict[str, Any] = Depends(verify_session),
) -> dict[str, Any]:
    if session["role"] != "SYSTEM_ADMIN":
        raise HTTPException(status_code=403, detail="시스템 관리자 권한이 필요합니다.")
    return session


async def require_company(
    session: dict[str, Any] = Depends(verify_session),
) -> dict[str, Any]:
    if session["company_id"] is None and session["role"] != "SYSTEM_ADMIN":
        raise HTTPException(status_code=403, detail="회사 연결이 필요합니다.")
    return session


async def require_project_access(
    request: Request, session: dict[str, Any] = Depends(verify_session)
) -> dict[str, Any]:
    """URL의 `project_id`가 **내 회사 것인지** 대조한다 (2026-09-01 신설).

    지금까지 B03~B07 라우터는 로그인·회사 소속만 확인하고 URL의 프로젝트가 누구 것인지는
    보지 않았다. 프로젝트 id만 알면 남의 회사 자료를 읽고 쓸 수 있었다 — 특히 B07 도각
    저장은 그 회사의 공용 양식을 덮어쓴다.

    `protected_with_company` 목록에 함께 걸어 두면 경로에 `project_id`가 있는 요청만
    검사하고 나머지는 그대로 지나간다. 검사는 한 곳에서만 돈다.
    """
    project_id = request.path_params.get("project_id")
    if not project_id:
        return session
    if session["role"] == "SYSTEM_ADMIN":
        return session

    pool = get_db_pool()
    async with pool.acquire() as connection, connection.cursor() as cursor:
        await cursor.execute(
            "SELECT company_id FROM projects WHERE id = %s AND deleted_at IS NULL",
            (str(project_id),),
        )
        row = await cursor.fetchone()

    if not row:
        raise HTTPException(status_code=404, detail="프로젝트를 찾을 수 없습니다.")
    if row[0] != session["company_id"]:
        raise HTTPException(status_code=403, detail="다른 회사의 프로젝트입니다.")
    return session