- Pydantic은 UUID4가 아닌 버전 무관 UUID 수용 - auth-server UUID4 생성 정책과 소비자 검증 정책 분리 명시 - email_integration/robeing-monitor UUID5 폴백은 장애 시만, 거부 금지 - system/external_service는 user_id 금지, actor 필드 사용 - 상위 SSOT: 0_VALUE coding-principles.md 링크 Made-with: Cursor
12 KiB
UUID 변환 시스템 아키텍처
작성일: 2025-08-21 (수정: 2026-03-27)
작성자: Claude (51123 서버 관리자)
상위 SSOT: coding-principles.md — 식별자·검증 계약 (워크스페이스 기준 상대 경로)
1. 개요
RO-BEING 시스템의 UUID 변환 체계 문서입니다. Google OAuth와 Slack 모두 UUID4를 생성하며, Slack은 slack_user_mapping 테이블을 통해 매핑합니다.
2. UUID 타입별 용도
2.1 UUID 타입 구분
| 타입 | 생성 방식 | 용도 | 특징 |
|---|---|---|---|
| UUID4 | 랜덤 생성 | Google OAuth 사용자 | auth-server에서 생성 |
| UUID | 매핑 테이블 조회 | Slack 사용자 | slack_user_mapping 테이블 사용 |
| DB 매핑 | slack_user_mapping | Slack 사용자 매핑 | 실제 사용 여부 확인 필요 |
2.2 네임스페이스
# DNS 네임스페이스 (표준 UUID)
NAMESPACE = uuid.UUID('6ba7b810-9dad-11d1-80b4-00c04fd430c8')
3. 사용자별 UUID 생성
3.1 Google OAuth 사용자 (UUID4)
flowchart LR
A[Google 로그인] --> B[이메일 확인]
B --> C{기존 사용자?}
C -->|Yes| D[기존 UUID 사용]
C -->|No| E[uuid4() 생성]
E --> F[DB 저장]
F --> G[JWT 생성]
구현 코드
import uuid
class GoogleAuthHandler:
async def create_or_get_user(self, email: str, name: str):
# 기존 사용자 확인
user = await self.db.fetchone(
"SELECT * FROM user WHERE email = $1",
email
)
if user:
return user['id']
# 신규 사용자 - UUID4 생성
new_user_id = str(uuid.uuid4())
await self.db.execute("""
INSERT INTO user (id, email, name, username, oauth_provider)
VALUES ($1, $2, $3, $4, 'google')
""", new_user_id, email, name, self.generate_username(email))
return new_user_id
3.2 Slack 사용자 (UUID 매핑)
flowchart LR
A[Slack 이벤트] --> B[Slack User ID]
B --> C[slack_user_mapping 조회]
C --> D[네임스페이스 + Slack ID]
D --> E[SHA-1 해시]
E --> F[결정적 UUID]
F --> G{DB 확인}
G -->|없음| H[신규 등록]
G -->|있음| I[기존 사용]
구현 코드
import uuid
class SlackUserHandler:
NAMESPACE = uuid.UUID('6ba7b810-9dad-11d1-80b4-00c04fd430c8')
def get_user_uuid(self, slack_user_id: str) -> str:
"""Slack User ID를 UUID로 변환 (매핑 테이블 조회)"""
# slack_user_mapping 테이블에서 조회
return get_uuid_from_mapping(slack_user_id)
async def get_or_create_user(self, slack_user_id: str, slack_user_info: dict):
# UUID 매핑 조회 (테이블에서 조회)
user_uuid = self.get_user_uuid(slack_user_id)
# 기존 사용자 확인
user = await self.db.fetchone(
"SELECT * FROM user WHERE id = $1",
user_uuid
)
if user:
return user_uuid
# 신규 사용자 등록
username = f"slack_{slack_user_id[:8]}"
await self.db.execute("""
INSERT INTO user (id, username, email, name, oauth_provider)
VALUES ($1, $2, $3, $4, 'slack')
""", user_uuid, username,
slack_user_info.get('email', ''),
slack_user_info.get('real_name', ''))
return user_uuid
4. Gateway에서의 UUID 변환
4.1 JWT Token → UUID 변환 플로우
sequenceDiagram
participant Client
participant Gateway
participant DB
participant Service
Client->>Gateway: Request + JWT (username)
Gateway->>Gateway: JWT 검증
Gateway->>Gateway: username 추출
Gateway->>DB: SELECT id FROM user WHERE username = ?
DB-->>Gateway: UUID
Gateway->>Service: Request + X-User-Id (UUID)
Service-->>Gateway: Response
Gateway-->>Client: Response
4.2 변환 구현
class GatewayUUIDConverter:
def __init__(self, db_pool):
self.db_pool = db_pool
self.cache = {} # username -> uuid 캐시
async def username_to_uuid(self, username: str) -> str:
"""Username을 UUID로 변환 (캐싱 포함)"""
# 캐시 확인
if username in self.cache:
return self.cache[username]
# DB 조회
async with self.db_pool.acquire() as conn:
row = await conn.fetchrow(
"SELECT id FROM user WHERE username = $1",
username
)
if not row:
raise UserNotFoundError(f"User {username} not found")
user_uuid = str(row['id'])
# 캐시 저장 (5분 TTL)
self.cache[username] = user_uuid
asyncio.create_task(self.clear_cache_after(username, 300))
return user_uuid
async def clear_cache_after(self, username: str, seconds: int):
"""일정 시간 후 캐시 제거"""
await asyncio.sleep(seconds)
self.cache.pop(username, None)
5. 테스트 사용자 UUID
5.1 하드코딩된 테스트 UUID
-- 테스트 사용자 (고정 UUID)
INSERT INTO user (id, username, email, name) VALUES
('aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', 'happybell80', 'goeun2dc@gmail.com', '김종태'),
('bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb', 'eagle0914', '0914eagle@gmail.com', '전희재'),
('cccccccc-cccc-cccc-cccc-cccccccccccc', 'test_user', 'test@example.com', 'Test User');
5.2 Slack 테스트 사용자 UUID 예시
# Slack User ID 예시
slack_ids = {
"U0925SXQFDK": "종태", # UUID: 매핑 테이블 조회
"U0123ABCDEF": "테스트" # UUID: 매핑 테이블 조회
}
# UUID 매핑 조회 예시
for slack_id, name in slack_ids.items():
user_uuid = get_uuid_from_mapping(slack_id) # DB 조회
print(f"{name} ({slack_id}): {user_uuid}")
6. UUID 마이그레이션
6.1 기존 VARCHAR user_id → UUID 마이그레이션
-- 1. 임시 컬럼 추가
ALTER TABLE gmail_token ADD COLUMN user_uuid UUID;
-- 2. UUID 매핑
UPDATE gmail_token gt
SET user_uuid = u.id
FROM user u
WHERE gt.user_id = u.username;
-- 3. 기존 컬럼 제거 및 이름 변경
ALTER TABLE gmail_token DROP COLUMN user_id;
ALTER TABLE gmail_token RENAME COLUMN user_uuid TO user_id;
-- 4. 외래키 제약 추가
ALTER TABLE gmail_token
ADD CONSTRAINT fk_user_id
FOREIGN KEY (user_id) REFERENCES user(id);
7. UUID 유효성 검증
7.1 검증 정책 (생성 vs 소비)
-
Pydantic·소비자 검증
API·도메인 모델에서 사용자 식별자를 받을 때는UUID4가 아니라UUID(버전 무관) 를 사용한다. Slack 매핑·기타 경로에서 올 수 있는 UUID5 등도 RFC 4122 준수 유효 식별자이므로, 소비자(rb8001 등)는 버전으로 거부하지 않는다. -
생성 정책과 검증 정책의 분리
auth-server가 신규 사용자에 대해 UUID4 를 쏘는 것은 생성(발급) 정책이다. 이는 소비자 측 검증 정책(임의 버전 UUID 허용)과 별도 축으로 둔다. 생성은 한 서비스에 국한하고, 검증은 전 구간에서 동일한 완화된 규칙을 쓴다. -
레거시 UUID5 폴백
email_integration.py,robeing-monitor등에서 쓰는 UUID5 폴백은 auth-server 장애 시에만 동작하는 탈출 해치에 해당한다. 이때 생성되는 값도 유효한 UUID 이므로, 소비자가 「UUID4만 허용」 등으로 거부하면 안 된다. -
예약어와
user_id
문자열 예약어system,external_service는user_id에 넣지 않는다. 시스템/외부 주체 구분은 별도actor(또는 동등한) 필드로 표현한다.user_id는 항상 UUID(문자열 표현 포함) 계약을 유지한다.
7.2 Pydantic 예시 (버전 무관 UUID)
from uuid import UUID
from pydantic import BaseModel, Field
class UserScopedRequest(BaseModel):
"""소비자 검증: UUID4 전용이 아닌 표준 UUID 수용."""
user_id: UUID = Field(..., description="RFC 4122 UUID, 버전 무관")
7.3 검증 함수
import re
def is_valid_uuid(uuid_string: str) -> bool:
"""UUID 형식 검증"""
uuid_pattern = re.compile(
r'^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$',
re.IGNORECASE
)
return bool(uuid_pattern.match(uuid_string))
def get_uuid_version(uuid_string: str) -> int:
"""UUID 버전 확인"""
try:
u = uuid.UUID(uuid_string)
return u.version
except ValueError:
return None
# 사용 예시
user_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
if is_valid_uuid(user_id):
version = get_uuid_version(user_id)
print(f"Valid UUID v{version}")
8. UUID 관련 API 응답
8.1 사용자 정보 API
@app.get("/api/users/me")
async def get_current_user(request: Request):
user_id = request.headers.get("X-User-Id")
username = request.headers.get("X-Username")
return {
"id": user_id, # UUID
"username": username,
"uuid_version": get_uuid_version(user_id),
"is_slack_user": get_uuid_version(user_id) == 5
}
8.2 UUID 디버깅 엔드포인트
@app.get("/api/debug/uuid/{username}")
async def debug_uuid(username: str):
"""개발 환경에서만 사용"""
user = await db.fetchone(
"SELECT id, username, oauth_provider FROM user WHERE username = $1",
username
)
if not user:
return {"error": "User not found"}
return {
"username": username,
"uuid": str(user['id']),
"uuid_version": get_uuid_version(str(user['id'])),
"oauth_provider": user['oauth_provider'],
"is_deterministic": get_uuid_version(str(user['id'])) == 5
}
9. 트러블슈팅
9.1 일반적인 문제
| 문제 | 원인 | 해결 방법 |
|---|---|---|
| UUID 불일치 | Slack ID 변경 | 매핑 테이블을 통해 일관된 UUID 관리 |
| 중복 UUID | UUID4 충돌 (매우 드물음) | 재생성 또는 UNIQUE 제약 확인 |
| 변환 실패 | username 없음 | users 테이블 username 필드 확인 |
| 캐시 불일치 | TTL 만료 전 DB 변경 | 캐시 무효화 또는 TTL 단축 |
9.2 디버깅 쿼리
-- UUID 버전별 사용자 수
SELECT
CASE
WHEN id::text LIKE '________-____-4___-____-____________' THEN 'UUID4 (Google)'
WHEN source = 'slack' THEN 'UUID (Slack 매핑)'
ELSE 'Other'
END as uuid_type,
COUNT(*) as count
FROM users
GROUP BY uuid_type;
-- Username-UUID 매핑 확인
SELECT username, id, oauth_provider
FROM user
WHERE username = 'happybell80';
-- Slack 사용자 UUID 검증
SELECT
username,
id,
id = uuid_generate_v5('6ba7b810-9dad-11d1-80b4-00c04fd430c8'::uuid,
SUBSTRING(username FROM 7)) as is_valid_uuid
FROM users
WHERE oauth_provider = 'slack';
10. 보안 고려사항
10.1 UUID 노출 최소화
class SecureUUIDHandler:
@staticmethod
def mask_uuid(uuid_string: str) -> str:
"""UUID 일부 마스킹"""
# aaaaaaaa-****-****-****-aaaaaaaaaaaa
parts = uuid_string.split('-')
return f"{parts[0]}-****-****-****-{parts[4]}"
@staticmethod
def should_expose_uuid(user_role: str) -> bool:
"""역할별 UUID 노출 여부"""
return user_role in ['admin', 'developer']
10.2 UUID 추측 방지
- UUID4: 122비트 랜덤 엔트로피로 추측 불가능
- UUID 매핑: Slack ID로 테이블 조회 필요
- 네임스페이스 비공개 유지 (소스코드에서만 관리)
문서 끝