🔐 지갑 서명 검증 구현하기

Web3 인증의 핵심은 지갑 서명 검증이야. 사용자가 정말 해당 지갑의 소유자인지 확인하는 과정이지. 🔑

from eth_account.messages import encode_defunct
from web3 import Web3
import secrets
from datetime import datetime, timedelta
from jose import JWTError, jwt
from typing import Optional

# JWT 설정
SECRET_KEY = "your-secret-key-here-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60

class WalletAuth:
    def __init__(self):
        self.w3 = Web3()
        self.nonce_storage = {}  # 실제로는 Redis 사용 권장
    
    def generate_nonce(self, wallet_address: str) -> str:
        """
        로그인을 위한 랜덤 nonce 생성
        """
        nonce = secrets.token_hex(16)
        # nonce는 5분간 유효
        expiry = datetime.utcnow() + timedelta(minutes=5)
        self.nonce_storage[wallet_address.lower()] = {
            'nonce': nonce,
            'expiry': expiry
        }
        return nonce
    
    def verify_signature(self, wallet_address: str, signature: str, message: str) -> bool:
        """
        지갑 서명 검증
        """
        try:
            # 메시지를 이더리움 서명 형식으로 인코딩
            encoded_message = encode_defunct(text=message)
            
            # 서명으로부터 주소 복구
            recovered_address = self.w3.eth.account.recover_message(
                encoded_message,
                signature=signature
            )
            
            # 복구된 주소와 요청 주소 비교
            return recovered_address.lower() == wallet_address.lower()
        
        except Exception as e:
            print(f"서명 검증 실패: {e}")
            return False
    
    def create_access_token(self, wallet_address: str, nft_contract: str, token_id: int) -> str:
        """
        JWT 액세스 토큰 생성
        """
        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        to_encode = {
            "sub": wallet_address.lower(),
            "nft_contract": nft_contract.lower(),
            "token_id": token_id,
            "exp": expire
        }
        encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
        return encoded_jwt
    
    def verify_token(self, token: str) -> Optional[dict]:
        """
        JWT 토큰 검증 및 페이로드 반환
        """
        try:
            payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
            return payload
        except JWTError:
            return None

이 코드의 핵심 포인트를 설명해줄게:

⛓️ NFT 소유권 확인 구현하기

이제 가장 중요한 부분! 블록체인에서 실제로 NFT를 보유하고 있는지 확인하는 코드야. 🔍

from web3 import Web3
from typing import Optional, List
import json

# ERC-721 표준 ABI (필요한 함수만)
ERC721_ABI = json.loads('''[
    {
        "constant": true,
        "inputs": [{"name": "_owner", "type": "address"}],
        "name": "balanceOf",
        "outputs": [{"name": "balance", "type": "uint256"}],
        "type": "function"
    },
    {
        "constant": true,
        "inputs": [{"name": "_tokenId", "type": "uint256"}],
        "name": "ownerOf",
        "outputs": [{"name": "owner", "type": "address"}],
        "type": "function"
    },
    {
        "constant": true,
        "inputs": [
            {"name": "_owner", "type": "address"},
            {"name": "_index", "type": "uint256"}
        ],
        "name": "tokenOfOwnerByIndex",
        "outputs": [{"name": "tokenId", "type": "uint256"}],
        "type": "function"
    }
]''')

