""" Service blockchain — interaction avec le smart contract CertiChain sur Polygon. Utilise web3.py pour ancrer, remplacer et vérifier des credentials académiques. """ import hashlib import json import logging from typing import Any from web3 import Web3 from web3.middleware import ExtraDataToPOAMiddleware from app.config import settings logger = logging.getLogger(__name__) # ─── ABI du contrat CertiChain (fonctions utilisées) ──────────────────────────── CONTRACT_ABI = json.loads(""" [ { "inputs": [ {"internalType": "bytes32", "name": "credentialHash", "type": "bytes32"}, {"internalType": "string", "name": "numeroSerie", "type": "string"}, {"internalType": "string", "name": "etablissementSlug", "type": "string"} ], "name": "anchor", "outputs": [], "stateMutability": "nonpayable", "type": "function" }, { "inputs": [ {"internalType": "bytes32", "name": "oldHash", "type": "bytes32"}, {"internalType": "bytes32", "name": "newHash", "type": "bytes32"}, {"internalType": "string", "name": "newNumeroSerie", "type": "string"}, {"internalType": "string", "name": "etablissementSlug", "type": "string"} ], "name": "replace", "outputs": [], "stateMutability": "nonpayable", "type": "function" }, { "inputs": [ {"internalType": "bytes32", "name": "credentialHash", "type": "bytes32"} ], "name": "verify", "outputs": [ {"internalType": "bool", "name": "anchored", "type": "bool"}, {"internalType": "bool", "name": "replaced", "type": "bool"}, {"internalType": "bytes32", "name": "replacedBy", "type": "bytes32"}, {"internalType": "uint256", "name": "anchoredAt", "type": "uint256"}, {"internalType": "string", "name": "numeroSerie", "type": "string"} ], "stateMutability": "view", "type": "function" } ] """) def compute_credential_hash(data: dict) -> str: """ Calcule le hash SHA-256 canonique d'un credential. Les clés sont triées pour garantir un hash déterministe. Seuls les champs qui font partie du credential sont inclus. """ canonical = { "numero_serie": data["numero_serie"], "etablissement_slug": data["etablissement_slug"], "titulaire_email": data["titulaire_email"], "type": data["type"], "intitule": data["intitule"], "mention": data.get("mention") or "", "date_emission": data["date_emission"], "annee_academique": data.get("annee_academique") or "", } payload = json.dumps(canonical, ensure_ascii=False, sort_keys=True) return hashlib.sha256(payload.encode("utf-8")).hexdigest() class BlockchainService: """Gère toutes les interactions avec le smart contract CertiChain.""" def __init__(self): self._w3: Web3 | None = None self._contract = None self._account = None self._initialized = False def _init(self) -> bool: """Initialise la connexion Web3 (lazy init).""" if self._initialized: return True if not settings.admin_private_key or not settings.contract_address: logger.warning("Blockchain non configurée (ADMIN_PRIVATE_KEY ou CONTRACT_ADDRESS manquant)") return False try: self._w3 = Web3(Web3.HTTPProvider(settings.polygon_rpc_url)) # Middleware PoA nécessaire pour Polygon self._w3.middleware_onion.inject(ExtraDataToPOAMiddleware, layer=0) if not self._w3.is_connected(): logger.error("Impossible de se connecter à Polygon : %s", settings.polygon_rpc_url) return False self._account = self._w3.eth.account.from_key(settings.admin_private_key) self._contract = self._w3.eth.contract( address=Web3.to_checksum_address(settings.contract_address), abi=CONTRACT_ABI, ) self._initialized = True logger.info("Web3 connecté — wallet CertiChain : %s", self._account.address) return True except Exception as exc: logger.exception("Erreur initialisation Web3 : %s", exc) return False # ─── Helpers ──────────────────────────────────────────────────────────────── def _hex_to_bytes32(self, hex_str: str) -> bytes: """Convertit un hash hex (64 chars) en bytes32.""" return bytes.fromhex(hex_str) def _send_transaction(self, func) -> dict[str, Any]: """Signe et envoie une transaction, attend la confirmation.""" nonce = self._w3.eth.get_transaction_count(self._account.address) gas_price = self._w3.eth.gas_price tx = func.build_transaction({ "from": self._account.address, "nonce": nonce, "gasPrice": gas_price, "chainId": settings.chain_id, }) try: tx["gas"] = int(self._w3.eth.estimate_gas(tx) * 1.2) except Exception: tx["gas"] = 300_000 signed = self._account.sign_transaction(tx) tx_hash = self._w3.eth.send_raw_transaction(signed.raw_transaction) logger.info("TX envoyée : %s", tx_hash.hex()) receipt = self._w3.eth.wait_for_transaction_receipt(tx_hash, timeout=120) logger.info("TX confirmée bloc #%s", receipt["blockNumber"]) return receipt # ─── Public API ───────────────────────────────────────────────────────────── def is_available(self) -> bool: if not self._init(): return False return self._w3.is_connected() def anchor_credential(self, credential_data: dict) -> dict: """ Calcule le hash canonique d'un credential et l'ancre sur Polygon. Retourne le hash + tx_hash + block_number. """ if not self._init(): raise RuntimeError("Service blockchain non disponible") hash_hex = compute_credential_hash(credential_data) hash_bytes = self._hex_to_bytes32(hash_hex) func = self._contract.functions.anchor( hash_bytes, credential_data["numero_serie"], credential_data["etablissement_slug"], ) receipt = self._send_transaction(func) return { "hash": hash_hex, "tx_hash": receipt["transactionHash"].hex(), "block_number": receipt["blockNumber"], "anchored": receipt["status"] == 1, } def replace_credential(self, old_hash_hex: str, new_credential_data: dict) -> dict: """ Remplace un credential ancré par un correctif. Calcule le hash du nouveau credential, ancre les deux en une TX. Retourne old_hash, new_hash, tx_hash, block_number. """ if not self._init(): raise RuntimeError("Service blockchain non disponible") new_hash_hex = compute_credential_hash(new_credential_data) old_hash_bytes = self._hex_to_bytes32(old_hash_hex) new_hash_bytes = self._hex_to_bytes32(new_hash_hex) func = self._contract.functions.replace( old_hash_bytes, new_hash_bytes, new_credential_data["numero_serie"], new_credential_data["etablissement_slug"], ) receipt = self._send_transaction(func) return { "old_hash": old_hash_hex, "new_hash": new_hash_hex, "tx_hash": receipt["transactionHash"].hex(), "block_number": receipt["blockNumber"], "replaced": receipt["status"] == 1, } def verify_credential(self, hash_hex: str) -> dict: """ Vérifie un credential sur la blockchain (lecture seule, pas de TX). """ if not self._init(): raise RuntimeError("Service blockchain non disponible") hash_bytes = self._hex_to_bytes32(hash_hex) anchored, replaced, replaced_by_bytes, anchored_at, numero_serie = \ self._contract.functions.verify(hash_bytes).call() replaced_by_hex = replaced_by_bytes.hex() if replaced and replaced_by_bytes != b'\x00' * 32 else None if anchored and not replaced: message = "Credential authentique et valide" elif anchored and replaced: message = "Credential remplacé par un correctif" else: message = "Credential non trouvé sur la blockchain" return { "hash": hash_hex, "anchored": anchored, "replaced": replaced, "replaced_by": replaced_by_hex, "anchored_at": anchored_at if anchored else None, "numero_serie": numero_serie if anchored else None, "message": message, } # Singleton blockchain_service = BlockchainService()