pip install web3
pip install ens
pip install python-dotenv

INFURA_PROJECT_ID=your_project_id_here
ETHEREUM_NETWORK=mainnet

from web3 import Web3
from ens import ENS
from dotenv import load_dotenv
import os

# 환경 변수 로드
load_dotenv()

# Infura 연결 설정
infura_url = f"https://mainnet.infura.io/v3/{os.getenv('INFURA_PROJECT_ID')}"
w3 = Web3(Web3.HTTPProvider(infura_url))

# ENS 인스턴스 생성
ns = ENS.fromWeb3(w3)

# 연결 확인
if w3.is_connected():
    print("✅ 이더리움 네트워크에 성공적으로 연결되었습니다!")
    print(f"현재 블록 번호: {w3.eth.block_number}")
else:
    print("❌ 연결 실패")

def validate_ens_name(ens_name):
    """
    ENS 이름의 유효성을 검증하는 함수
    
    Args:
        ens_name (str): 검증할 ENS 이름 (예: 'vitalik.eth')
    
    Returns:
        dict: 검증 결과와 관련 정보
    """
    result = {
        'is_valid': False,
        'ens_name': ens_name,
        'ethereum_address': None,
        'error': None
    }
    
    try:
        # ENS 이름 형식 검증
        if not ens_name.endswith('.eth'):
            result['error'] = "ENS 이름은 .eth로 끝나야 합니다"
            return result
        
        # ENS 이름 길이 검증 (최소 3자)
        name_without_suffix = ens_name.replace('.eth', '')
        if len(name_without_suffix) < 3:
            result['error'] = "ENS 이름은 최소 3자 이상이어야 합니다"
            return result
        
        # 이더리움 주소 조회
        address = ns.address(ens_name)
        
        if address:
            result['is_valid'] = True
            result['ethereum_address'] = address
            print(f"✅ {ens_name} → {address}")
        else:
            result['error'] = "등록되지 않은 ENS 이름입니다"
            print(f"❌ {ens_name}는 등록되지 않았습니다")
            
    except Exception as e:
        result['error'] = f"검증 중 오류 발생: {str(e)}"
        print(f"⚠️ 오류: {str(e)}")
    
    return result

# 사용 예시
test_names = ['vitalik.eth', 'invalid.eth', 'ab.eth', 'test123.eth']

for name in test_names:
    print(f"\n🔍 검증 중: {name}")
    result = validate_ens_name(name)
    print(f"결과: {result}")

def reverse_lookup(ethereum_address):
    """
    이더리움 주소로 ENS 이름 조회
    
    Args:
        ethereum_address (str): 이더리움 주소
    
    Returns:
        str: ENS 이름 또는 None
    """
    try:
        # 주소 형식 검증
        if not Web3.is_address(ethereum_address):
            print(f"❌ 유효하지 않은 이더리움 주소: {ethereum_address}")
            return None
        
        # 체크섬 주소로 변환
        checksum_address = Web3.to_checksum_address(ethereum_address)
        
        # ENS 이름 조회
        ens_name = ns.name(checksum_address)
        
        if ens_name:
            print(f"✅ {checksum_address} → {ens_name}")
            return ens_name
        else:
            print(f"ℹ️ 이 주소에 연결된 ENS 이름이 없습니다")
            return None
            
    except Exception as e:
        print(f"⚠️ 역방향 조회 오류: {str(e)}")
        return None

# 사용 예시
vitalik_address = "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"
result = reverse_lookup(vitalik_address)

import time
from typing import List, Dict

def batch_validate_ens(ens_names: List[str], delay: float = 0.5) -> List[Dict]:
    """
    여러 ENS 이름을 배치로 검증
    
    Args:
        ens_names (List[str]): 검증할 ENS 이름 리스트
        delay (float): 각 요청 사이의 지연 시간 (초)
    
    Returns:
        List[Dict]: 각 ENS 이름의 검증 결과 리스트
    """
    results = []
    total = len(ens_names)
    
    print(f"🔄 총 {total}개의 ENS 이름 검증 시작...\n")
    
    for idx, ens_name in enumerate(ens_names, 1):
        print(f"[{idx}/{total}] 검증 중: {ens_name}")
        
        result = validate_ens_name(ens_name)
        results.append(result)
        
        # API 요청 제한 방지를 위한 지연
        if idx < total:
            time.sleep(delay)
    
    # 통계 출력
    valid_count = sum(1 for r in results if r['is_valid'])
    invalid_count = total - valid_count
    
    print(f"\n📊 검증 완료!")
    print(f"✅ 유효: {valid_count}개")
    print(f"❌ 무효: {invalid_count}개")
    
    return results