class NFTChecker:
    def __init__(self, provider_url: str):
        """
        provider_url: Infura, Alchemy 등의 RPC URL
        예: https://mainnet.infura.io/v3/YOUR-PROJECT-ID
        """
        self.w3 = Web3(Web3.HTTPProvider(provider_url))
        
        if not self.w3.is_connected():
            raise Exception("블록체인 노드 연결 실패!")
    
    def check_ownership(self, wallet_address: str, contract_address: str, 
                       token_id: Optional[int] = None) -> bool:
        """
        특정 NFT 소유 여부 확인
        
        Args:
            wallet_address: 확인할 지갑 주소
            contract_address: NFT 컨트랙트 주소
            token_id: 특정 토큰 ID (None이면 컬렉션 내 아무 NFT나 보유하면 OK)
        
        Returns:
            bool: 소유 여부
        """
        try:
            # 주소를 체크섬 형식으로 변환
            wallet_address = self.w3.to_checksum_address(wallet_address)
            contract_address = self.w3.to_checksum_address(contract_address)
            
            # 컨트랙트 인스턴스 생성
            contract = self.w3.eth.contract(
                address=contract_address,
                abi=ERC721_ABI
            )
            
            if token_id is not None:
                # 특정 토큰 ID의 소유자 확인
                owner = contract.functions.ownerOf(token_id).call()
                return owner.lower() == wallet_address.lower()
            else:
                # 컬렉션 내 NFT를 하나라도 보유하고 있는지 확인
                balance = contract.functions.balanceOf(wallet_address).call()
                return balance > 0
        
        except Exception as e:
            print(f"NFT 소유권 확인 실패: {e}")
            return False
    
    def get_owned_tokens(self, wallet_address: str, contract_address: str, 
                        max_tokens: int = 100) -> List[int]:
        """
        지갑이 보유한 모든 토큰 ID 조회
        """
        try:
            wallet_address = self.w3.to_checksum_address(wallet_address)
            contract_address = self.w3.to_checksum_address(contract_address)
            
            contract = self.w3.eth.contract(
                address=contract_address,
                abi=ERC721_ABI
            )
            
            balance = contract.functions.balanceOf(wallet_address).call()
            
            if balance == 0:
                return []
            
            # 보유한 토큰 ID 목록 가져오기
            token_ids = []
            for i in range(min(balance, max_tokens)):
                token_id = contract.functions.tokenOfOwnerByIndex(
                    wallet_address, i
                ).call()
                token_ids.append(token_id)
            
            return token_ids
        
        except Exception as e:
            print(f"토큰 목록 조회 실패: {e}")
            return []
    
    def check_multiple_collections(self, wallet_address: str, 
                                   contract_addresses: List[str]) -> dict:
        """
        여러 NFT 컬렉션에서 소유 여부 확인
        """
        results = {}
        for contract_address in contract_addresses:
            results[contract_address] = self.check_ownership(
                wallet_address, 
                contract_address
            )
        return results

이 코드는 ERC-721 표준을 따르는 모든 NFT에서 작동해. 주요 기능들을 살펴보자:

1. check_ownership()
가장 기본적인 함수야. 특정 지갑이 특정 NFT를 보유하고 있는지 확인해. token_id를 지정하면 정확히 그 NFT를, 지정하지 않으면 해당 컬렉션의 아무 NFT나 하나라도 있으면 true를 반환해.

2. get_owned_tokens()
지갑이 보유한 모든 토큰 ID를 가져와. 이건 "당신이 가진 NFT 목록"을 보여줄 때 유용해. 다만 가스비가 들지 않는 읽기 전용 호출이라 걱정 안 해도 돼! 💰

3. check_multiple_collections()
여러 NFT 컬렉션을 동시에 체크해. 예를 들어 "BAYC나 CryptoPunks 중 하나라도 있으면 VIP"같은 로직을 구현할 때 쓰지.

🚀 FastAPI로 API 엔드포인트 구축하기

이제 모든 조각을 합쳐서 실제 API를 만들어보자! FastAPI는 정말 직관적이고 빠르니까 금방 익숙해질 거야. ⚡

from fastapi import FastAPI, Depends, HTTPException, status, Header
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from typing import Optional
import os
from dotenv import load_dotenv

from auth_utils import WalletAuth
from nft_checker import NFTChecker

# 환경 변수 로드
load_dotenv()

app = FastAPI(
    title="NFT Gated API",
    description="NFT 보유자 전용 API 서비스",
    version="1.0.0"
)

# 인증 스키마
security = HTTPBearer()

# 유틸리티 인스턴스
wallet_auth = WalletAuth()
nft_checker = NFTChecker(
    provider_url=os.getenv("WEB3_PROVIDER_URL")
)

# NFT 컨트랙트 주소 (환경 변수로 관리)
REQUIRED_NFT_CONTRACT = os.getenv("NFT_CONTRACT_ADDRESS")

# ============= 데이터 모델 =============

class NonceRequest(BaseModel):
    wallet_address: str

class LoginRequest(BaseModel):
    wallet_address: str
    signature: str
    message: str

class TokenResponse(BaseModel):
    access_token: str
    token_type: str = "bearer"

class UserInfo(BaseModel):
    wallet_address: str
    nft_balance: int
    owned_tokens: list

