8장. 보안 - 안전한 도구 실행

따라하기 1 — 도구 모듈 haru_tools.py 만들기

목표

본인 확인, 마스킹, 예외 처리를 갖춘 도구 여섯 개를 한 파일에 담습니다. 이 파일 haru_tools.py는 이번 장에서 끝나지 않고 9장부터 마지막 장까지 계속 불러 씁니다. 이 절에서는 파일을 만들고 구조를 읽습니다. 실행해서 확인하는 일은 다음 절에서 합니다.

---

0. 실습 준비

VS Code에서 haru-market 폴더를 열고 터미널에서 환경을 켭니다.

$ conda activate myenv

이번 장은 새로 설치할 것이 없습니다. 이번 장의 파일은 6장·7장의 파일을 불러 쓰지 않습니다. config.py와 data 폴더만 있으면 됩니다.

이번 장에서는 파일을 두 개 만듭니다. 순서가 중요합니다.

두 번째 파일이 첫 번째 파일을 불러 쓰므로 haru_tools.py를 먼저 만듭니다.

---

1. 파일 만들기

haru-market 폴더 맨 위에 새 파일을 만듭니다. 이번에는 이름이 lesson으로 시작하지 않습니다.

haru_tools.py

파일 이름을 정확히 haru_tools.py로 합니다. 뒤의 장들이 from haru_tools import …로 이 이름을 찾습니다. 밑줄(_)이 하이픈(-)이 되거나 대소문자가 다르면 찾지 못합니다.

아래 코드 전체를 복사해 붙여 넣고 저장합니다.

# -*- coding: utf-8 -*-
"""하루마켓 도구 계층 (8장 산출물) — 이후 모든 장이 이 모듈을 재사용한다.

설계 원칙
  1) 본인 확인: 주문 조회는 '로그인한 고객 자신'의 주문만 허용한다.
     남의 주문번호에는 없는 주문번호와 같은 응답을 돌려준다.
  2) 개인정보 마스킹: 이름·전화번호는 마스킹해서 모델에게 전달한다.
     (모델 입력으로 들어간 데이터는 답변에 노출될 수 있다고 가정한다)
  3) 읽기 도구 vs 쓰기 도구 구분: 쓰기(티켓 생성 등)는 명시적으로 표시하고
     되돌릴 수 없는 것(환불 실행)은 도구로 만들지 않는다 → 23장 승인 게이트.
  4) 도구는 절대 예외로 죽지 않는다: 항상 dict 를 반환해 모델이 상황을 알게 한다.
"""
import json
from datetime import datetime
import pandas as pd

from config import DATA_DIR, ROOT

ORDERS = pd.read_csv(DATA_DIR / "orders.csv", dtype=str)
PRODUCTS = pd.read_csv(DATA_DIR / "products.csv", dtype=str)
CUSTOMERS = pd.read_csv(DATA_DIR / "customers.csv", dtype=str)

TICKETS_PATH = ROOT / "memory_store" / "tickets.json"


# ── 개인정보 마스킹 유틸 ─────────────────────────────────────────────
def mask_name(name: str) -> str:
    """김민준 → 김*준"""
    if len(name) <= 1:
        return name
    return name[0] + "*" * (len(name) - 2) + name[-1] if len(name) > 2 else name[0] + "*"


def mask_phone(phone: str) -> str:
    """010-1234-5678 → 010-****-5678"""
    parts = phone.split("-")
    if len(parts) == 3:
        return f"{parts[0]}-****-{parts[2]}"
    return "****"


# ── 세션 컨텍스트 — "지금 로그인한 고객이 누구인가" ───────────────────
# 실제 서비스라면 웹 세션/JWT 에서 온다. 수업에서는 명시적으로 지정한다.
class CustomerSession:
    def __init__(self, customer_id: str):
        row = CUSTOMERS[CUSTOMERS["customer_id"] == customer_id]
        if row.empty:
            raise ValueError(f"존재하지 않는 고객: {customer_id}")
        self.customer_id = customer_id
        self.name = row.iloc[0]["name"]
        self.grade = row.iloc[0]["grade"]
        self.points = int(row.iloc[0]["points"])

    def __repr__(self):
        return f"CustomerSession({self.customer_id}, {mask_name(self.name)})"