# 사용 예시
test_batch = [
    'vitalik.eth',
    'ethereum.eth',
    'invalid123.eth',
    'test.eth',
    'opensea.eth'
]

batch_results = batch_validate_ens(test_batch)

def get_ens_metadata(ens_name):
    """
    ENS 이름의 상세 메타데이터 조회
    
    Args:
        ens_name (str): ENS 이름
    
    Returns:
        dict: 메타데이터 정보
    """
    metadata = {
        'ens_name': ens_name,
        'ethereum_address': None,
        'owner': None,
        'resolver': None,
        'ttl': None,
        'text_records': {}
    }
    
    try:
        # 기본 주소 조회
        address = ns.address(ens_name)
        if not address:
            print(f"❌ {ens_name}는 등록되지 않은 이름입니다")
            return metadata
        
        metadata['ethereum_address'] = address
        
        # 소유자 조회
        owner = ns.owner(ens_name)
        metadata['owner'] = owner
        
        # Resolver 주소 조회
        resolver_address = w3.ens.resolver(ens_name).address
        metadata['resolver'] = resolver_address
        
        # 텍스트 레코드 조회 (일반적인 필드들)
        text_fields = ['email', 'url', 'avatar', 'description', 
                      'com.twitter', 'com.github']
        
        for field in text_fields:
            try:
                value = ns.get_text(ens_name, field)
                if value:
                    metadata['text_records'][field] = value
            except:
                pass
        
        print(f"✅ {ens_name} 메타데이터 조회 완료")
        
    except Exception as e:
        print(f"⚠️ 메타데이터 조회 오류: {str(e)}")
    
    return metadata

# 사용 예시
metadata = get_ens_metadata('vitalik.eth')

print("\n📋 메타데이터:")
for key, value in metadata.items():
    if value:
        print(f"  {key}: {value}")

import redis
import json
from datetime import timedelta

# Redis 연결
redis_client = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True
)

def validate_ens_with_cache(ens_name, cache_ttl=3600):
    """
    캐싱을 적용한 ENS 검증
    
    Args:
        ens_name (str): ENS 이름
        cache_ttl (int): 캐시 유효 시간 (초)
    
    Returns:
        dict: 검증 결과
    """
    cache_key = f"ens:{ens_name}"
    
    # 캐시 확인
    cached_result = redis_client.get(cache_key)
    if cached_result:
        print(f"💾 캐시에서 조회: {ens_name}")
        return json.loads(cached_result)
    
    # 캐시 미스 - 실제 조회
    print(f"🔍 블록체인에서 조회: {ens_name}")
    result = validate_ens_name(ens_name)
    
    # 결과 캐싱 (유효한 경우만)
    if result['is_valid']:
        redis_client.setex(
            cache_key,
            cache_ttl,
            json.dumps(result)
        )
    
    return result

# 사용 예시
result1 = validate_ens_with_cache('vitalik.eth')  # 블록체인 조회
result2 = validate_ens_with_cache('vitalik.eth')  # 캐시에서 조회 (빠름!)

import asyncio
from web3 import AsyncWeb3
from web3.providers.async_rpc import AsyncHTTPProvider

async def async_validate_ens(ens_name, w3_async, ns_async):
    """비동기 ENS 검증"""
    try:
        address = await ns_async.address(ens_name)
        return {
            'ens_name': ens_name,
            'is_valid': address is not None,
            'ethereum_address': address
        }
    except Exception as e:
        return {
            'ens_name': ens_name,
            'is_valid': False,
            'error': str(e)
        }

async def batch_validate_async(ens_names):
    """여러 ENS를 비동기로 검증"""
    # 비동기 Web3 설정
    infura_url = f"https://mainnet.infura.io/v3/{os.getenv('INFURA_PROJECT_ID')}"
    w3_async = AsyncWeb3(AsyncHTTPProvider(infura_url))
    ns_async = ENS.fromWeb3(w3_async)
    
    # 모든 검증 작업을 동시에 실행
    tasks = [async_validate_ens(name, w3_async, ns_async) for name in ens_names]
    results = await asyncio.gather(*tasks)
    
    return results