# ============= 인증 엔드포인트 =============

@app.post("/auth/nonce", summary="로그인용 Nonce 생성")
async def get_nonce(request: NonceRequest):
    """
    지갑 주소로 로그인용 nonce를 요청합니다.
    클라이언트는 이 nonce를 개인키로 서명해야 합니다.
    """
    nonce = wallet_auth.generate_nonce(request.wallet_address)
    message = f"재능넷 NFT 인증\n\nNonce: {nonce}\n\n이 서명은 로그인 목적으로만 사용됩니다."
    
    return {
        "message": message,
        "nonce": nonce,
        "wallet_address": request.wallet_address
    }

@app.post("/auth/login", response_model=TokenResponse, summary="지갑 서명으로 로그인")
async def login(request: LoginRequest):
    """
    지갑 서명을 검증하고 NFT 보유 여부를 확인한 후 JWT 토큰을 발급합니다.
    """
    # 1. 서명 검증
    if not wallet_auth.verify_signature(
        request.wallet_address,
        request.signature,
        request.message
    ):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="서명 검증 실패"
        )
    
    # 2. NFT 보유 여부 확인
    has_nft = nft_checker.check_ownership(
        request.wallet_address,
        REQUIRED_NFT_CONTRACT
    )
    
    if not has_nft:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="필요한 NFT를 보유하고 있지 않습니다"
        )
    
    # 3. 보유한 토큰 ID 가져오기
    owned_tokens = nft_checker.get_owned_tokens(
        request.wallet_address,
        REQUIRED_NFT_CONTRACT
    )
    
    # 4. JWT 토큰 생성
    access_token = wallet_auth.create_access_token(
        wallet_address=request.wallet_address,
        nft_contract=REQUIRED_NFT_CONTRACT,
        token_id=owned_tokens[0] if owned_tokens else 0
    )
    
    return TokenResponse(access_token=access_token)

# ============= 보호된 엔드포인트 =============

async def verify_nft_holder(
    credentials: HTTPAuthorizationCredentials = Depends(security)
) -> dict:
    """
    JWT 토큰을 검증하고 NFT 보유 여부를 확인하는 의존성 함수
    """
    token = credentials.credentials
    
    # JWT 검증
    payload = wallet_auth.verify_token(token)
    if not payload:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="유효하지 않은 토큰"
        )
    
    wallet_address = payload.get("sub")
    nft_contract = payload.get("nft_contract")
    
    # 실시간 NFT 보유 확인 (선택사항 - 캐싱 권장)
    has_nft = nft_checker.check_ownership(wallet_address, nft_contract)
    if not has_nft:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="NFT를 더 이상 보유하고 있지 않습니다"
        )
    
    return payload

@app.get("/api/user/info", response_model=UserInfo, summary="사용자 정보 조회")
async def get_user_info(user: dict = Depends(verify_nft_holder)):
    """
    인증된 사용자의 정보를 반환합니다. (NFT 보유자 전용)
    """
    wallet_address = user["sub"]
    nft_contract = user["nft_contract"]
    
    owned_tokens = nft_checker.get_owned_tokens(wallet_address, nft_contract)
    
    return UserInfo(
        wallet_address=wallet_address,
        nft_balance=len(owned_tokens),
        owned_tokens=owned_tokens
    )

@app.get("/api/premium/content", summary="프리미엄 콘텐츠")
async def get_premium_content(user: dict = Depends(verify_nft_holder)):
    """
    NFT 보유자만 접근 가능한 프리미엄 콘텐츠
    """
    return {
        "message": "축하합니다! NFT 보유자 전용 콘텐츠입니다.",
        "data": {
            "exclusive_data": "여기에 특별한 데이터가...",
            "special_features": ["기능1", "기능2", "기능3"],
            "vip_benefits": "VIP 혜택 정보"
        },
        "wallet": user["sub"]
    }

@app.get("/api/public/info", summary="공개 정보")
async def get_public_info():
    """
    누구나 접근 가능한 공개 엔드포인트
    """
    return {
        "message": "이것은 공개 정보입니다",
        "required_nft": REQUIRED_NFT_CONTRACT,
        "description": "해당 NFT를 보유하면 프리미엄 기능에 접근할 수 있습니다"
    }

