Python SDK 예제
이 페이지에서는 Ruby Team API를 위한 완전한 Python 연동 예제를 제공합니다. HMAC-SHA256 서명 유틸리티, HTTP 클라이언트 래퍼, 그리고 Seamless Wallet 연동을 위한 콜백(Callback) 검증 기능을 포함합니다.
의존성: requests (pip install requests로 설치)
서명 유틸리티 및 HTTP 클라이언트
import hashlib
import hmac
import json
import time
from urllib.parse import urlparse
import requests
class RubyApiClient:
"""
HTTP client for the Ruby Team API with automatic HMAC-SHA256 signing.
Usage:
client = RubyApiClient(
base_url="https://api-test.ruby-gaming.com",
api_key="your_team_api_key",
api_secret="your_team_api_secret",
)
"""
def __init__(self, base_url: str, api_key: str, api_secret: str) -> None:
self.base_url = base_url.rstrip("/")
self.api_key = api_key
self.api_secret = api_secret
def _sign(self, method: str, path: str, body: str = "") -> tuple[int, str]:
"""
Compute the HMAC-SHA256 signature for a Team API request.
Signature string: {timestamp}{METHOD}{path}{raw_body}
- timestamp: Unix seconds (integer, converted to string)
- METHOD: Uppercase HTTP method (GET, POST, PUT, ...)
- path: Request path including query string if present
- raw_body: Raw request body string; empty string for no-body requests
Returns (timestamp, hex_signature).
"""
timestamp = int(time.time())
message = f"{timestamp}{method.upper()}{path}{body}"
signature = hmac.new(
self.api_secret.encode("utf-8"),
message.encode("utf-8"),
hashlib.sha256,
).hexdigest()
return timestamp, signature
def _build_headers(self, method: str, path: str, body: str = "") -> dict[str, str]:
"""Build the three required authentication headers plus Content-Type."""
timestamp, signature = self._sign(method, path, body)
return {
"X-Team-Key": self.api_key,
"X-Team-Timestamp": str(timestamp),
"X-Team-Signature": signature,
"Content-Type": "application/json",
}
def _path_from_url(self, url: str) -> str:
"""Extract path (with query string) from a full URL or a bare path."""
parsed = urlparse(url)
if parsed.netloc:
# Full URL — strip the scheme + host
path = parsed.path
if parsed.query:
path = f"{path}?{parsed.query}"
return path
# Already a bare path
return url
def get(self, path: str, params: dict | None = None) -> requests.Response:
"""Send an authenticated GET request."""
url = self.base_url + path
if params:
from urllib.parse import urlencode
url = f"{url}?{urlencode(params)}"
signed_path = self._path_from_url(url)
headers = self._build_headers("GET", signed_path, "")
return requests.get(url, headers=headers)
def post(self, path: str, body: dict) -> requests.Response:
"""Send an authenticated POST request with a JSON body."""
raw_body = json.dumps(body, separators=(",", ":"))
headers = self._build_headers("POST", path, raw_body)
return requests.post(
self.base_url + path,
headers=headers,
data=raw_body.encode("utf-8"),
)
def put(self, path: str, body: dict) -> requests.Response:
"""Send an authenticated PUT request with a JSON body."""
raw_body = json.dumps(body, separators=(",", ":"))
headers = self._build_headers("PUT", path, raw_body)
return requests.put(
self.base_url + path,
headers=headers,
data=raw_body.encode("utf-8"),
)
본문 직렬화
서명에 포함되는 본문은 HTTP 요청으로 전송되는 본문과 바이트 단위로 완전히 동일해야 합니다. 위 예제에서는 json.dumps(..., separators=(",", ":")) (: 또는 , 뒤에 공백 없음)를 사용합니다. 서명 후 본문을 다시 직렬화하거나 포맷을 변경하지 마십시오.
예제: 브랜드 생성
client = RubyApiClient(
base_url="https://api-test.ruby-gaming.com",
api_key="your_team_api_key",
api_secret="your_team_api_secret",
)
response = client.post(
"/api/brand/create",
body={
"name": "My Brand",
"code": "mybrand01",
"wallet_mode": "seamless",
"callback_url": "https://mybrand.example.com/ruby/callback",
"currency": "KRW",
},
)
data = response.json()
print(data)
# {
# "id": 42,
# "name": "My Brand",
# "code": "mybrand01",
# "api_key": "aBcDeFgH...",
# "api_secret": "xYzSeCrEt...", <-- save this, returned only once
# "wallet_mode": "seamless",
# "status": 1,
# ...
# }
brand_api_key = data["api_key"]
brand_api_secret = data["api_secret"] # Store securely — never returned again
경고
api_secret을 즉시 저장하십시오api_secret 필드는 생성 응답에서만 한 번 반환됩니다. 다음 단계로 진행하기 전에 반드시 시크릿 매니저(Secret Manager)에 저장하십시오.
예제: 베팅 목록 조회
response = client.get(
"/api/bet/list",
params={
"page": 1,
"size": 20,
"brand_id": 42,
},
)
data = response.json()
print(f"Total bets: {data['total']}")
for bet in data["items"]:
print(bet)
콜백 서명 검증
브랜드에서 wallet_mode: "seamless"를 사용하는 경우, Ruby는 잔액 조회(Balance), 차감(Debit), 적립(Credit), 롤백(Rollback) 작업을 위해 귀하의 서버를 호출합니다. 각 요청은 HMAC-SHA256 서명이 포함되어 있으며, 요청을 처리하기 전에 반드시 이를 검증해야 합니다.
다른 알고리즘
콜백 검증은 Team API와 다른 서명 알고리즘을 사용합니다. HMAC 입력값은 body_bytes + timestamp_bytes (본문이 먼저, 타임스탬프가 뒤에)입니다. 키는 team_api_secret이 아닌 brand.api_secret입니다.
import hashlib
import hmac
import time
def verify_callback_signature(
api_secret: str,
body_bytes: bytes,
timestamp: str,
signature: str,
max_age_seconds: int = 300,
) -> bool:
"""
Verify the HMAC-SHA256 signature on an incoming Ruby callback request.
Args:
api_secret: Your brand's api_secret (from the create brand response).
body_bytes: Raw HTTP request body bytes — do NOT parse before calling this.
timestamp: Value of the X-Aggregator-Timestamp header (string).
signature: Value of the X-Aggregator-Signature header (hex string).
max_age_seconds: Reject requests older than this many seconds (default 300).
Returns:
True if the signature is valid and the request is fresh, False otherwise.
"""
# Validate timestamp freshness
try:
ts = int(timestamp)
except ValueError:
return False
if abs(int(time.time()) - ts) > max_age_seconds:
return False
# Callback signing: HMAC-SHA256(key=api_secret, data=body_bytes + timestamp_bytes)
hmac_input = body_bytes + timestamp.encode("utf-8")
expected = hmac.new(
api_secret.encode("utf-8"),
hmac_input,
hashlib.sha256,
).hexdigest()
# Constant-time comparison to prevent timing attacks
return hmac.compare_digest(expected, signature)
Flask 연동 예제
from flask import Flask, abort, request
app = Flask(__name__)
BRAND_API_KEY = "key_brandabc" # From create brand response
BRAND_API_SECRET = "my_brand_secret" # From create brand response
def _verify_or_abort():
"""Verify the callback signature; abort with 401 if invalid."""
raw_body = request.get_data()
valid = verify_callback_signature(
api_secret=BRAND_API_SECRET,
body_bytes=raw_body,
timestamp=request.headers.get("X-Aggregator-Timestamp", ""),
signature=request.headers.get("X-Aggregator-Signature", ""),
)
if not valid:
abort(401)
@app.route("/callback/balance", methods=["POST"])
def handle_balance():
# Read raw bytes BEFORE any parsing
_verify_or_abort()
# Return current player balance
return {"balance": "1250.00"}
@app.route("/callback/debit", methods=["POST"])
def handle_debit():
_verify_or_abort()
payload = request.get_json()
# Deduct from player wallet (must be idempotent on transaction_id)
return {"balance": "1149.50", "balance_before": "1250.00"}
@app.route("/callback/credit", methods=["POST"])
def handle_credit():
_verify_or_abort()
payload = request.get_json()
# Add winnings to player wallet (must be idempotent on transaction_id)
return {"balance": "1350.00", "balance_before": "1250.00"}
@app.route("/callback/rollback", methods=["POST"])
def handle_rollback():
_verify_or_abort()
payload = request.get_json()
# Reverse a previous debit (must be idempotent on transaction_id)
return {"balance": "1250.00"}
자주 발생하는 실수
| 실수 | 결과 | 해결 방법 |
|---|---|---|
| 서명 시 경로에서 쿼리 문자열 누락 | 서명 불일치 (401) | GET 요청 서명 시 경로에 항상 ?key=value를 포함하십시오 |
| 서명 후 JSON 본문 재직렬화 | 서명 불일치 (401) | 전송하는 바이트와 동일한 문자열로 서명하십시오 -- 포맷을 변경하지 마십시오 |
| 만료된 타임스탬프 (300초 이상 차이) | 요청 거부 (401) | 시스템 시계를 NTP와 동기화하십시오 |
콜백 검증에 team_api_secret 사용 | 검증 실패 | 콜백은 team_api_secret이 아닌 brand.api_secret을 사용합니다 |
verify_callback_signature 호출 전에 본문 파싱 | 서명 불일치 | 먼저 원시 바이트를 읽고, 검증 후에만 JSON을 파싱하십시오 |