# 사용 예시
async def main():
    test_names = ['vitalik.eth', 'ethereum.eth', 'test.eth']
    
    import time
    start = time.time()
    results = await batch_validate_async(test_names)
    elapsed = time.time() - start
    
    print(f"⚡ {len(test_names)}개 검증 완료: {elapsed:.2f}초")
    for result in results:
        print(result)

# 실행
asyncio.run(main())

def authenticate_with_ens(ens_name, signature):
    """
    ENS 기반 사용자 인증
    
    Args:
        ens_name (str): 사용자의 ENS 이름
        signature (str): 서명된 메시지
    
    Returns:
        dict: 인증 결과
    """
    # ENS 검증
    result = validate_ens_name(ens_name)
    
    if not result['is_valid']:
        return {
            'authenticated': False,
            'error': 'Invalid ENS name'
        }
    
    ethereum_address = result['ethereum_address']
    
    # 서명 검증 (실제로는 더 복잡한 로직 필요)
    message = f"Login to DApp with {ens_name}"
    
    try:
        # 서명 복구 및 검증
        recovered_address = w3.eth.account.recover_message(
            encode_defunct(text=message),
            signature=signature
        )
        
        if recovered_address.lower() == ethereum_address.lower():
            return {
                'authenticated': True,
                'ens_name': ens_name,
                'address': ethereum_address
            }
        else:
            return {
                'authenticated': False,
                'error': 'Signature verification failed'
            }
            
    except Exception as e:
        return {
            'authenticated': False,
            'error': str(e)
        }

def send_nft_to_ens(nft_contract, token_id, recipient_ens):
    """
    ENS 이름으로 NFT 전송
    
    Args:
        nft_contract (str): NFT 컨트랙트 주소
        token_id (int): 토큰 ID
        recipient_ens (str): 수신자 ENS 이름
    
    Returns:
        dict: 전송 결과
    """
    # ENS 검증
    result = validate_ens_name(recipient_ens)
    
    if not result['is_valid']:
        return {
            'success': False,
            'error': f'{recipient_ens}는 유효하지 않은 ENS 이름입니다'
        }
    
    recipient_address = result['ethereum_address']
    
    print(f"📦 NFT 전송 준비")
    print(f"  수신자: {recipient_ens} ({recipient_address})")
    print(f"  토큰 ID: {token_id}")
    
    # 실제 NFT 전송 로직
    # (여기서는 시뮬레이션)
    
    return {
        'success': True,
        'recipient_ens': recipient_ens,
        'recipient_address': recipient_address,
        'token_id': token_id,
        'tx_hash': '0x...'  # 실제 트랜잭션 해시
    }

# 사용 예시
result = send_nft_to_ens(
    nft_contract='0x...',
    token_id=1234,
    recipient_ens='vitalik.eth'
)
print(result)

def create_social_profile(ens_name):
    """
    ENS 메타데이터로 소셜 프로필 생성
    
    Args:
        ens_name (str): ENS 이름
    
    Returns:
        dict: 소셜 프로필 정보
    """
    metadata = get_ens_metadata(ens_name)
    
    if not metadata['ethereum_address']:
        return None
    
    profile = {
        'username': ens_name,
        'display_name': ens_name.replace('.eth', ''),
        'ethereum_address': metadata['ethereum_address'],
        'avatar': None,
        'bio': None,
        'social_links': {}
    }
    
    # 텍스트 레코드에서 정보 추출
    text_records = metadata.get('text_records', {})
    
    if 'avatar' in text_records:
        profile['avatar'] = text_records['avatar']
    
    if 'description' in text_records:
        profile['bio'] = text_records['description']
    
    if 'com.twitter' in text_records:
        profile['social_links']['twitter'] = text_records['com.twitter']
    
    if 'com.github' in text_records:
        profile['social_links']['github'] = text_records['com.github']
    
    if 'url' in text_records:
        profile['social_links']['website'] = text_records['url']
    
    return profile

# 사용 예시
profile = create_social_profile('vitalik.eth')
if profile:
    print(f"👤 프로필 생성 완료: {profile['username']}")
    print(f"   주소: {profile['ethereum_address']}")
    if profile['bio']:
        print(f"   소개: {profile['bio']}")
    if profile['social_links']:
        print(f"   소셜 링크: {profile['social_links']}")

# Dockerfile
FROM python:3.10-slim

WORKDIR /app

# 의존성 설치
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 애플리케이션 코드 복사
COPY . .