# ============= 헬스체크 =============

@app.get("/health")
async def health_check():
    """
    서버 상태 확인
    """
    return {
        "status": "healthy",
        "blockchain_connected": nft_checker.w3.is_connected()
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

와! 코드가 좀 길지? 하지만 구조는 단순해. 하나씩 뜯어보자. 🔬

⚙️ 환경 설정 및 배포 준비

코드는 완성됐어! 이제 실제로 실행하기 위한 설정을 해보자. 🔧

# Web3 Provider (Infura, Alchemy 등)
WEB3_PROVIDER_URL=https://mainnet.infura.io/v3/YOUR_PROJECT_ID

# NFT 컨트랙트 주소
NFT_CONTRACT_ADDRESS=0x1234567890123456789012345678901234567890

# JWT 시크릿 키 (반드시 변경!)
JWT_SECRET_KEY=your-super-secret-key-change-this-in-production

# Redis 설정 (선택사항)
REDIS_HOST=localhost
REDIS_PORT=6379

# 서버 설정
API_HOST=0.0.0.0
API_PORT=8000

fastapi==0.104.1
uvicorn[standard]==0.24.0
web3==6.11.3
python-jose[cryptography]==3.3.0
eth-account==0.10.0
pydantic==2.5.0
python-multipart==0.0.6
python-dotenv==1.0.0
redis==5.0.1

이제 서버를 실행해보자!

# 개발 모드 (자동 리로드)
uvicorn main:app --reload --host 0.0.0.0 --port 8000

# 프로덕션 모드
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

서버가 실행되면 http://localhost:8000/docs로 접속해봐. FastAPI가 자동으로 생성한 멋진 API 문서를 볼 수 있어! 📚

🎨 프론트엔드 연동 예제

백엔드만 있으면 뭐해? 실제로 사용자가 접속할 프론트엔드도 필요하지! 간단한 JavaScript 예제를 보여줄게. 💻

<!DOCTYPE html>
<html>
<head>
    <title>NFT Gated Access</title>
    <script src="https://cdn.ethers.io/lib/ethers-5.7.umd.min.js"></script>
</head>
<body>
    <h1>NFT 보유자 전용 서비스</h1>
    <button id="connectBtn">지갑 연결</button>
    <div id="status"></div>
    <div id="content"></div>

    <script>
        const API_URL = 'http://localhost:8000';
        let provider, signer, walletAddress;

        document.getElementById('connectBtn').addEventListener('click', connectWallet);

        async function connectWallet() {
            try {
                // MetaMask 연결
                if (typeof window.ethereum === 'undefined') {
                    alert('MetaMask를 설치해주세요!');
                    return;
                }

                provider = new ethers.providers.Web3Provider(window.ethereum);
                await provider.send("eth_requestAccounts", []);
                signer = provider.getSigner();
                walletAddress = await signer.getAddress();

                document.getElementById('status').innerHTML = 
                    `연결된 지갑: ${walletAddress}`;

                // 로그인 프로세스 시작
                await login();
            } catch (error) {
                console.error('지갑 연결 실패:', error);
                alert('지갑 연결에 실패했습니다.');
            }
        }

        async function login() {
            try {
                // 1. Nonce 요청
                const nonceResponse = await fetch(`${API_URL}/auth/nonce`, {
                    method: 'POST',
                    headers: { 'Content-Type': 'application/json' },
                    body: JSON.stringify({ wallet_address: walletAddress })
                });
                const nonceData = await nonceResponse.json();

                // 2. 메시지 서명
                const signature = await signer.signMessage(nonceData.message);

                // 3. 로그인 요청
                const loginResponse = await fetch(`${API_URL}/auth/login`, {
                    method: 'POST',
                    headers: { 'Content-Type': 'application/json' },
                    body: JSON.stringify({
                        wallet_address: walletAddress,
                        signature: signature,
                        message: nonceData.message
                    })
                });

                if (!loginResponse.ok) {
                    const error = await loginResponse.json();
                    alert(`로그인 실패: ${error.detail}`);
                    return;
                }

                const loginData = await loginResponse.json();
                const accessToken = loginData.access_token;

                // 토큰 저장
                localStorage.setItem('access_token', accessToken);

                document.getElementById('status').innerHTML += 
                    '<br>✅ 로그인 성공!';

                // 프리미엄 콘텐츠 로드
                await loadPremiumContent(accessToken);

            } catch (error) {
                console.error('로그인 실패:', error);
                alert('로그인에 실패했습니다.');
            }
        }

        async function loadPremiumContent(token) {
            try {
                const response = await fetch(`${API_URL}/api/premium/content`, {
                    headers: {
                        'Authorization': `Bearer ${token}`
                    }
                });

                if (!response.ok) {
                    throw new Error('콘텐츠 로드 실패');
                }

                const data = await response.json();
                
                document.getElementById('content').innerHTML = `
                    <h2>🎉 프리미엄 콘텐츠</h2>
                    <p>${data.message}</p>
                    <pre>${JSON.stringify(data.data, null, 2)}</pre>
                `;

            } catch (error) {
                console.error('콘텐츠 로드 실패:', error);
            }
        }

        // 페이지 로드 시 저장된 토큰 확인
        window.addEventListener('load', async () => {
            const token = localStorage.getItem('access_token');
            if (token) {
                // 토큰이 있으면 자동으로 콘텐츠 로드
                await loadPremiumContent(token);
            }
        });
    </script>
</body>
</html>

이 코드는 ethers.js를 사용해서 MetaMask와 연동하고, 우리가 만든 API와 통신해. 실제로 동작하는 모습을 보면 정말 신기할 거야! ✨

🔒 보안 고려사항

Web3 개발에서 보안은 정말정말 중요해. 한 번 실수하면 돈이 날아갈 수도 있거든! ⚠️

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from fastapi import Request

limiter = Limiter(key_func=get_remote_address)

# main.py에 추가
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# 엔드포인트에 적용
@app.post("/auth/login")
@limiter.limit("5/minute")  # 분당 5회 제한
async def login(request: Request, login_data: LoginRequest):
    # ... 기존 코드

⚡ 성능 최적화 전략

블록체인 조회는 느릴 수 있어. 사용자 경험을 개선하기 위한 최적화 방법들을 알아보자! 🚀

import redis
import json
from typing import Optional
from datetime import timedelta

class CacheManager:
    def __init__(self, redis_host='localhost', redis_port=6379):
        self.redis_client = redis.Redis(
            host=redis_host,
            port=redis_port,
            decode_responses=True
        )
    
    def cache_nft_ownership(self, wallet_address: str, contract_address: str, 
                           has_nft: bool, ttl_seconds: int = 300):
        """
        NFT 소유권 정보를 캐싱 (기본 5분)
        """
        key = f"nft:{contract_address}:{wallet_address}"
        self.redis_client.setex(
            key,
            timedelta(seconds=ttl_seconds),
            json.dumps({"has_nft": has_nft})
        )
    
    def get_cached_ownership(self, wallet_address: str, 
                            contract_address: str) -> Optional[bool]:
        """
        캐시된 소유권 정보 조회
        """
        key = f"nft:{contract_address}:{wallet_address}"
        cached = self.redis_client.get(key)
        
        if cached:
            data = json.loads(cached)
            return data.get("has_nft")
        return None
    
    def invalidate_cache(self, wallet_address: str, contract_address: str):
        """
        캐시 무효화 (NFT 전송 시 호출)
        """
        key = f"nft:{contract_address}:{wallet_address}"
        self.redis_client.delete(key)

# 사용 예제
cache = CacheManager()

def check_ownership_with_cache(wallet_address: str, contract_address: str) -> bool:
    # 1. 캐시 확인
    cached_result = cache.get_cached_ownership(wallet_address, contract_address)
    if cached_result is not None:
        return cached_result
    
    # 2. 캐시 미스 - 블록체인 조회
    has_nft = nft_checker.check_ownership(wallet_address, contract_address)
    
    # 3. 결과 캐싱
    cache.cache_nft_ownership(wallet_address, contract_address, has_nft)
    
    return has_nft

여러 사용자의 NFT를 한 번에 확인해야 할 때는 배치 처리가 효율적이야.

from typing import List, Dict
import asyncio
from concurrent.futures import ThreadPoolExecutor

class BatchNFTChecker:
    def __init__(self, nft_checker: NFTChecker, max_workers: int = 10):
        self.nft_checker = nft_checker
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
    
    async def check_multiple_owners(self, wallet_addresses: List[str], 
                                   contract_address: str) -> Dict[str, bool]:
        """
        여러 지갑의 NFT 소유 여부를 병렬로 확인
        """
        loop = asyncio.get_event_loop()
        
        tasks = [
            loop.run_in_executor(
                self.executor,
                self.nft_checker.check_ownership,
                wallet,
                contract_address
            )
            for wallet in wallet_addresses
        ]
        
        results = await asyncio.gather(*tasks)
        
        return {
            wallet: result 
            for wallet, result in zip(wallet_addresses, results)
        }

# 사용 예제
batch_checker = BatchNFTChecker(nft_checker)

@app.post("/api/admin/check-batch")
async def check_batch_ownership(wallet_addresses: List[str]):
    """
    관리자용: 여러 지갑의 NFT 보유 여부를 한 번에 확인
    """
    results = await batch_checker.check_multiple_owners(
        wallet_addresses,
        REQUIRED_NFT_CONTRACT
    )
    return results

📊 모니터링과 로깅

프로덕션 환경에서는 무슨 일이 일어나는지 추적하는 게 중요해. 제대로 된 로깅 시스템을 구축해보자! 🔍

import logging
from logging.handlers import RotatingFileHandler
import json
from datetime import datetime
from typing import Any, Dict

class JSONFormatter(logging.Formatter):
    """
    JSON 형식으로 로그를 출력하는 포매터
    """
    def format(self, record: logging.LogRecord) -> str:
        log_data = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno
        }
        
        # 추가 필드가 있으면 포함
        if hasattr(record, 'wallet_address'):
            log_data['wallet_address'] = record.wallet_address
        if hasattr(record, 'nft_contract'):
            log_data['nft_contract'] = record.nft_contract
        if hasattr(record, 'action'):
            log_data['action'] = record.action
        
        return json.dumps(log_data)