def make_tools(session: CustomerSession):
    """로그인한 고객 세션에 묶인 도구 세트를 생성한다.

    포인트: 도구 함수의 인자에 customer_id 가 없다.
    '누구의 주문인가'는 모델이 정하는 값이 아니라 세션이 정하는 값이다.
    모델에게 권한 판단을 맡기지 않는다 — 권한 경계는 코드에 있다.
    """

    # ---------- 읽기 도구 ----------
    def get_my_orders() -> list[dict]:
        """로그인한 고객 본인의 최근 주문 목록을 조회한다.
        고객이 주문번호를 모를 때 먼저 이 도구로 주문을 찾는다.
        """
        rows = ORDERS[ORDERS["customer_id"] == session.customer_id]
        rows = rows.sort_values("ordered_at", ascending=False).head(10)
        if rows.empty:
            return [{"found": False, "message": "주문 내역이 없습니다."}]
        return [
            {"order_id": r["order_id"], "product_name": r["product_name"],
             "option": r["option"], "order_amount": r["order_amount"],
             "ordered_at": r["ordered_at"], "status": r["order_status"]}
            for _, r in rows.iterrows()
        ]

    def get_order_status(order_id: str) -> dict:
        """주문번호로 본인 주문의 상태·배송 정보를 조회한다.

        Args:
            order_id: 하루마켓 주문번호. 'HR'로 시작한다. 예: HR20260701023
        """
        # 없는 주문번호와 남의 주문번호에 돌려줄 응답은 하나다.
        # 둘이 다르면 그 차이만으로 주문번호가 실제로 있는지 알아낼 수 있다.
        not_found = {"found": False,
                     "message": f"주문번호 {order_id} 를 찾을 수 없습니다. "
                                "번호를 다시 확인해 주세요."}
        row = ORDERS[ORDERS["order_id"] == order_id]
        if row.empty:
            return not_found
        r = row.iloc[0]
        # 본인 확인 — 다른 고객의 주문이면 없는 주문과 같은 응답을 돌려준다
        if r["customer_id"] != session.customer_id:
            return not_found
        return {
            "found": True, "order_id": r["order_id"],
            "product_name": r["product_name"], "option": r["option"],
            "quantity": r["quantity"], "order_amount": r["order_amount"],
            "ordered_at": r["ordered_at"], "status": r["order_status"],
            "courier": r["courier"] if isinstance(r["courier"], str) else "",
            "tracking_no": r["tracking_no"] if isinstance(r["tracking_no"], str) else "",
            "shipped_at": r["shipped_at"] if isinstance(r["shipped_at"], str) else "",
            "delivered_at": r["delivered_at"] if isinstance(r["delivered_at"], str) else "",
        }

    def search_products(keyword: str) -> list[dict]:
        """상품명·카테고리·설명에서 키워드로 판매 상품을 검색한다.
        가격·옵션·특징 등 '상품 자체' 문의에 사용한다. 주문 조회에는 사용하지 않는다.

        Args:
            keyword: 검색어. 예: '텀블러', '이불커버'
        """
        try:
            mask = (
                PRODUCTS["product_name"].str.contains(keyword, na=False)
                | PRODUCTS["category"].str.contains(keyword, na=False)
                | PRODUCTS["description"].str.contains(keyword, na=False)
            )
            hits = PRODUCTS[mask].head(5)
        except Exception as e:                      # 어떤 입력에도 죽지 않는다
            return [{"found": False, "error": str(e)}]
        if hits.empty:
            return [{"found": False, "message": f"'{keyword}' 검색 결과가 없습니다."}]
        return [
            {"product_id": r["product_id"], "product_name": r["product_name"],
             "category": r["category"], "price": r["price"], "options": r["options"],
             "rating": r["rating"], "description": r["description"]}
            for _, r in hits.iterrows()
        ]

    def get_my_membership() -> dict:
        """로그인한 고객 본인의 하루클럽 멤버십 등급과 보유 하루포인트를 조회한다.
        등급별 혜택·산정 기준 등 '정책 내용'은 이 도구가 아니라 정책 문서에서 확인한다.
        """
        return {"customer_id": session.customer_id,
                "name": mask_name(session.name),
                "grade": session.grade, "points": session.points}

    def check_stock(product_id: str) -> dict:
        """상품ID로 재고 수량을 확인한다. 교환 가능 여부·재입고 문의에 사용한다.

        Args:
            product_id: 상품ID. 'P'로 시작한다. 예: P008
        """
        row = PRODUCTS[PRODUCTS["product_id"] == product_id]
        if row.empty:
            return {"found": False, "message": f"상품 {product_id} 를 찾을 수 없습니다."}
        r = row.iloc[0]
        stock = int(r["stock"])
        return {"found": True, "product_id": product_id,
                "product_name": r["product_name"], "stock": stock, "in_stock": stock > 0}

    # ---------- 쓰기 도구 (부작용 있음 — 파일에 기록된다) ----------
    def create_ticket(category: str, summary: str, urgency: str = "보통") -> dict:
        """[쓰기] 상담원 처리가 필요한 문의를 티켓으로 등록한다.
        AI가 해결하지 못한 문의, 상담원 연결 요청, 계정·결제 문제에 사용한다.

        Args:
            category: 티켓 분류. 제품문의/주문배송조회/환불교환/계정결제/
                상담원연결/기타 중 하나
            summary: 상담원이 볼 한 줄 요약 (고객 상황 + 요청 사항)
            urgency: 낮음/보통/높음
        """
        TICKETS_PATH.parent.mkdir(exist_ok=True)
        tickets = []
        if TICKETS_PATH.exists():
            tickets = json.loads(TICKETS_PATH.read_text(encoding="utf-8"))
        ticket = {
            "ticket_id": f"T{len(tickets) + 1:05d}",
            "created_at": datetime.now().strftime("%Y-%m-%d %H:%M"),
            "customer_id": session.customer_id,
            "customer_name": mask_name(session.name),   # 마스킹된 이름만 기록
            "category": category, "summary": summary,
            "urgency": urgency, "status": "접수",
        }
        tickets.append(ticket)
        TICKETS_PATH.write_text(json.dumps(tickets, ensure_ascii=False, indent=2),
                                encoding="utf-8")
        return {"created": True, "ticket_id": ticket["ticket_id"],
                "message": f"티켓 {ticket['ticket_id']} 접수 완료. "
                           "상담원이 순차적으로 연락드립니다."}

    return {
        "get_my_orders": get_my_orders,
        "get_order_status": get_order_status,
        "get_my_membership": get_my_membership,
        "search_products": search_products,
        "check_stock": check_stock,
        "create_ticket": create_ticket,
    }

