저는 최근 이커머스 플랫폼에서 AI 고객 서비스 시스템을 구축하면서 예상치 못한 보안 문제를 경험했습니다. 사용자가 상품 검색 AI 비서에게 "北京市の天気"라는 텍스트를 입력했고, 이 값이 Function Calling의 파라미터로 직접 전달되어 데이터베이스 查询에 오류가 발생한 것이죠. 이 경험이 오늘 포스팅의 출발점입니다. Function Calling은 AI 응답의 정확도를 크게 높이지만, 잘못된 파라미터 검증은 치명적인 보안 취약점이 될 수 있습니다.
Function Calling 보안 위험 이해
Function Calling은 AI가 외부 도구를 호출할 수 있게 하지만, 이것은 곧 외부 입력이 시스템 내부로 유입되는 통로가 됩니다. 대표적인 보안 위협 세 가지를 먼저 이해해야 합니다.
1. 파라미터 주입 공격
악의적인 사용자가 Function Calling 스키마의 파라미터에 특수 문자나 SQL/NoSQL 문법을 삽입하여 의도하지 않은 동작을 유도합니다. 예를 들어, 상품명 검색 필드에 "; DROP TABLE users; -- 같은 문자열을 삽입하는 것이죠.
2. 타입 검증 실패
AI가 생성한 파라미터의 타입이 예상과 다를 수 있습니다. 문자열이어야 할 곳에 배열이 오거나, 정수여야 할 곳에 부동소수점이 오는 경우가 발생합니다. 이는 런타임 오류와 예상치 못한 시스템 상태로 이어집니다.
3. 범위 밖 값 유입
열거형(enum)이나 허용 가능한 범위가 정의된 파라미터에, 해당하지 않는 값이 전달될 수 있습니다. 예컨대 페이지 번호에 음수나超大 값이 들어오는 것이죠.
안전한 Function Calling 구현
제가 실제 프로젝트에서 검증착용한 아키텍처를 공유합니다. HolySheep AI의 unified endpoint를 사용하면 다양한 모델을 동일한 검증 로직으로 처리할 수 있어 매우 편리합니다.
핵심 검증 클래스 구현
import json
import re
from typing import Any, Callable, Dict, List, Optional, get_type_hints
from dataclasses import dataclass
from enum import Enum
class ValidationError(Exception):
"""파라미터 검증 실패 시 발생하는 예외"""
def __init__(self, param_name: str, expected: str, received: Any):
self.param_name = param_name
self.expected = expected
self.received = received
super().__init__(
f"파라미터 검증 실패: '{param_name}' - "
f"예상: {expected}, 수신: {type(received).__name__}={received}"
)
class ParameterValidator:
"""Function Calling 파라미터 검증기"""
def __init__(self):
self.errors: List[str] = []
def validate_type(self, value: Any, expected_type: type, param_name: str) -> bool:
"""타입 검증 - 예상치 못한 타입 수신 시 False 반환"""
if expected_type == int and isinstance(value, float):
# 정수 타입 기대 시 float의 소수점 부분 확인
if value != int(value):
self.errors.append(
f"'{param_name}': 정수가 필요합니다. 현재 {value}"
)
return False
return True
type_map = {
str: (str,),
int: (int,),
float: (int, float), # int는 float의 하위 타입으로 처리
bool: (bool,),
list: (list, tuple),
dict: (dict,)
}
if expected_type not in type_map:
return True
if not isinstance(value, type_map[expected_type]):
self.errors.append(
f"'{param_name}': {expected_type.__name__} 타입 필요. "
f"현재 {type(value).__name__}"
)
return False
return True
def validate_enum(self, value: Any, enum_class: type, param_name: str) -> bool:
"""열거형 검증"""
try:
if isinstance(value, str):
enum_values = [e.value if hasattr(e, 'value') else e for e in enum_class]
if value not in enum_values:
self.errors.append(
f"'{param_name}': 허용되지 않은 값 '{value}'. "
f"허용 목록: {enum_values}"
)
return False
except Exception as e:
self.errors.append(f"'{param_name}' enum 검증 실패: {str(e)}")
return False
return True
def validate_range(
self,
value: numeric,
param_name: str,
min_val: Optional[float] = None,
max_val: Optional[float] = None
) -> bool:
"""숫자 범위 검증"""
if min_val is not None and value < min_val:
self.errors.append(
f"'{param_name}': 최소값 {min_val} 이상 필요. 현재 {value}"
)
return False
if max_val is not None and value > max_val:
self.errors.append(
f"'{param_name}': 최대값 {max_val} 이하여야 합니다. 현재 {value}"
)
return False
return True
def sanitize_string(self, value: str, param_name: str, max_length: int = 1000) -> str:
"""문자열 정화 - 위험한 패턴 제거"""
# 제어 문자 및 다국어 문제 문자 제거
dangerous_patterns = [
(r'[\x00-\x08\x0b\x0c\x0e-\x1f]', ''), # 제어 문자
(r']*>.*?', '', re.IGNORECASE), # XSS 스크립트
(r'javascript:', '', re.IGNORECASE), # javascript 프로토콜
(r'on\w+\s*=', '', re.IGNORECASE), # 이벤트 핸들러
]
sanitized = value
for pattern, replacement, *flags in dangerous_patterns:
if flags:
sanitized = re.sub(pattern, replacement, sanitized, flags=flags[0])
else:
sanitized = re.sub(pattern, replacement, sanitized)
# 길이 제한
if len(sanitized) > max_length:
sanitized = sanitized[:max_length]
self.errors.append(
f"'{param_name}': 최대 길이 {max_length}자 초과. 잘림"
)
return sanitized.strip()
def validate_string_pattern(
self,
value: str,
pattern: str,
param_name: str
) -> bool:
"""정규식 패턴 검증"""
if not re.match(pattern, value):
self.errors.append(
f"'{param_name}': 패턴 '{pattern}'과 일치하지 않습니다. "
f"현재 값: {value[:50]}..."
)
return False
return True
def get_errors(self) -> List[str]:
"""검증 오류 목록 반환"""
return self.errors.copy()
def clear_errors(self):
"""오류 목록 초기화"""
self.errors = [] 이커머스 상품 검색 Function Calling
실제 이커머스 시나리오에서 상품 검색 AI 함수의 안전한 구현 예제입니다. HolySheep AI를 통해 여러 모델로 동일하게 테스트할 수 있습니다.
import openai
from typing import Optional, List
from enum import Enum
HolySheep AI 설정
client = openai.OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
class CategoryEnum(str, Enum):
ELECTRONICS = "electronics"
FASHION = "fashion"
HOME = "home"
FOOD = "food"
SPORTS = "sports"
BOOKS = "books"
@dataclass
class ProductSearchParams:
"""검증된 검색 파라미터"""
query: str
category: Optional[str] = None
min_price: Optional[int] = None
max_price: Optional[int] = None
page: int = 1
limit: int = 20
class ProductSearchValidator:
"""상품 검색 파라미터 검증기"""
def __init__(self):
self.validator = ParameterValidator()
def validate(self, raw_params: Dict[str, Any]) -> ProductSearchParams:
"""
AI로부터 수신한 원시 파라미터를 검증하고 정화된 객체 반환
"""
self.validator.clear_errors()
validated = {}
# 1. 검색어 검증 (필수)
query = raw_params.get('query', '')
if not query or not isinstance(query, str):
raise ValidationError('query', 'non-empty string', query)
validated['query'] = self.validator.sanitize_string(
query, 'query', max_length=200
)
if len(validated['query']) < 2:
raise ValidationError('query', '2자 이상 문자열', query)
# 2. 카테고리 검증 (선택, enum)
category = raw_params.get('category')
if category is not None:
if not self.validator.validate_enum(category, CategoryEnum, 'category'):
category = None # 잘못된 값이면 무시
else:
validated['category'] = category
# 3. 가격 범위 검증
min_price = raw_params.get('min_price')
max_price = raw_params.get('max_price')
if min_price is not None:
if not self.validator.validate_type(min_price, int, 'min_price'):
min_price = None
elif not self.validator.validate_range(min_price, 'min_price', min_val=0):
min_price = None
if max_price is not None:
if not self.validator.validate_type(max_price, int, 'max_price'):
max_price = None
elif not self.validator.validate_range(max_price, 'max_price', max_val=10000000):
max_price = None
# 가격 범위 논리 검증
if min_price is not None and max_price is not None:
if min_price > max_price:
self.validator.errors.append(
"min_price가 max_price보다 클 수 없습니다. 스왑됨"
)
min_price, max_price = max_price, min_price
validated['min_price'] = min_price
validated['max_price'] = max_price
# 4. 페이지네이션 검증
page = raw_params.get('page', 1)
if not self.validator.validate_type(page, int, 'page'):
page = 1
elif not self.validator.validate_range(page, 'page', min_val=1, max_val=100):
page = 1
validated['page'] = page
limit = raw_params.get('limit', 20)
if not self.validator.validate_type(limit, int, 'limit'):
limit = 20
elif not self.validator.validate_range(limit, 'limit', min_val=1, max_val=100):
limit = 20
validated['limit'] = limit
# 검증 오류 로깅
errors = self.validator.get_errors()
if errors:
print(f"[검증 경고] {errors}")
return ProductSearchParams(**validated)
def search_products_mock(params: ProductSearchParams) -> List[Dict]:
"""
실제 구현에서는 데이터베이스 查询
현재는 목업 데이터 반환
"""
return [
{
"id": "PROD001",
"name": f"{params.query} 관련 상품",
"price": 29900,
"category": params.category or "general"
}
]
Function Calling 정의
functions = [
{
"type": "function",
"function": {
"name": "search_products",
"description": "사용자의 검색어를 기반으로 이커머스 상품을 검색합니다",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "상품 검색어 (2자 이상 200자 이하)"
},
"category": {
"type": "string",
"enum": ["electronics", "fashion", "home", "food", "sports", "books"],
"description": "상품 카테고리 필터"
},
"min_price": {
"type": "integer",
"description": "최소 가격 (0 이상)"
},
"max_price": {
"type": "integer",
"description": "최대 가격 (1천만원 이하)"
},
"page": {
"type": "integer",
"description": "페이지 번호 (기본값: 1)",
"default": 1
},
"limit": {
"type": "integer",
"description": "페이지당 결과 수 (기본값: 20, 최대 100)",
"default": 20
}
},
"required": ["query"]
}
}
}
]
def process_user_query(user_message: str):
"""사용자 메시지 처리 및 Function Calling 실행"""
response = client.chat.completions.create(
model="gpt-4.1",
messages=[
{
"role": "system",
"content": "당신은 이커머스 상품 검색 어시스턴트입니다. "
"상품 검색 시 항상 search_products 함수를 사용하세요."
},
{"role": "user", "content": user_message}
],
tools=functions,
tool_choice="auto"
)
message = response.choices[0].message
# Function Calling이 수행된 경우
if message.tool_calls:
for tool_call in message.tool_calls:
function_name = tool_call.function.name
raw_args = json.loads(tool_call.function.arguments)
print(f"[호출 함수] {function_name}")
print(f"[원시 인자] {raw_args}")
# 파라미터 검증
validator = ProductSearchValidator()
try:
validated_params = validator.validate(raw_args)
print(f"[검증된 인자] {validated_params}")
# 함수 실행
results = search_products_mock(validated_params)
return results
except ValidationError as e:
print(f"[검증 오류] {e}")
return {"error": str(e), "results": []}
# 일반 응답
return {"response": message.content}
테스트 실행
if __name__ == "__main__":
# 정상적인 검색
result1 = process_user_query("아이폰 케이스 추천")
print(f"결과: {result1}")
# 악의적인 입력 테스트 - 파라미터 주입 시도
result2 = process_user_query(
"'; DROP TABLE products; -- 로 검색해줘"
)
print(f"결과: {result2}")
# 다국어 혼합 입력 테스트
result3 = process_user_query("北京市的电子产品")
print(f"결과: {result3}")실행 결과 및 검증 확인
# 정상 입력 시뮬레이션 출력:
[호출 함수] search_products
[원시 인자] {'query': '아이폰 케이스', 'category': 'electronics', 'limit': 10}
[검증된 인자] ProductSearchParams(query='아이폰 케이스', category='electronics', ...)
결과: [{'id': 'PROD001', 'name': '아이폰 케이스 관련 상품', ...}]
악의적 입력 시뮬레이션 출력:
[호출 함수] search_products
[원시 인자] {'query': "'; DROP TABLE products; --"}
[검증된 인자] ProductSearchParams(query="DROP TABLE products", ...) # 위험 문자 제거됨
⚠️ 제어 문자가 제거되어 의도한 SQL 주입이 실행되지 않음
다국어 입력 시뮬레이션 출력:
[호출 함수] search_products
[원시 인자] {'query': '北京市的电子产品'}
[검증된 인자] ProductSearchParams(query='北京市的电子产品', ...)
✅ 한자/한국어 혼용 입력도 안전하게 처리됨
기업 RAG 시스템에서의 Function Calling 보안
기업용 RAG(Retrieval-Augmented Generation) 시스템에서는 Function Calling이 문서 查询 및 분석 파이프라인의 핵심 역할을 합니다. 이 경우 보안 검증이 더욱 중요합니다.
import hashlib
import hmac
import time
from typing import Any, Dict, Optional
from dataclasses import dataclass
@dataclass
class FunctionCallAudit:
"""Function Calling 감사 로그"""
timestamp: float
function_name: str
params: Dict[str, Any]
user_id: str
ip_address: str
success: bool
error_message: Optional[str] = None
class SecureFunctionRegistry:
"""보안 검증이 적용된 함수 레지스트리"""
def __init__(self, secret_key: str):
self.secret_key = secret_key
self.audit_log: list[FunctionCallAudit] = []
self.rate_limits: Dict[str, list] = {} # 함수별 레이트 리밋
self.max_calls_per_minute = 60
def generate_request_id(self, user_id: str, function_name: str) -> str:
"""요청 ID 생성 - 감사 추적용"""
timestamp = str(time.time())
data = f"{user_id}:{function_name}:{timestamp}"
return hashlib.sha256(data.encode()).hexdigest()[:16]
def verify_signature(self, payload: str, signature: str) -> bool:
"""요청 무결성 검증"""
expected = hmac.new(
self.secret_key.encode(),
payload.encode(),
hashlib.sha256
).hexdigest()
return hmac.compare_digest(expected, signature)
def check_rate_limit(self, user_id: str, function_name: str) -> bool:
"""레이트 리밋 확인 - 1분당 최대 호출 횟수 제한"""
key = f"{user_id}:{function_name}"
now = time.time()
if key not in self.rate_limits:
self.rate_limits[key] = []
# 1분 이내 호출 기록 필터링
self.rate_limits[key] = [
t for t in self.rate_limits[key]
if now - t < 60
]
if len(self.rate_limits[key]) >= self.max_calls_per_minute:
return False
self.rate_limits[key].append(now)
return True
def audit(
self,
function_name: str,
params: Dict[str, Any],
user_id: str,
ip_address: str,
success: bool,
error_message: Optional[str] = None
):
"""Function Calling 감사 로그 기록"""
audit_entry = FunctionCallAudit(
timestamp=time.time(),
function_name=function_name,
params=self._sanitize_for_audit(params), # 민감 정보 마스킹
user_id=user_id,
ip_address=ip_address,
success=success,
error_message=error_message
)
self.audit_log.append(audit_entry)
# 비정상 패턴 감지
if not success:
self._detect_anomaly(user_id, function_name, error_message)
def _sanitize_for_audit(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""감사 로그용 파라미터 정화 - 민감 정보 마스킹"""
sensitive_keys = {'password', 'token', 'api_key', 'secret', 'ssn', 'card'}
sanitized = {}
for key, value in params.items():
if any(s in key.lower() for s in sensitive_keys):
sanitized[key] = "***MASKED***"
elif isinstance(value, dict):
sanitized[key] = self._sanitize_for_audit(value)
else:
sanitized[key] = value
return sanitized
def _detect_anomaly(
self,
user_id: str,
function_name: str,
error: Optional[str]
):
"""비정상 패턴 감지 - 보안 경고"""
recent_failures = sum(
1 for log in self.audit_log[-100:]
if not log.success and log.user_id == user_id
)
if recent_failures > 10:
print(
f"[🚨 보안 경고] 사용자 {user_id}가 최근 100건 중 "
f"{recent_failures}건 실패 - 자동 차단 검토 필요"
)
HolySheep AI 기반 RAG 검색 함수
def rag_search_documents(
registry: SecureFunctionRegistry,
query: str,
collection: str,
top_k: int = 5
) -> Dict[str, Any]:
"""RAG 문서 검색 - 보안 검증 적용"""
request_id = registry.generate_request_id("system", "rag_search")
# 1단계: 입력 검증
if not isinstance(query, str) or len(query) > 1000:
registry.audit(
"rag_search", {"query": query[:100]},
"system", "internal", False, "잘못된 쿼리 형식"
)
return {"error": "Invalid query format"}
# 2단계: 범위 검증
if not 1 <= top_k <= 50:
registry.audit(
"rag_search", {"top_k": top_k},
"system", "internal", False, "top_k 범위 초과"
)
top_k = min(max(top_k, 1), 50)
# 3단계: 레이트 리밋 확인
if not registry.check_rate_limit("system", "rag_search"):
return {"error