def setup_logger(name: str = "nft_api") -> logging.Logger:
    """
    로거 설정
    """
    logger = logging.getLogger(name)
    logger.setLevel(logging.INFO)
    
    # 콘솔 핸들러
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(JSONFormatter())
    logger.addHandler(console_handler)
    
    # 파일 핸들러 (10MB마다 로테이션, 최대 5개 파일)
    file_handler = RotatingFileHandler(
        'nft_api.log',
        maxBytes=10*1024*1024,
        backupCount=5
    )
    file_handler.setFormatter(JSONFormatter())
    logger.addHandler(file_handler)
    
    return logger

# 로거 인스턴스
logger = setup_logger()

# 사용 예제
def log_authentication_attempt(wallet_address: str, success: bool, 
                               nft_contract: str = None):
    """
    인증 시도 로깅
    """
    extra = {
        'wallet_address': wallet_address,
        'action': 'authentication',
        'success': success
    }
    
    if nft_contract:
        extra['nft_contract'] = nft_contract
    
    if success:
        logger.info("Authentication successful", extra=extra)
    else:
        logger.warning("Authentication failed", extra=extra)

# API에 적용
@app.post("/auth/login")
async def login(request: LoginRequest):
    try:
        # ... 기존 로직
        
        # 성공 로깅
        log_authentication_attempt(
            wallet_address=request.wallet_address,
            success=True,
            nft_contract=REQUIRED_NFT_CONTRACT
        )
        
        return {"access_token": token}
    
    except Exception as e:
        # 실패 로깅
        log_authentication_attempt(
            wallet_address=request.wallet_address,
            success=False
        )
        logger.error(f"Login error: {str(e)}", exc_info=True)
        raise