---

2. 코드에서 볼 곳

파일은 위에서부터 네 덩어리입니다.

make_tools가 돌려주는 도구입니다.

여섯 도구 어디에도 customer_id 인자가 없습니다.

(1) 도구가 함수 안에 들어 있다

def make_tools(session: CustomerSession):
    ...
    # ---------- 읽기 도구 ----------
    def get_my_orders() -> list[dict]:
        """로그인한 고객 본인의 최근 주문 목록을 조회한다.
        고객이 주문번호를 모를 때 먼저 이 도구로 주문을 찾는다.
        """
        rows = ORDERS[ORDERS["customer_id"] == session.customer_id]

7장까지는 도구 함수가 파일 맨 왼쪽에서 시작했습니다. 여기서는 make_tools 안에 들여쓰기 되어 있습니다. 그래서 안쪽 함수들이 바깥의 session을 쓸 수 있습니다. 앞에서 본 클로저입니다.

(2) 본인 확인은 두 줄이다

not_found = {"found": False,
             "message": f"주문번호 {order_id} 를 찾을 수 없습니다. "
                        "번호를 다시 확인해 주세요."}
row = ORDERS[ORDERS["order_id"] == order_id]
if row.empty:
    return not_found
r = row.iloc[0]
# 본인 확인 — 다른 고객의 주문이면 없는 주문과 같은 응답을 돌려준다
if r["customer_id"] != session.customer_id:
    return not_found

