""" Módulo de integración con Polygon para publicación de digest de cumplimiento Este módulo publica hashes mensuales de autorizaciones en Polygon blockchain para garantizar trazabilidad y cumplimiento normativo público (AI Act, GDPR). """ import os import json import hashlib from typing import Dict, Any, List, Optional from datetime import datetime, timezone from dataclasses import dataclass import logging # Imports comentados hasta configuración # from web3 import Web3 # Configuración logger = logging.getLogger(__name__) @dataclass class DigestRecord: """Registro de digest para publicación en Polygon""" period: str # "2025-11" formato YYYY-MM root_hash: str # Hash SHA-256 de todas las autorizaciones del período authorization_count: int timestamp: str publisher_address: str transaction_hash: Optional[str] = None block_number: Optional[int] = None gas_used: Optional[int] = None class PolygonDigestPublisher: """Publicador de digest en Polygon blockchain""" def __init__(self): # Configuración Web3 (comentada hasta activación) """ self.w3 = Web3(Web3.HTTPProvider(os.getenv("POLYGON_RPC_URL"))) self.private_key = os.getenv("POLYGON_WALLET_PRIVATE_KEY") self.account = self.w3.eth.account.from_key(self.private_key) self.chain_id = int(os.getenv("POLYGON_CHAIN_ID", "137")) # 137 mainnet, 80002 Amoy testnet self.contract_addr = os.getenv("DIGEST_CONTRACT_ADDR") self.contract_abi = json.loads(os.getenv("DIGEST_CONTRACT_ABI", "[]")) """ # Temporal: configuración simulada self.w3 = None self.account = None self.contract_addr = "0x0000000000000000000000000000000000000000" # Placeholder self.chain_id = 137 logger.info("PolygonDigestPublisher inicializado (modo simulado)") def _compute_monthly_digest(self, authorizations: List[Dict[str, Any]]) -> str: """ Calcula hash SHA-256 de todas las autorizaciones del mes Args: authorizations: Lista de registros de autorización del período Returns: Hash hexadecimal de 64 caracteres """ # Ordenar autorizaciones por timestamp para consistencia sorted_auths = sorted(authorizations, key=lambda x: x.get('timestamp', '')) # Crear string concatenado con todos los datos relevantes digest_data = "" for auth in sorted_auths: # Campos relevantes para el digest relevant_data = { 'user_email': auth.get('user_email', ''), 'video_hash': auth.get('video_hash', ''), 'timestamp': auth.get('timestamp', ''), 'consent_accepted': auth.get('consent_accepted', False), 'validation_status': auth.get('validation_status', ''), 'document_id': auth.get('document_id', '') } # Convertir a JSON ordenado y añadir al digest auth_json = json.dumps(relevant_data, sort_keys=True, separators=(',', ':')) digest_data += auth_json + "|" # Calcular hash SHA-256 digest_hash = hashlib.sha256(digest_data.encode('utf-8')).hexdigest() logger.info(f"Digest calculado: {len(authorizations)} autorizaciones → {digest_hash[:16]}...") return digest_hash def _get_period_from_timestamp(self, timestamp: str) -> str: """ Extrae período YYYY-MM del timestamp ISO Args: timestamp: Timestamp en formato ISO 8601 Returns: Período en formato YYYY-MM """ try: dt = datetime.fromisoformat(timestamp.replace('Z', '+00:00')) return dt.strftime("%Y-%m") except Exception as e: logger.error(f"Error parseando timestamp {timestamp}: {e}") return datetime.now(timezone.utc).strftime("%Y-%m") def publish_monthly_digest(self, authorizations: List[Dict[str, Any]]) -> Optional[DigestRecord]: """ Publica digest mensual en Polygon blockchain Args: authorizations: Lista de autorizaciones del período a publicar Returns: DigestRecord con resultado de la publicación """ if not authorizations: logger.warning("No hay autorizaciones para publicar") return None try: # Calcular período y hash first_auth = authorizations[0] period = self._get_period_from_timestamp(first_auth.get('timestamp', '')) root_hash = self._compute_monthly_digest(authorizations) # Crear registro digest_record = DigestRecord( period=period, root_hash=root_hash, authorization_count=len(authorizations), timestamp=datetime.now(timezone.utc).isoformat(), publisher_address=self.account.address if self.account else "0x0000000000000000000000000000000000000000" ) # Publicar en blockchain (comentado hasta activación) """ contract = self.w3.eth.contract( address=Web3.to_checksum_address(self.contract_addr), abi=self.contract_abi ) nonce = self.w3.eth.get_transaction_count(self.account.address) tx = contract.functions.publish( Web3.to_bytes(hexstr=root_hash), period ).build_transaction({ "from": self.account.address, "nonce": nonce, "gas": 120000, "maxFeePerGas": self.w3.to_wei('60', 'gwei'), "maxPriorityFeePerGas": self.w3.to_wei('2', 'gwei'), "chainId": self.chain_id }) signed = self.w3.eth.account.sign_transaction(tx, self.private_key) tx_hash = self.w3.eth.send_raw_transaction(signed.rawTransaction) receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash) # Actualizar registro con datos de la transacción digest_record.transaction_hash = receipt.transactionHash.hex() digest_record.block_number = receipt.blockNumber digest_record.gas_used = receipt.gasUsed """ # Temporal: simulación de publicación simulated_tx_hash = f"0x{'0123456789abcdef' * 4}" # 64 chars hex digest_record.transaction_hash = simulated_tx_hash digest_record.block_number = 12345678 digest_record.gas_used = 87654 logger.info(f"Digest publicado simulado: {period} → {simulated_tx_hash}") return digest_record except Exception as e: logger.error(f"Error publicando digest: {e}") return None def verify_digest_on_chain(self, period: str, expected_hash: str) -> bool: """ Verifica que el digest publicado en blockchain coincide con el hash esperado Args: period: Período YYYY-MM a verificar expected_hash: Hash esperado del digest Returns: True si coincide, False si no """ try: # Consultar blockchain (comentado hasta activación) """ contract = self.w3.eth.contract( address=Web3.to_checksum_address(self.contract_addr), abi=self.contract_abi ) on_chain_hash = contract.functions.digests(period).call() on_chain_hex = self.w3.to_hex(on_chain_hash) return on_chain_hex == expected_hash """ # Temporal: simulación de verificación logger.info(f"Verificación simulada: {period} → {expected_hash[:16]}...") return True except Exception as e: logger.error(f"Error verificando digest: {e}") return False def get_published_digests(self) -> List[Dict[str, Any]]: """ Obtiene lista de todos los digest publicados (simulado) Returns: Lista de digest publicados con metadata """ # Temporal: retorno simulado return [ { "period": "2025-11", "transaction_hash": "0x0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", "block_number": 12345678, "timestamp": "2025-11-03T14:30:00Z", "authorization_count": 42 } ] # Instancia global digest_publisher = PolygonDigestPublisher()