더 나아가서 Prometheus로 메트릭을 수집하면 시스템 상태를 실시간으로 모니터링할 수 있어!

from prometheus_client import Counter, Histogram, Gauge
from prometheus_fastapi_instrumentator import Instrumentator

# 메트릭 정의
authentication_attempts = Counter(
    'nft_auth_attempts_total',
    'Total number of authentication attempts',
    ['status', 'chain']
)

nft_check_duration = Histogram(
    'nft_check_duration_seconds',
    'Time spent checking NFT ownership',
    ['chain']
)

active_users = Gauge(
    'nft_active_users',
    'Number of currently authenticated users'
)

# FastAPI에 Prometheus 추가
instrumentator = Instrumentator()

@app.on_event("startup")
async def startup():
    instrumentator.instrument(app).expose(app)

# 사용 예제
@app.post("/auth/login")
async def login(request: LoginRequest):
    with nft_check_duration.labels(chain='ethereum').time():
        has_nft = nft_checker.check_ownership(
            request.wallet_address,
            REQUIRED_NFT_CONTRACT
        )
    
    if has_nft:
        authentication_attempts.labels(status='success', chain='ethereum').inc()
        active_users.inc()
    else:
        authentication_attempts.labels(status='failed', chain='ethereum').inc()
    
    # ... 나머지 로직