주문에 적힌 고객과 세션의 고객이 다르면, 주문 내용을 담기 전에 돌아갑니다. 이때 돌려주는 값은 주문번호가 없을 때와 같은 not_found 입니다. 두 경우의 응답이 다르면 그 차이로 주문번호가 실제로 있는지 알아낼 수 있기 때문입니다. 6장의 get_order_status와 track_shipping 두 도구는 여기서 하나로 합쳐졌고, 택배사와 송장번호도 이 도구가 함께 돌려줍니다.

(3) 목록 도구는 요약만 준다

return [
    {"order_id": r["order_id"], "product_name": r["product_name"],
     "option": r["option"], "order_amount": r["order_amount"],
     "ordered_at": r["ordered_at"], "status": r["order_status"]}
    for _, r in rows.iterrows()
]

get_my_orders의 결과에는 택배사와 송장번호가 없습니다. 목록은 "어떤 주문이 있는가"만 알려 주고, 자세한 것은 get_order_status로 한 건씩 조회합니다. 이 나눔이 9장에서 다시 중요해집니다.

(4) 도구를 딕셔너리로 돌려준다

return {
    "get_my_orders": get_my_orders,
    "get_order_status": get_order_status,
    ...
    "create_ticket": create_ticket,
}

이름으로 꺼내 쓸 수 있게 딕셔너리로 돌려줍니다. 코드에서 직접 부를 때는 tools["get_my_orders"](), 모델에게 넘길 때는 list(tools.values())입니다.

(5) 티켓 파일의 위치

TICKETS_PATH = ROOT / "memory_store" / "tickets.json"

ROOT는 config.py가 알려 주는 프로젝트 폴더입니다. 티켓은 haru-market/memory_store/tickets.json에 저장됩니다. 지금은 이 폴더가 없습니다. create_ticket이 처음 불릴 때 아래 줄이 폴더를 만듭니다.

TICKETS_PATH.parent.mkdir(exist_ok=True)

---

3. 실행하기

이 파일에는 if __name__ == "__main__": 부분이 없습니다. 불러 쓰라고 만든 파일이기 때문입니다. 그래도 한 번 실행해 오류가 없는지 확인합니다.

$ python haru_tools.py

아무것도 출력되지 않고 프롬프트로 돌아오면 성공입니다. 세 개의 CSV를 읽고 함수와 클래스를 정의한 뒤 끝난 것입니다.

---

4. 무엇을 관찰했나

아직 눈에 보이는 결과는 없습니다. 이 절에서 확인한 것은 구조입니다.

• 도구 여섯 개가 make_tools(session) 안에 있고, 그 안에서만 session을 씁니다.
• 고객을 가리키는 값은 인자로 받는 곳이 한 군데도 없습니다.
• 쓰기 도구는 create_ticket 하나이고, 환불이나 취소를 실행하는 도구는 없습니다.
• 이름을 내보내는 자리(get_my_membership, 티켓 기록, __repr__)에는 모두 mask_name이 걸려 있습니다.

---

5. 지금 폴더의 모습

haru-market/
├── config.py
├── data/
├── haru_tools.py                 ← 이번 장 (이후 모든 장이 불러 쓴다)
├── …
├── lesson06_order_tools.py
└── lesson07_multi_tools.py

---

핵심 정리

• haru_tools.py는 불러 쓰는 모듈입니다. 실행하면 아무것도 출력하지 않는 것이 정상입니다.
• 도구는 make_tools(session) 안에 정의되어 세션을 기억합니다.
• 도구 여섯 개 중 읽기 다섯, 쓰기 하나입니다. 고객 ID를 인자로 받는 도구는 없습니다.
• get_my_orders는 요약만, get_order_status는 한 건의 상세를 돌려줍니다.
• memory_store 폴더는 티켓이 처음 접수될 때 자동으로 만들어집니다.