# 환경 변수 설정
ENV FLASK_APP=app.py
ENV FLASK_ENV=production

# 포트 노출
EXPOSE 5000

# 헬스체크
HEALTHCHECK --interval=30s --timeout=3s \
  CMD curl -f http://localhost:5000/health || exit 1

# 애플리케이션 실행
CMD ["gunicorn", "--bind", "0.0.0.0:5000", "--workers", "4", "app:app"]

# docker-compose.yml
version: '3.8'

services:
  ens-validator:
    build: .
    ports:
      - "5000:5000"
    environment:
      - INFURA_PROJECT_ID=${INFURA_PROJECT_ID}
      - REDIS_HOST=redis
    depends_on:
      - redis
    restart: unless-stopped
    
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis-data:/data
    restart: unless-stopped
    
  prometheus:
    image: prom/prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    restart: unless-stopped

volumes:
  redis-data:

# config.py
import os

class Config:
    """기본 설정"""
    SECRET_KEY = os.getenv('SECRET_KEY', 'dev-secret-key')
    INFURA_PROJECT_ID = os.getenv('INFURA_PROJECT_ID')
    REDIS_HOST = os.getenv('REDIS_HOST', 'localhost')
    REDIS_PORT = int(os.getenv('REDIS_PORT', 6379))
    CACHE_TTL = int(os.getenv('CACHE_TTL', 3600))

class DevelopmentConfig(Config):
    """개발 환경 설정"""
    DEBUG = True
    TESTING = False
    ETHEREUM_NETWORK = 'goerli'  # 테스트넷 사용

class ProductionConfig(Config):
    """프로덕션 환경 설정"""
    DEBUG = False
    TESTING = False
    ETHEREUM_NETWORK = 'mainnet'
    
    # 프로덕션에서는 반드시 환경 변수 필요
    if not Config.INFURA_PROJECT_ID:
        raise ValueError("INFURA_PROJECT_ID must be set in production")

class TestingConfig(Config):
    """테스트 환경 설정"""
    DEBUG = True
    TESTING = True
    ETHEREUM_NETWORK = 'goerli'

# 환경에 따라 설정 선택
config = {
    'development': DevelopmentConfig,
    'production': ProductionConfig,
    'testing': TestingConfig,
    'default': DevelopmentConfig
}

def get_config():
    env = os.getenv('FLASK_ENV', 'development')
    return config.get(env, config['default'])

# .github/workflows/deploy.yml
name: Deploy ENS Validator

on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

jobs:
  test:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.10'
    
    - name: Install dependencies
      run: |
        pip install -r requirements.txt
        pip install pytest pytest-cov
    
    - name: Run tests
      run: pytest --cov=./ --cov-report=xml
    
    - name: Upload coverage
      uses: codecov/codecov-action@v3

  deploy:
    needs: test
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    
    steps:
    - uses: actions/checkout@v3
    
    - name: Deploy to production
      env:
        DEPLOY_KEY: ${{ secrets.DEPLOY_KEY }}
      run: |
        # 실제 배포 스크립트
        echo "Deploying to production..."

class MultiChainENSValidator:
    """멀티체인 ENS 검증기"""
    
    def __init__(self):
        self.validators = {
            'ethereum': EthereumENSValidator(),
            'polygon': PolygonENSValidator(),
            'bnb': BNBENSValidator()
        }
    
    def validate(self, name, chain='ethereum'):
        """체인별 검증"""
        validator = self.validators.get(chain)
        if not validator:
            raise ValueError(f"Unsupported chain: {chain}")
        
        return validator.validate(name)
    
    def validate_all_chains(self, name):
        """모든 체인에서 검증"""
        results = {}
        for chain, validator in self.validators.items():
            try:
                results[chain] = validator.validate(name)
            except Exception as e:
                results[chain] = {'error': str(e)}
        
        return results

import requests

class ENSWebhookNotifier:
    """ENS 이벤트 웹훅 알림"""
    
    def __init__(self, webhook_url):
        self.webhook_url = webhook_url
    
    def notify_registration(self, ens_name, address):
        """새 ENS 등록 알림"""
        payload = {
            'event': 'ens_registered',
            'ens_name': ens_name,
            'ethereum_address': address,
            'timestamp': datetime.utcnow().isoformat()
        }
        
        try:
            response = requests.post(
                self.webhook_url,
                json=payload,
                timeout=5
            )
            response.raise_for_status()
            print(f"✅ 웹훅 전송 성공: {ens_name}")
        except Exception as e:
            print(f"❌ 웹훅 전송 실패: {str(e)}")
    
    def notify_transfer(self, ens_name, from_address, to_address):
        """ENS 소유권 이전 알림"""
        payload = {
            'event': 'ens_transferred',
            'ens_name': ens_name,
            'from_address': from_address,
            'to_address': to_address,
            'timestamp': datetime.utcnow().isoformat()
        }
        
        requests.post(self.webhook_url, json=payload)