이제 http://localhost:8000/metrics에서 메트릭을 확인할 수 있어! Grafana와 연동하면 멋진 대시보드도 만들 수 있지. 📈

💡 실전 활용 사례

이론은 충분히 배웠으니, 실제로 어떻게 활용할 수 있는지 구체적인 예시를 보여줄게! 🌟

🔮 고급 기능 추가하기

기본 기능은 다 구현했으니, 이제 좀 더 고급 기능들을 추가해보자! ⚡

NFT의 속성(traits)에 따라 다른 권한을 부여할 수 있어. 예를 들어 "레어" 속성을 가진 NFT는 더 많은 기능에 접근할 수 있게 하는 거지.

from typing import Dict, List
import requests

class NFTMetadataChecker:
    def __init__(self, nft_checker: NFTChecker):
        self.nft_checker = nft_checker
    
    def get_token_metadata(self, contract_address: str, token_id: int) -> Dict:
        """
        NFT 메타데이터 조회
        """
        # ERC-721 tokenURI 함수 호출
        contract = self.nft_checker.w3.eth.contract(
            address=contract_address,
            abi=[{
                "constant": true,
                "inputs": [{"name": "_tokenId", "type": "uint256"}],
                "name": "tokenURI",
                "outputs": [{"name": "", "type": "string"}],
                "type": "function"
            }]
        )
        
        token_uri = contract.functions.tokenURI(token_id).call()
        
        # IPFS URI를 HTTP로 변환
        if token_uri.startswith('ipfs://'):
            token_uri = token_uri.replace('ipfs://', 'https://ipfs.io/ipfs/')
        
        # 메타데이터 다운로드
        response = requests.get(token_uri)
        return response.json()
    
    def check_trait(self, metadata: Dict, trait_type: str, 
                   trait_value: str) -> bool:
        """
        특정 속성 보유 여부 확인
        """
        attributes = metadata.get('attributes', [])
        for attr in attributes:
            if attr.get('trait_type') == trait_type:
                return attr.get('value') == trait_value
        return False
    
    def calculate_rarity_score(self, metadata: Dict) -> float:
        """
        희귀도 점수 계산 (간단한 예시)
        """
        attributes = metadata.get('attributes', [])
        
        # 속성 개수가 적을수록 희귀
        base_score = 100 - len(attributes) * 5
        
        # 특정 희귀 속성 보너스
        rare_traits = ['Legendary', 'Epic', 'Mythic']
        for attr in attributes:
            if attr.get('value') in rare_traits:
                base_score += 50
        
        return max(0, base_score)

class PermissionManager:
    def __init__(self, metadata_checker: NFTMetadataChecker):
        self.metadata_checker = metadata_checker
    
    def get_user_permissions(self, wallet_address: str, 
                            contract_address: str,
                            token_id: int) -> List[str]:
        """
        사용자의 권한 목록 반환
        """
        permissions = ['basic_access']  # 기본 권한
        
        # 메타데이터 조회
        metadata = self.metadata_checker.get_token_metadata(
            contract_address,
            token_id
        )
        
        # 희귀도 기반 권한
        rarity_score = self.metadata_checker.calculate_rarity_score(metadata)
        
        if rarity_score >= 80:
            permissions.append('premium_access')
        if rarity_score >= 90:
            permissions.append('vip_access')
        
        # 특정 속성 기반 권한
        if self.metadata_checker.check_trait(metadata, 'Background', 'Gold'):
            permissions.append('golden_lounge')
        
        if self.metadata_checker.check_trait(metadata, 'Type', 'Founder'):
            permissions.append('founder_benefits')
        
        return permissions

# API에 적용
permission_manager = PermissionManager(
    NFTMetadataChecker(nft_checker)
)

@app.get("/api/user/permissions")
async def get_permissions(user: dict = Depends(verify_nft_holder)):
    """
    사용자의 권한 목록 조회
    """
    wallet_address = user["sub"]
    nft_contract = user["nft_contract"]
    token_id = user["token_id"]
    
    permissions = permission_manager.get_user_permissions(
        wallet_address,
        nft_contract,
        token_id
    )
    
    return {
        "wallet_address": wallet_address,
        "token_id": token_id,
        "permissions": permissions
    }

# 권한 체크 데코레이터
def require_permission(required_permission: str):
    async def permission_checker(user: dict = Depends(verify_nft_holder)):
        permissions = permission_manager.get_user_permissions(
            user["sub"],
            user["nft_contract"],
            user["token_id"]
        )
        
        if required_permission not in permissions:
            raise HTTPException(
                status_code=403,
                detail=f"'{required_permission}' 권한이 필요합니다"
            )
        
        return user
    
    return permission_checker

# 사용 예제
@app.get("/api/vip/exclusive")
async def vip_content(user: dict = Depends(require_permission('vip_access'))):
    """
    VIP 전용 콘텐츠 (희귀도 90+ NFT 보유자만)
    """
    return {
        "message": "VIP 전용 콘텐츠입니다!",
        "data": "초특급 비밀 정보..."
    }

NFT를 일정 기간 동안만 보유해야 접근할 수 있게 하거나, 특정 시간대에만 접근을 허용할 수도 있어.

from datetime import datetime, timedelta
from typing import Optional

class TimeBasedAccessControl:
    def __init__(self, nft_checker: NFTChecker):
        self.nft_checker = nft_checker
        self.holding_records = {}  # 실제로는 DB 사용
    
    def record_first_holding(self, wallet_address: str, 
                            contract_address: str,
                            token_id: int):
        """
        NFT를 처음 보유한 시점 기록
        """
        key = f"{wallet_address}:{contract_address}:{token_id}"
        if key not in self.holding_records:
            self.holding_records[key] = datetime.utcnow()
    
    def get_holding_duration(self, wallet_address: str,
                            contract_address: str,
                            token_id: int) -> Optional[timedelta]:
        """
        NFT 보유 기간 조회
        """
        key = f"{wallet_address}:{contract_address}:{token_id}"
        first_holding = self.holding_records.get(key)
        
        if not first_holding:
            return None
        
        return datetime.utcnow() - first_holding
    
    def check_minimum_holding_period(self, wallet_address: str,
                                    contract_address: str,
                                    token_id: int,
                                    min_days: int) -> bool:
        """
        최소 보유 기간 확인
        """
        duration = self.get_holding_duration(
            wallet_address,
            contract_address,
            token_id
        )
        
        if not duration:
            return False
        
        return duration.days >= min_days
    
    def check_time_window(self, start_hour: int, end_hour: int) -> bool:
        """
        특정 시간대 확인 (UTC 기준)
        """
        current_hour = datetime.utcnow().hour
        return start_hour <= current_hour < end_hour

# API에 적용
time_access = TimeBasedAccessControl(nft_checker)

@app.get("/api/veteran/content")
async def veteran_content(user: dict = Depends(verify_nft_holder)):
    """
    장기 보유자 전용 콘텐츠 (30일 이상 보유)
    """
    is_veteran = time_access.check_minimum_holding_period(
        user["sub"],
        user["nft_contract"],
        user["token_id"],
        min_days=30
    )
    
    if not is_veteran:
        raise HTTPException(
            status_code=403,
            detail="30일 이상 NFT를 보유해야 접근할 수 있습니다"
        )
    
    return {
        "message": "장기 보유자 감사 콘텐츠",
        "bonus": "특별 보상"
    }

@app.get("/api/happy-hour/deals")
async def happy_hour_deals(user: dict = Depends(verify_nft_holder)):
    """
    해피아워 특가 (UTC 14:00-16:00만 접근 가능)
    """
    if not time_access.check_time_window(14, 16):
        raise HTTPException(
            status_code=403,
            detail="해피아워 시간(14:00-16:00 UTC)에만 접근 가능합니다"
        )
    
    return {
        "message": "해피아워 특가!",
        "deals": ["50% 할인", "2+1 이벤트"]
    }