from ariadne import QueryType, make_executable_schema, graphql_sync
from ariadne.constants import PLAYGROUND_HTML

type_defs = """
    type Query {
        validateENS(name: String!): ENSValidationResult!
        reverseLookup(address: String!): String
        getMetadata(name: String!): ENSMetadata
    }
    
    type ENSValidationResult {
        isValid: Boolean!
        ensName: String!
        ethereumAddress: String
        error: String
    }
    
    type ENSMetadata {
        ensName: String!
        ethereumAddress: String!
        owner: String
        textRecords: [TextRecord!]!
    }
    
    type TextRecord {
        key: String!
        value: String!
    }
"""

query = QueryType()

@query.field("validateENS")
def resolve_validate_ens(_, info, name):
    result = validate_ens_name(name)
    return {
        'isValid': result['is_valid'],
        'ensName': result['ens_name'],
        'ethereumAddress': result['ethereum_address'],
        'error': result['error']
    }

@query.field("reverseLookup")
def resolve_reverse_lookup(_, info, address):
    return reverse_lookup(address)

schema = make_executable_schema(type_defs, query)

@app.route('/graphql', methods=['GET'])
def graphql_playground():
    return PLAYGROUND_HTML, 200

@app.route('/graphql', methods=['POST'])
def graphql_server():
    data = request.get_json()
    success, result = graphql_sync(
        schema,
        data,
        context_value=request,
        debug=app.debug
    )
    return jsonify(result), 200 if success else 400

import difflib

class ENSFraudDetector:
    """ENS 사기 탐지기"""
    
    def __init__(self):
        # 유명한 ENS 이름들
        self.known_names = [
            'vitalik.eth',
            'ethereum.eth',
            'opensea.eth',
            'uniswap.eth'
        ]
    
    def check_typosquatting(self, ens_name):
        """타이포스쿼팅 검사"""
        name_without_suffix = ens_name.replace('.eth', '')
        
        for known_name in self.known_names:
            known_without_suffix = known_name.replace('.eth', '')
            
            # 유사도 계산
            similarity = difflib.SequenceMatcher(
                None,
                name_without_suffix,
                known_without_suffix
            ).ratio()
            
            # 매우 유사하지만 정확히 일치하지 않으면 의심
            if 0.8 < similarity < 1.0:
                return {
                    'is_suspicious': True,
                    'reason': 'typosquatting',
                    'similar_to': known_name,
                    'similarity': similarity
                }
        
        return {'is_suspicious': False}
    
    def check_homograph_attack(self, ens_name):
        """호모그래프 공격 검사"""
        # 혼동하기 쉬운 문자들
        confusables = {
            'o': ['0', 'ο', 'о'],  # o, 숫자 0, 그리스 오미크론, 키릴 o
            'l': ['1', 'I', 'і'],  # l, 숫자 1, 대문자 I, 키릴 i
            'a': ['а', 'α'],       # a, 키릴 a, 그리스 알파
        }
        
        for char in ens_name:
            for original, confusable_chars in confusables.items():
                if char in confusable_chars:
                    return {
                        'is_suspicious': True,
                        'reason': 'homograph_attack',
                        'suspicious_char': char
                    }
        
        return {'is_suspicious': False}

# 사용 예시
detector = ENSFraudDetector()

test_names = [
    'vitalik.eth',      # 정상
    'vital1k.eth',      # 타이포스쿼팅 의심
    'vіtalik.eth',      # 호모그래프 공격 (키릴 і)
]

for name in test_names:
    typo_result = detector.check_typosquatting(name)
    homo_result = detector.check_homograph_attack(name)
    
    print(f"\n🔍 검사: {name}")
    if typo_result['is_suspicious']:
        print(f"   ⚠️ 타이포스쿼팅 의심: {typo_result}")
    if homo_result['is_suspicious']:
        print(f"   ⚠️ 호모그래프 공격 의심: {homo_result}")