1419 lines
53 KiB
Python
1419 lines
53 KiB
Python
"""
|
|
Darkroom Token Deployer — Multi-Chain Token Factory
|
|
====================================================
|
|
Deploy, mint, and manage tokens across 5+ chains from the admin backend.
|
|
|
|
Supported chains:
|
|
• Ethereum (ERC-20)
|
|
• Base (ERC-20)
|
|
• BSC / BNB Chain (ERC-20)
|
|
• TRON (TRC-20)
|
|
• Solana (SPL Token)
|
|
|
|
Each chain has a dedicated deployer class implementing the same interface:
|
|
- deploy_token(params) -> deployment_record
|
|
- mint_tokens(contract, to, amount) -> tx_hash
|
|
- burn_tokens(contract, amount) -> tx_hash
|
|
- transfer_ownership(contract, new_owner) -> tx_hash
|
|
- renounce_ownership(contract) -> tx_hash
|
|
- get_token_info(contract) -> metadata
|
|
|
|
Architecture:
|
|
TokenDeployerFactory -> picks chain deployer
|
|
ChainDeployer (base) -> abstract interface
|
|
EVMDeployer -> ETH, Base, BSC (shared Web3 logic)
|
|
SolanaDeployer -> SPL tokens via solana-py
|
|
TronDeployer -> TRC-20 via tronpy
|
|
|
|
Security:
|
|
- Admin-only API endpoints (X-Admin-Key required)
|
|
- Private keys stored in env vars, never logged
|
|
- Deployment records stored in Supabase + Redis
|
|
- Ownership can be renounced or transferred
|
|
|
|
Usage (admin API):
|
|
POST /api/v1/admin/tokens/deploy
|
|
POST /api/v1/admin/tokens/mint
|
|
POST /api/v1/admin/tokens/burn
|
|
POST /api/v1/admin/tokens/transfer-ownership
|
|
GET /api/v1/admin/tokens/list
|
|
GET /api/v1/admin/tokens/{deployment_id}
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import asdict, dataclass, field
|
|
from datetime import datetime
|
|
from typing import ClassVar, Any
|
|
|
|
logger = logging.getLogger("darkroom_token_deployer")
|
|
|
|
# ── Deployment Record ─────────────────────────────────────────
|
|
|
|
|
|
@dataclass
|
|
class TokenDeployment:
|
|
"""Record of a deployed token."""
|
|
|
|
deployment_id: str
|
|
chain: str
|
|
name: str
|
|
symbol: str
|
|
decimals: int
|
|
total_supply: str
|
|
contract_address: str
|
|
deployer_address: str
|
|
tx_hash: str
|
|
block_number: int | None = None
|
|
status: str = "deployed" # deployed, minting, burned, ownership_transferred
|
|
owner_address: str = ""
|
|
mintable: bool = True
|
|
burnable: bool = True
|
|
pausable: bool = False
|
|
blacklist_enabled: bool = False
|
|
max_wallet_limit: str | None = None
|
|
max_tx_limit: str | None = None
|
|
trading_enabled: bool = True
|
|
anti_bot_delay: int = 0
|
|
max_supply: str | None = None
|
|
metadata_uri: str | None = None
|
|
created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())
|
|
updated_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())
|
|
extra: dict[str, Any] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict:
|
|
d = asdict(self)
|
|
d["created_at"] = self.created_at
|
|
d["updated_at"] = self.updated_at
|
|
return d
|
|
|
|
|
|
@dataclass
|
|
class DeployParams:
|
|
"""Parameters for token deployment."""
|
|
|
|
chain: str
|
|
name: str
|
|
symbol: str
|
|
decimals: int = 18
|
|
initial_supply: str = "1000000"
|
|
max_supply: str | None = None
|
|
mintable: bool = True
|
|
burnable: bool = True
|
|
pausable: bool = False
|
|
owner_address: str | None = None
|
|
metadata_uri: str | None = None
|
|
# EVM-specific
|
|
evm_version: str = "paris"
|
|
# Solana-specific
|
|
freeze_authority: str | None = None
|
|
# TRON-specific
|
|
token_type: str = "trc20"
|
|
# Advanced
|
|
tax_rate: float | None = None
|
|
tax_recipient: str | None = None
|
|
deflationary: bool = False
|
|
# Blacklist / anti-bot
|
|
blacklist_enabled: bool = True
|
|
blacklist_addresses: list[str] = field(default_factory=list)
|
|
max_wallet_limit: str | None = None
|
|
max_tx_limit: str | None = None
|
|
trading_enabled: bool = True
|
|
anti_bot_delay: int = 0
|
|
|
|
|
|
# ── Base Deployer Interface ───────────────────────────────────
|
|
|
|
|
|
class ChainDeployer(ABC):
|
|
"""Abstract base for all chain deployers."""
|
|
|
|
def __init__(self, chain: str, private_key: str, rpc_url: str):
|
|
self.chain = chain
|
|
self.private_key = private_key
|
|
self.rpc_url = rpc_url
|
|
self.deployer_address = self._derive_address()
|
|
|
|
@abstractmethod
|
|
def _derive_address(self) -> str:
|
|
"""Derive deployer wallet address from private key."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def deploy_token(self, params: DeployParams) -> TokenDeployment:
|
|
"""Deploy a new token contract."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def mint_tokens(self, contract_address: str, to_address: str, amount: str) -> str:
|
|
"""Mint additional tokens. Returns tx hash."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def burn_tokens(self, contract_address: str, amount: str) -> str:
|
|
"""Burn tokens from deployer balance. Returns tx hash."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def transfer_ownership(self, contract_address: str, new_owner: str) -> str:
|
|
"""Transfer contract ownership. Returns tx hash."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def renounce_ownership(self, contract_address: str) -> str:
|
|
"""Renounce contract ownership (immutable). Returns tx hash."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_token_info(self, contract_address: str) -> dict[str, Any]:
|
|
"""Get token metadata from contract."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_balance(self, contract_address: str, wallet_address: str) -> str:
|
|
"""Get token balance for a wallet."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def blacklist_add(self, contract_address: str, address: str) -> str:
|
|
"""Add an address to the token blacklist. Returns tx hash."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def blacklist_remove(self, contract_address: str, address: str) -> str:
|
|
"""Remove an address from the token blacklist. Returns tx hash."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def is_blacklisted(self, contract_address: str, address: str) -> bool:
|
|
"""Check if an address is blacklisted."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def set_trading_enabled(self, contract_address: str, enabled: bool) -> str:
|
|
"""Enable or disable trading. Returns tx hash."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def set_max_wallet(self, contract_address: str, max_amount: str) -> str:
|
|
"""Set max wallet holding limit. Returns tx hash."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def set_max_tx(self, contract_address: str, max_amount: str) -> str:
|
|
"""Set max transaction limit. Returns tx hash."""
|
|
pass
|
|
|
|
def _generate_deployment_id(self, chain: str, symbol: str, tx_hash: str) -> str:
|
|
"""Generate unique deployment ID."""
|
|
raw = f"{chain}:{symbol}:{tx_hash}:{time.time()}"
|
|
return hashlib.sha256(raw.encode()).hexdigest()[:16]
|
|
|
|
|
|
# ── EVM Deployer (Ethereum, Base, BSC) ─────────────────────────
|
|
|
|
|
|
class EVMDeployer(ChainDeployer):
|
|
"""
|
|
Deploy ERC-20 tokens on EVM chains (Ethereum, Base, BSC).
|
|
Uses web3.py for contract deployment and interaction.
|
|
"""
|
|
|
|
# Standard ERC-20 bytecode + ABI (minimal, optimized)
|
|
# This is a compact ERC-20 with mint/burn/ownership from OpenZeppelin patterns
|
|
ERC20_BYTECODE = "0x" + (
|
|
"608060405234801561001057600080fd5b50604051610d82380380610d82833981016040819052"
|
|
"61002f9161007d565b600380546001600160a01b03191633179055604051819061005090610070"
|
|
"565b6001600160a01b039091168152602001604051809103906000f08015801561007a573d6000"
|
|
"803e3d6000fd5b50506100a9565b6000806040838503121561009057600080fd5b505080516020"
|
|
"909101519092909150565b610cc5806100b86000396000f3fe"
|
|
)
|
|
|
|
def __init__(self, chain: str, private_key: str, rpc_url: str):
|
|
from eth_account import Account
|
|
from web3 import Web3
|
|
|
|
self.w3 = Web3(Web3.HTTPProvider(rpc_url))
|
|
self.account = Account.from_key(private_key)
|
|
super().__init__(chain, private_key, rpc_url)
|
|
|
|
def _derive_address(self) -> str:
|
|
return self.account.address
|
|
|
|
def _build_contract_code(self, params: DeployParams) -> tuple[str, str]:
|
|
"""Build contract bytecode with constructor args embedded."""
|
|
# For production, we use a pre-compiled factory or deploy via CREATE2
|
|
# Here we construct a minimal ERC-20 with the parameters
|
|
params.name.encode("utf-8").hex()
|
|
params.symbol.encode("utf-8").hex()
|
|
decimals = params.decimals
|
|
supply = int(params.initial_supply)
|
|
owner = params.owner_address or self.deployer_address
|
|
|
|
# Build constructor args: (name, symbol, decimals, initialSupply, owner, mintable, burnable)
|
|
# This is a simplified approach — in production use a proper factory contract
|
|
from web3 import Web3
|
|
|
|
args_encoded = (
|
|
Web3.to_bytes(text=params.name).ljust(32, b"\x00").hex()
|
|
+ Web3.to_bytes(text=params.symbol).ljust(32, b"\x00").hex()
|
|
+ hex(decimals)[2:].zfill(64)
|
|
+ hex(supply)[2:].zfill(64)
|
|
+ owner[2:].zfill(64)
|
|
+ ("1" if params.mintable else "0").zfill(64)
|
|
+ ("1" if params.burnable else "0").zfill(64)
|
|
)
|
|
|
|
# For now, return a placeholder — real deployment uses factory contract
|
|
bytecode = self.ERC20_BYTECODE + args_encoded
|
|
return bytecode, owner
|
|
|
|
async def deploy_token(self, params: DeployParams) -> TokenDeployment:
|
|
"""Deploy ERC-20 token via factory or direct deploy."""
|
|
try:
|
|
# Check if factory contract is configured
|
|
factory_address = os.getenv(f"{self.chain.upper()}_TOKEN_FACTORY", "")
|
|
|
|
if factory_address:
|
|
return await self._deploy_via_factory(factory_address, params)
|
|
else:
|
|
return await self._deploy_direct(params)
|
|
except Exception as e:
|
|
logger.error(f"EVM deploy failed on {self.chain}: {e}")
|
|
raise
|
|
|
|
async def _deploy_direct(self, params: DeployParams) -> TokenDeployment:
|
|
"""Direct contract deployment (simplified — uses raw tx)."""
|
|
bytecode, owner = self._build_contract_code(params)
|
|
|
|
# Build and sign transaction
|
|
nonce = self.w3.eth.get_transaction_count(self.deployer_address)
|
|
gas_price = self.w3.eth.gas_price
|
|
|
|
tx = {
|
|
"from": self.deployer_address,
|
|
"nonce": nonce,
|
|
"gasPrice": gas_price,
|
|
"data": bytecode,
|
|
"chainId": self.w3.eth.chain_id,
|
|
}
|
|
|
|
# Estimate gas
|
|
try:
|
|
tx["gas"] = self.w3.eth.estimate_gas(tx)
|
|
except Exception:
|
|
tx["gas"] = 3000000 # fallback
|
|
|
|
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, timeout=120)
|
|
|
|
contract_address = receipt.contractAddress
|
|
|
|
deployment = TokenDeployment(
|
|
deployment_id=self._generate_deployment_id(self.chain, params.symbol, tx_hash.hex()),
|
|
chain=self.chain,
|
|
name=params.name,
|
|
symbol=params.symbol,
|
|
decimals=params.decimals,
|
|
total_supply=params.initial_supply,
|
|
contract_address=contract_address,
|
|
deployer_address=self.deployer_address,
|
|
tx_hash=tx_hash.hex(),
|
|
block_number=receipt.blockNumber,
|
|
owner_address=owner,
|
|
mintable=params.mintable,
|
|
burnable=params.burnable,
|
|
pausable=params.pausable,
|
|
max_supply=params.max_supply,
|
|
metadata_uri=params.metadata_uri,
|
|
)
|
|
|
|
logger.info(f"Deployed {params.symbol} on {self.chain} at {contract_address}")
|
|
return deployment
|
|
|
|
async def _deploy_via_factory(self, factory_address: str, params: DeployParams) -> TokenDeployment:
|
|
"""Deploy via factory contract for gas efficiency."""
|
|
# Factory ABI (minimal)
|
|
factory_abi = [
|
|
{
|
|
"inputs": [
|
|
{"name": "name", "type": "string"},
|
|
{"name": "symbol", "type": "string"},
|
|
{"name": "decimals", "type": "uint8"},
|
|
{"name": "initialSupply", "type": "uint256"},
|
|
{"name": "owner", "type": "address"},
|
|
{"name": "mintable", "type": "bool"},
|
|
{"name": "burnable", "type": "bool"},
|
|
],
|
|
"name": "createToken",
|
|
"outputs": [{"name": "tokenAddress", "type": "address"}],
|
|
"type": "function",
|
|
}
|
|
]
|
|
|
|
factory = self.w3.eth.contract(address=factory_address, abi=factory_abi)
|
|
owner = params.owner_address or self.deployer_address
|
|
supply = int(params.initial_supply)
|
|
|
|
tx = factory.functions.createToken(
|
|
params.name,
|
|
params.symbol,
|
|
params.decimals,
|
|
supply,
|
|
owner,
|
|
params.mintable,
|
|
params.burnable,
|
|
).build_transaction(
|
|
{
|
|
"from": self.deployer_address,
|
|
"nonce": self.w3.eth.get_transaction_count(self.deployer_address),
|
|
"gasPrice": self.w3.eth.gas_price,
|
|
"chainId": self.w3.eth.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, timeout=120)
|
|
|
|
# Parse event log for token address
|
|
contract_address = None
|
|
for log in receipt.logs:
|
|
if log.address.lower() == factory_address.lower():
|
|
# TokenCreated event — 32 bytes padded address at position 0
|
|
contract_address = "0x" + log.data[-40:]
|
|
break
|
|
|
|
if not contract_address:
|
|
raise RuntimeError("Factory deployment failed — no TokenCreated event found")
|
|
|
|
deployment = TokenDeployment(
|
|
deployment_id=self._generate_deployment_id(self.chain, params.symbol, tx_hash.hex()),
|
|
chain=self.chain,
|
|
name=params.name,
|
|
symbol=params.symbol,
|
|
decimals=params.decimals,
|
|
total_supply=params.initial_supply,
|
|
contract_address=contract_address,
|
|
deployer_address=self.deployer_address,
|
|
tx_hash=tx_hash.hex(),
|
|
block_number=receipt.blockNumber,
|
|
owner_address=owner,
|
|
mintable=params.mintable,
|
|
burnable=params.burnable,
|
|
pausable=params.pausable,
|
|
max_supply=params.max_supply,
|
|
metadata_uri=params.metadata_uri,
|
|
)
|
|
|
|
logger.info(f"Factory-deployed {params.symbol} on {self.chain} at {contract_address}")
|
|
return deployment
|
|
|
|
async def mint_tokens(self, contract_address: str, to_address: str, amount: str) -> str:
|
|
"""Mint tokens to an address."""
|
|
erc20_abi = [
|
|
{
|
|
"inputs": [
|
|
{"name": "to", "type": "address"},
|
|
{"name": "amount", "type": "uint256"},
|
|
],
|
|
"name": "mint",
|
|
"outputs": [],
|
|
"type": "function",
|
|
}
|
|
]
|
|
|
|
contract = self.w3.eth.contract(address=contract_address, abi=erc20_abi)
|
|
tx = contract.functions.mint(to_address, int(amount)).build_transaction(
|
|
{
|
|
"from": self.deployer_address,
|
|
"nonce": self.w3.eth.get_transaction_count(self.deployer_address),
|
|
"gasPrice": self.w3.eth.gas_price,
|
|
"chainId": self.w3.eth.chain_id,
|
|
}
|
|
)
|
|
|
|
signed = self.w3.eth.account.sign_transaction(tx, self.private_key)
|
|
tx_hash = self.w3.eth.send_raw_transaction(signed.rawTransaction)
|
|
self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=60)
|
|
|
|
return tx_hash.hex()
|
|
|
|
async def burn_tokens(self, contract_address: str, amount: str) -> str:
|
|
"""Burn tokens from deployer balance."""
|
|
erc20_abi = [
|
|
{
|
|
"inputs": [{"name": "amount", "type": "uint256"}],
|
|
"name": "burn",
|
|
"outputs": [],
|
|
"type": "function",
|
|
}
|
|
]
|
|
|
|
contract = self.w3.eth.contract(address=contract_address, abi=erc20_abi)
|
|
tx = contract.functions.burn(int(amount)).build_transaction(
|
|
{
|
|
"from": self.deployer_address,
|
|
"nonce": self.w3.eth.get_transaction_count(self.deployer_address),
|
|
"gasPrice": self.w3.eth.gas_price,
|
|
"chainId": self.w3.eth.chain_id,
|
|
}
|
|
)
|
|
|
|
signed = self.w3.eth.account.sign_transaction(tx, self.private_key)
|
|
tx_hash = self.w3.eth.send_raw_transaction(signed.rawTransaction)
|
|
self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=60)
|
|
|
|
return tx_hash.hex()
|
|
|
|
async def transfer_ownership(self, contract_address: str, new_owner: str) -> str:
|
|
"""Transfer contract ownership."""
|
|
ownable_abi = [
|
|
{
|
|
"inputs": [{"name": "newOwner", "type": "address"}],
|
|
"name": "transferOwnership",
|
|
"outputs": [],
|
|
"type": "function",
|
|
}
|
|
]
|
|
|
|
contract = self.w3.eth.contract(address=contract_address, abi=ownable_abi)
|
|
tx = contract.functions.transferOwnership(new_owner).build_transaction(
|
|
{
|
|
"from": self.deployer_address,
|
|
"nonce": self.w3.eth.get_transaction_count(self.deployer_address),
|
|
"gasPrice": self.w3.eth.gas_price,
|
|
"chainId": self.w3.eth.chain_id,
|
|
}
|
|
)
|
|
|
|
signed = self.w3.eth.account.sign_transaction(tx, self.private_key)
|
|
tx_hash = self.w3.eth.send_raw_transaction(signed.rawTransaction)
|
|
self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=60)
|
|
|
|
return tx_hash.hex()
|
|
|
|
async def renounce_ownership(self, contract_address: str) -> str:
|
|
"""Renounce ownership (make contract immutable)."""
|
|
ownable_abi = [
|
|
{
|
|
"inputs": [],
|
|
"name": "renounceOwnership",
|
|
"outputs": [],
|
|
"type": "function",
|
|
}
|
|
]
|
|
|
|
contract = self.w3.eth.contract(address=contract_address, abi=ownable_abi)
|
|
tx = contract.functions.renounceOwnership().build_transaction(
|
|
{
|
|
"from": self.deployer_address,
|
|
"nonce": self.w3.eth.get_transaction_count(self.deployer_address),
|
|
"gasPrice": self.w3.eth.gas_price,
|
|
"chainId": self.w3.eth.chain_id,
|
|
}
|
|
)
|
|
|
|
signed = self.w3.eth.account.sign_transaction(tx, self.private_key)
|
|
tx_hash = self.w3.eth.send_raw_transaction(signed.rawTransaction)
|
|
self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=60)
|
|
|
|
return tx_hash.hex()
|
|
|
|
async def get_token_info(self, contract_address: str) -> dict[str, Any]:
|
|
"""Read token metadata from contract."""
|
|
erc20_abi = [
|
|
{"inputs": [], "name": "name", "outputs": [{"type": "string"}], "type": "function"},
|
|
{"inputs": [], "name": "symbol", "outputs": [{"type": "string"}], "type": "function"},
|
|
{"inputs": [], "name": "decimals", "outputs": [{"type": "uint8"}], "type": "function"},
|
|
{
|
|
"inputs": [],
|
|
"name": "totalSupply",
|
|
"outputs": [{"type": "uint256"}],
|
|
"type": "function",
|
|
},
|
|
{"inputs": [], "name": "owner", "outputs": [{"type": "address"}], "type": "function"},
|
|
]
|
|
|
|
contract = self.w3.eth.contract(address=contract_address, abi=erc20_abi)
|
|
|
|
return {
|
|
"name": contract.functions.name().call(),
|
|
"symbol": contract.functions.symbol().call(),
|
|
"decimals": contract.functions.decimals().call(),
|
|
"total_supply": str(contract.functions.totalSupply().call()),
|
|
"owner": contract.functions.owner().call(),
|
|
}
|
|
|
|
async def get_balance(self, contract_address: str, wallet_address: str) -> str:
|
|
"""Get token balance."""
|
|
erc20_abi = [
|
|
{
|
|
"inputs": [{"name": "account", "type": "address"}],
|
|
"name": "balanceOf",
|
|
"outputs": [{"type": "uint256"}],
|
|
"type": "function",
|
|
}
|
|
]
|
|
|
|
contract = self.w3.eth.contract(address=contract_address, abi=erc20_abi)
|
|
balance = contract.functions.balanceOf(wallet_address).call()
|
|
return str(balance)
|
|
|
|
async def blacklist_add(self, contract_address: str, address: str) -> str:
|
|
"""Add address to blacklist."""
|
|
blacklist_abi = [
|
|
{
|
|
"inputs": [{"name": "account", "type": "address"}],
|
|
"name": "blacklist",
|
|
"outputs": [],
|
|
"type": "function",
|
|
}
|
|
]
|
|
contract = self.w3.eth.contract(address=contract_address, abi=blacklist_abi)
|
|
tx = contract.functions.blacklist(address).build_transaction(
|
|
{
|
|
"from": self.deployer_address,
|
|
"nonce": self.w3.eth.get_transaction_count(self.deployer_address),
|
|
"gasPrice": self.w3.eth.gas_price,
|
|
"chainId": self.w3.eth.chain_id,
|
|
}
|
|
)
|
|
signed = self.w3.eth.account.sign_transaction(tx, self.private_key)
|
|
tx_hash = self.w3.eth.send_raw_transaction(signed.rawTransaction)
|
|
self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=60)
|
|
return tx_hash.hex()
|
|
|
|
async def blacklist_remove(self, contract_address: str, address: str) -> str:
|
|
"""Remove address from blacklist."""
|
|
unblacklist_abi = [
|
|
{
|
|
"inputs": [{"name": "account", "type": "address"}],
|
|
"name": "unblacklist",
|
|
"outputs": [],
|
|
"type": "function",
|
|
}
|
|
]
|
|
contract = self.w3.eth.contract(address=contract_address, abi=unblacklist_abi)
|
|
tx = contract.functions.unblacklist(address).build_transaction(
|
|
{
|
|
"from": self.deployer_address,
|
|
"nonce": self.w3.eth.get_transaction_count(self.deployer_address),
|
|
"gasPrice": self.w3.eth.gas_price,
|
|
"chainId": self.w3.eth.chain_id,
|
|
}
|
|
)
|
|
signed = self.w3.eth.account.sign_transaction(tx, self.private_key)
|
|
tx_hash = self.w3.eth.send_raw_transaction(signed.rawTransaction)
|
|
self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=60)
|
|
return tx_hash.hex()
|
|
|
|
async def is_blacklisted(self, contract_address: str, address: str) -> bool:
|
|
"""Check if address is blacklisted."""
|
|
check_abi = [
|
|
{
|
|
"inputs": [{"name": "account", "type": "address"}],
|
|
"name": "isBlacklisted",
|
|
"outputs": [{"type": "bool"}],
|
|
"type": "function",
|
|
}
|
|
]
|
|
contract = self.w3.eth.contract(address=contract_address, abi=check_abi)
|
|
return contract.functions.isBlacklisted(address).call()
|
|
|
|
async def set_trading_enabled(self, contract_address: str, enabled: bool) -> str:
|
|
"""Enable or disable trading."""
|
|
trading_abi = [
|
|
{
|
|
"inputs": [{"name": "enabled", "type": "bool"}],
|
|
"name": "setTradingEnabled",
|
|
"outputs": [],
|
|
"type": "function",
|
|
}
|
|
]
|
|
contract = self.w3.eth.contract(address=contract_address, abi=trading_abi)
|
|
tx = contract.functions.setTradingEnabled(enabled).build_transaction(
|
|
{
|
|
"from": self.deployer_address,
|
|
"nonce": self.w3.eth.get_transaction_count(self.deployer_address),
|
|
"gasPrice": self.w3.eth.gas_price,
|
|
"chainId": self.w3.eth.chain_id,
|
|
}
|
|
)
|
|
signed = self.w3.eth.account.sign_transaction(tx, self.private_key)
|
|
tx_hash = self.w3.eth.send_raw_transaction(signed.rawTransaction)
|
|
self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=60)
|
|
return tx_hash.hex()
|
|
|
|
async def set_max_wallet(self, contract_address: str, max_amount: str) -> str:
|
|
"""Set max wallet limit."""
|
|
limit_abi = [
|
|
{
|
|
"inputs": [{"name": "maxAmount", "type": "uint256"}],
|
|
"name": "setMaxWalletAmount",
|
|
"outputs": [],
|
|
"type": "function",
|
|
}
|
|
]
|
|
contract = self.w3.eth.contract(address=contract_address, abi=limit_abi)
|
|
tx = contract.functions.setMaxWalletAmount(int(max_amount)).build_transaction(
|
|
{
|
|
"from": self.deployer_address,
|
|
"nonce": self.w3.eth.get_transaction_count(self.deployer_address),
|
|
"gasPrice": self.w3.eth.gas_price,
|
|
"chainId": self.w3.eth.chain_id,
|
|
}
|
|
)
|
|
signed = self.w3.eth.account.sign_transaction(tx, self.private_key)
|
|
tx_hash = self.w3.eth.send_raw_transaction(signed.rawTransaction)
|
|
self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=60)
|
|
return tx_hash.hex()
|
|
|
|
async def set_max_tx(self, contract_address: str, max_amount: str) -> str:
|
|
"""Set max transaction limit."""
|
|
tx_abi = [
|
|
{
|
|
"inputs": [{"name": "maxAmount", "type": "uint256"}],
|
|
"name": "setMaxTxAmount",
|
|
"outputs": [],
|
|
"type": "function",
|
|
}
|
|
]
|
|
contract = self.w3.eth.contract(address=contract_address, abi=tx_abi)
|
|
tx = contract.functions.setMaxTxAmount(int(max_amount)).build_transaction(
|
|
{
|
|
"from": self.deployer_address,
|
|
"nonce": self.w3.eth.get_transaction_count(self.deployer_address),
|
|
"gasPrice": self.w3.eth.gas_price,
|
|
"chainId": self.w3.eth.chain_id,
|
|
}
|
|
)
|
|
signed = self.w3.eth.account.sign_transaction(tx, self.private_key)
|
|
tx_hash = self.w3.eth.send_raw_transaction(signed.rawTransaction)
|
|
self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=60)
|
|
return tx_hash.hex()
|
|
|
|
|
|
# ── Solana Deployer (SPL Token) ───────────────────────────────
|
|
|
|
|
|
class SolanaDeployer(ChainDeployer):
|
|
"""
|
|
Deploy SPL tokens on Solana.
|
|
Uses solana-py for token creation and management.
|
|
"""
|
|
|
|
def __init__(self, private_key: str, rpc_url: str):
|
|
from solana.rpc.api import Client
|
|
from solders.keypair import Keypair
|
|
|
|
self.client = Client(rpc_url)
|
|
# Private key is base58-encoded
|
|
self.keypair = Keypair.from_base58_string(private_key)
|
|
super().__init__("solana", private_key, rpc_url)
|
|
|
|
def _derive_address(self) -> str:
|
|
return str(self.keypair.pubkey())
|
|
|
|
async def deploy_token(self, params: DeployParams) -> TokenDeployment:
|
|
"""Create a new SPL token mint."""
|
|
from solana.transaction import Transaction
|
|
from solders.pubkey import Pubkey
|
|
from solders.system_program import CreateAccountParams, create_account
|
|
from spl.token.constants import TOKEN_PROGRAM_ID
|
|
from spl.token.instructions import create_mint, get_associated_token_address, mint_to
|
|
|
|
try:
|
|
# Create mint account
|
|
mint = Keypair()
|
|
|
|
# Get rent-exempt balance for mint
|
|
rent = self.client.get_minimum_balance_for_rent_exemption(82)["result"]
|
|
|
|
# Build transaction
|
|
tx = Transaction()
|
|
|
|
# Create account for mint
|
|
create_account_ix = create_account(
|
|
CreateAccountParams(
|
|
from_pubkey=self.keypair.pubkey(),
|
|
to_pubkey=mint.pubkey(),
|
|
lamports=rent,
|
|
space=82,
|
|
program_id=TOKEN_PROGRAM_ID,
|
|
)
|
|
)
|
|
tx.add(create_account_ix)
|
|
|
|
# Initialize mint
|
|
init_mint_ix = create_mint(
|
|
mint=mint.pubkey(),
|
|
decimals=params.decimals,
|
|
mint_authority=self.keypair.pubkey(),
|
|
freeze_authority=Pubkey.from_string(params.freeze_authority) if params.freeze_authority else None,
|
|
program_id=TOKEN_PROGRAM_ID,
|
|
)
|
|
tx.add(init_mint_ix)
|
|
|
|
# Create associated token account for deployer
|
|
ata = get_associated_token_address(self.keypair.pubkey(), mint.pubkey())
|
|
|
|
# Mint initial supply to deployer
|
|
mint_to_ix = mint_to(
|
|
mint=mint.pubkey(),
|
|
dest=ata,
|
|
mint_authority=self.keypair.pubkey(),
|
|
amount=int(params.initial_supply) * (10**params.decimals),
|
|
program_id=TOKEN_PROGRAM_ID,
|
|
)
|
|
tx.add(mint_to_ix)
|
|
|
|
# Send transaction
|
|
tx.sign(self.keypair, mint)
|
|
result = self.client.send_transaction(tx, self.keypair, mint)
|
|
|
|
tx_hash = result["result"]
|
|
|
|
deployment = TokenDeployment(
|
|
deployment_id=self._generate_deployment_id("solana", params.symbol, tx_hash),
|
|
chain="solana",
|
|
name=params.name,
|
|
symbol=params.symbol,
|
|
decimals=params.decimals,
|
|
total_supply=params.initial_supply,
|
|
contract_address=str(mint.pubkey()),
|
|
deployer_address=self.deployer_address,
|
|
tx_hash=tx_hash,
|
|
owner_address=self.deployer_address,
|
|
mintable=params.mintable,
|
|
burnable=params.burnable,
|
|
pausable=params.pausable,
|
|
max_supply=params.max_supply,
|
|
metadata_uri=params.metadata_uri,
|
|
)
|
|
|
|
logger.info(f"Deployed SPL token {params.symbol} on Solana at {mint.pubkey()}")
|
|
return deployment
|
|
|
|
except Exception as e:
|
|
logger.error(f"Solana deploy failed: {e}")
|
|
raise
|
|
|
|
async def mint_tokens(self, contract_address: str, to_address: str, amount: str) -> str:
|
|
"""Mint SPL tokens to an address."""
|
|
from solana.transaction import Transaction
|
|
from solders.pubkey import Pubkey
|
|
from spl.token.constants import TOKEN_PROGRAM_ID
|
|
from spl.token.instructions import get_associated_token_address, mint_to
|
|
|
|
mint = Pubkey.from_string(contract_address)
|
|
dest = get_associated_token_address(Pubkey.from_string(to_address), mint)
|
|
|
|
tx = Transaction()
|
|
mint_to_ix = mint_to(
|
|
mint=mint,
|
|
dest=dest,
|
|
mint_authority=self.keypair.pubkey(),
|
|
amount=int(amount),
|
|
program_id=TOKEN_PROGRAM_ID,
|
|
)
|
|
tx.add(mint_to_ix)
|
|
tx.sign(self.keypair)
|
|
|
|
result = self.client.send_transaction(tx, self.keypair)
|
|
return result["result"]
|
|
|
|
async def burn_tokens(self, contract_address: str, amount: str) -> str:
|
|
"""Burn SPL tokens from deployer ATA."""
|
|
from solana.transaction import Transaction
|
|
from solders.pubkey import Pubkey
|
|
from spl.token.constants import TOKEN_PROGRAM_ID
|
|
from spl.token.instructions import burn
|
|
|
|
mint = Pubkey.from_string(contract_address)
|
|
ata = get_associated_token_address(self.keypair.pubkey(), mint)
|
|
|
|
tx = Transaction()
|
|
burn_ix = burn(
|
|
account=ata,
|
|
mint=mint,
|
|
owner=self.keypair.pubkey(),
|
|
amount=int(amount),
|
|
program_id=TOKEN_PROGRAM_ID,
|
|
)
|
|
tx.add(burn_ix)
|
|
tx.sign(self.keypair)
|
|
|
|
result = self.client.send_transaction(tx, self.keypair)
|
|
return result["result"]
|
|
|
|
async def transfer_ownership(self, contract_address: str, new_owner: str) -> str:
|
|
"""Transfer mint authority on Solana."""
|
|
from solana.transaction import Transaction
|
|
from solders.pubkey import Pubkey
|
|
from spl.token.constants import TOKEN_PROGRAM_ID
|
|
from spl.token.instructions import AuthorityType, set_authority
|
|
|
|
mint = Pubkey.from_string(contract_address)
|
|
new_authority = Pubkey.from_string(new_owner)
|
|
|
|
tx = Transaction()
|
|
set_auth_ix = set_authority(
|
|
account=mint,
|
|
current_authority=self.keypair.pubkey(),
|
|
authority_type=AuthorityType.MINT_TOKENS,
|
|
new_authority=new_authority,
|
|
program_id=TOKEN_PROGRAM_ID,
|
|
)
|
|
tx.add(set_auth_ix)
|
|
tx.sign(self.keypair)
|
|
|
|
result = self.client.send_transaction(tx, self.keypair)
|
|
return result["result"]
|
|
|
|
async def renounce_ownership(self, contract_address: str) -> str:
|
|
"""Renounce mint authority (disable minting)."""
|
|
from solana.transaction import Transaction
|
|
from solders.pubkey import Pubkey
|
|
from spl.token.constants import TOKEN_PROGRAM_ID
|
|
from spl.token.instructions import AuthorityType, set_authority
|
|
|
|
mint = Pubkey.from_string(contract_address)
|
|
|
|
tx = Transaction()
|
|
set_auth_ix = set_authority(
|
|
account=mint,
|
|
current_authority=self.keypair.pubkey(),
|
|
authority_type=AuthorityType.MINT_TOKENS,
|
|
new_authority=None, # None = renounce
|
|
program_id=TOKEN_PROGRAM_ID,
|
|
)
|
|
tx.add(set_auth_ix)
|
|
tx.sign(self.keypair)
|
|
|
|
result = self.client.send_transaction(tx, self.keypair)
|
|
return result["result"]
|
|
|
|
async def get_token_info(self, contract_address: str) -> dict[str, Any]:
|
|
"""Get SPL token metadata."""
|
|
from solders.pubkey import Pubkey
|
|
|
|
mint = Pubkey.from_string(contract_address)
|
|
account_info = self.client.get_account_info(mint)
|
|
|
|
# Parse mint data
|
|
data = account_info["result"]["value"]["data"][0]
|
|
# Decode base64 mint data
|
|
import base64
|
|
|
|
raw = base64.b64decode(data)
|
|
|
|
# Mint layout: mint_authority_option (1), mint_authority (32), supply (8), decimals (1), ...
|
|
decimals = raw[44]
|
|
supply = int.from_bytes(raw[36:44], "little")
|
|
|
|
# Get metadata if available
|
|
metadata = {}
|
|
try:
|
|
from metaplex.metadata import get_metadata
|
|
|
|
metadata = get_metadata(self.client, contract_address)
|
|
except Exception:
|
|
pass
|
|
|
|
return {
|
|
"address": contract_address,
|
|
"decimals": decimals,
|
|
"supply": str(supply),
|
|
"metadata": metadata,
|
|
}
|
|
|
|
async def get_balance(self, contract_address: str, wallet_address: str) -> str:
|
|
"""Get SPL token balance."""
|
|
from solders.pubkey import Pubkey
|
|
from spl.token.instructions import get_associated_token_address
|
|
|
|
mint = Pubkey.from_string(contract_address)
|
|
wallet = Pubkey.from_string(wallet_address)
|
|
ata = get_associated_token_address(wallet, mint)
|
|
|
|
balance = self.client.get_token_account_balance(str(ata))
|
|
return balance["result"]["value"]["amount"]
|
|
|
|
async def blacklist_add(self, contract_address: str, address: str) -> str:
|
|
"""Freeze SPL token account (Solana's equivalent of blacklist)."""
|
|
from solana.transaction import Transaction
|
|
from solders.pubkey import Pubkey
|
|
from spl.token.constants import TOKEN_PROGRAM_ID
|
|
from spl.token.instructions import freeze_account
|
|
|
|
mint = Pubkey.from_string(contract_address)
|
|
account = get_associated_token_address(Pubkey.from_string(address), mint)
|
|
|
|
tx = Transaction()
|
|
freeze_ix = freeze_account(
|
|
account=account,
|
|
mint=mint,
|
|
owner=self.keypair.pubkey(),
|
|
freeze_authority=self.keypair.pubkey(),
|
|
program_id=TOKEN_PROGRAM_ID,
|
|
)
|
|
tx.add(freeze_ix)
|
|
tx.sign(self.keypair)
|
|
|
|
result = self.client.send_transaction(tx, self.keypair)
|
|
return result["result"]
|
|
|
|
async def blacklist_remove(self, contract_address: str, address: str) -> str:
|
|
"""Thaw (unfreeze) SPL token account."""
|
|
from solana.transaction import Transaction
|
|
from solders.pubkey import Pubkey
|
|
from spl.token.constants import TOKEN_PROGRAM_ID
|
|
from spl.token.instructions import thaw_account
|
|
|
|
mint = Pubkey.from_string(contract_address)
|
|
account = get_associated_token_address(Pubkey.from_string(address), mint)
|
|
|
|
tx = Transaction()
|
|
thaw_ix = thaw_account(
|
|
account=account,
|
|
mint=mint,
|
|
owner=self.keypair.pubkey(),
|
|
freeze_authority=self.keypair.pubkey(),
|
|
program_id=TOKEN_PROGRAM_ID,
|
|
)
|
|
tx.add(thaw_ix)
|
|
tx.sign(self.keypair)
|
|
|
|
result = self.client.send_transaction(tx, self.keypair)
|
|
return result["result"]
|
|
|
|
async def is_blacklisted(self, contract_address: str, address: str) -> bool:
|
|
"""Check if SPL token account is frozen."""
|
|
from solders.pubkey import Pubkey
|
|
from spl.token.instructions import get_associated_token_address
|
|
|
|
mint = Pubkey.from_string(contract_address)
|
|
account = get_associated_token_address(Pubkey.from_string(address), mint)
|
|
|
|
info = self.client.get_account_info(str(account))
|
|
if info["result"]["value"]:
|
|
data = info["result"]["value"]["data"][0]
|
|
import base64
|
|
|
|
raw = base64.b64decode(data)
|
|
# State is at byte 64 (0 = uninitialized, 1 = initialized, 2 = frozen)
|
|
return raw[64] == 2 if len(raw) > 64 else False
|
|
return False
|
|
|
|
async def set_trading_enabled(self, contract_address: str, enabled: bool) -> str:
|
|
"""Enable/disable trading by setting mint authority."""
|
|
if enabled:
|
|
return "Trading already enabled (mint authority active)"
|
|
# Disable minting = no new tokens = trading still works but supply fixed
|
|
return await self.renounce_ownership(contract_address)
|
|
|
|
async def set_max_wallet(self, contract_address: str, max_amount: str) -> str:
|
|
"""Not natively supported on SPL — would require custom program."""
|
|
logger.warning("Max wallet not natively supported on Solana SPL")
|
|
return "not_supported"
|
|
|
|
async def set_max_tx(self, contract_address: str, max_amount: str) -> str:
|
|
"""Not natively supported on SPL — would require custom program."""
|
|
logger.warning("Max tx not natively supported on Solana SPL")
|
|
return "not_supported"
|
|
|
|
|
|
# ── TRON Deployer (TRC-20) ─────────────────────────────────────
|
|
|
|
|
|
class TronDeployer(ChainDeployer):
|
|
"""
|
|
Deploy TRC-20 tokens on TRON.
|
|
Uses tronpy for contract deployment and interaction.
|
|
"""
|
|
|
|
def __init__(self, private_key: str, rpc_url: str = "https://api.trongrid.io"):
|
|
from tronpy import Tron
|
|
from tronpy.keys import PrivateKey
|
|
|
|
self.client = Tron(network="mainnet" if "mainnet" in rpc_url else "shasta")
|
|
self.private_key_obj = PrivateKey(bytes.fromhex(private_key.replace("0x", "")))
|
|
super().__init__("tron", private_key, rpc_url)
|
|
|
|
def _derive_address(self) -> str:
|
|
return self.private_key_obj.public_key.to_base58check_address()
|
|
|
|
async def deploy_token(self, params: DeployParams) -> TokenDeployment:
|
|
"""Deploy TRC-20 token on TRON."""
|
|
try:
|
|
# TRON uses TVM (Tron Virtual Machine) — similar to EVM
|
|
# For production, deploy via a factory or use pre-compiled bytecode
|
|
|
|
# Simplified: deploy via TronGrid API call
|
|
# In practice, you'd compile a TRC-20 contract and deploy it
|
|
|
|
# Placeholder for actual deployment
|
|
# Real implementation requires compiling Solidity to TVM bytecode
|
|
|
|
# For now, create a deployment record that can be used with
|
|
# a TRON deployment service or manual deployment
|
|
|
|
deployment_id = self._generate_deployment_id("tron", params.symbol, str(time.time()))
|
|
|
|
logger.warning(
|
|
"TRON deployment requires manual contract compilation. "
|
|
"Use TRON Station or TronGrid for deployment, then record here."
|
|
)
|
|
|
|
# Return a pending deployment record
|
|
deployment = TokenDeployment(
|
|
deployment_id=deployment_id,
|
|
chain="tron",
|
|
name=params.name,
|
|
symbol=params.symbol,
|
|
decimals=params.decimals,
|
|
total_supply=params.initial_supply,
|
|
contract_address="", # To be filled after manual deployment
|
|
deployer_address=self.deployer_address,
|
|
tx_hash="",
|
|
status="pending",
|
|
owner_address=params.owner_address or self.deployer_address,
|
|
mintable=params.mintable,
|
|
burnable=params.burnable,
|
|
pausable=params.pausable,
|
|
max_supply=params.max_supply,
|
|
metadata_uri=params.metadata_uri,
|
|
extra={"note": "TRON deployment requires manual contract compilation and deployment"},
|
|
)
|
|
|
|
return deployment
|
|
|
|
except Exception as e:
|
|
logger.error(f"TRON deploy failed: {e}")
|
|
raise
|
|
|
|
async def mint_tokens(self, contract_address: str, to_address: str, amount: str) -> str:
|
|
"""Mint TRC-20 tokens."""
|
|
contract = self.client.get_contract(contract_address)
|
|
tx = contract.functions.mint(to_address, int(amount))
|
|
txb = tx.build().sign(self.private_key_obj)
|
|
result = txb.broadcast()
|
|
return result["txid"]
|
|
|
|
async def burn_tokens(self, contract_address: str, amount: str) -> str:
|
|
"""Burn TRC-20 tokens."""
|
|
contract = self.client.get_contract(contract_address)
|
|
tx = contract.functions.burn(int(amount))
|
|
txb = tx.build().sign(self.private_key_obj)
|
|
result = txb.broadcast()
|
|
return result["txid"]
|
|
|
|
async def transfer_ownership(self, contract_address: str, new_owner: str) -> str:
|
|
"""Transfer TRC-20 ownership."""
|
|
contract = self.client.get_contract(contract_address)
|
|
tx = contract.functions.transferOwnership(new_owner)
|
|
txb = tx.build().sign(self.private_key_obj)
|
|
result = txb.broadcast()
|
|
return result["txid"]
|
|
|
|
async def renounce_ownership(self, contract_address: str) -> str:
|
|
"""Renounce TRC-20 ownership."""
|
|
contract = self.client.get_contract(contract_address)
|
|
tx = contract.functions.renounceOwnership()
|
|
txb = tx.build().sign(self.private_key_obj)
|
|
result = txb.broadcast()
|
|
return result["txid"]
|
|
|
|
async def get_token_info(self, contract_address: str) -> dict[str, Any]:
|
|
"""Get TRC-20 token info."""
|
|
contract = self.client.get_contract(contract_address)
|
|
|
|
return {
|
|
"name": contract.functions.name().call(),
|
|
"symbol": contract.functions.symbol().call(),
|
|
"decimals": contract.functions.decimals().call(),
|
|
"total_supply": str(contract.functions.totalSupply().call()),
|
|
"owner": contract.functions.owner().call() if hasattr(contract.functions, "owner") else None,
|
|
}
|
|
|
|
async def get_balance(self, contract_address: str, wallet_address: str) -> str:
|
|
"""Get TRC-20 balance."""
|
|
contract = self.client.get_contract(contract_address)
|
|
balance = contract.functions.balanceOf(wallet_address).call()
|
|
return str(balance)
|
|
|
|
async def blacklist_add(self, contract_address: str, address: str) -> str:
|
|
"""Add address to TRC-20 blacklist."""
|
|
contract = self.client.get_contract(contract_address)
|
|
tx = contract.functions.blacklist(address)
|
|
txb = tx.build().sign(self.private_key_obj)
|
|
result = txb.broadcast()
|
|
return result["txid"]
|
|
|
|
async def blacklist_remove(self, contract_address: str, address: str) -> str:
|
|
"""Remove address from TRC-20 blacklist."""
|
|
contract = self.client.get_contract(contract_address)
|
|
tx = contract.functions.unblacklist(address)
|
|
txb = tx.build().sign(self.private_key_obj)
|
|
result = txb.broadcast()
|
|
return result["txid"]
|
|
|
|
async def is_blacklisted(self, contract_address: str, address: str) -> bool:
|
|
"""Check if address is blacklisted on TRC-20."""
|
|
contract = self.client.get_contract(contract_address)
|
|
return contract.functions.isBlacklisted(address).call()
|
|
|
|
async def set_trading_enabled(self, contract_address: str, enabled: bool) -> str:
|
|
"""Enable/disable trading on TRC-20."""
|
|
contract = self.client.get_contract(contract_address)
|
|
tx = contract.functions.setTradingEnabled(enabled)
|
|
txb = tx.build().sign(self.private_key_obj)
|
|
result = txb.broadcast()
|
|
return result["txid"]
|
|
|
|
async def set_max_wallet(self, contract_address: str, max_amount: str) -> str:
|
|
"""Set max wallet limit on TRC-20."""
|
|
contract = self.client.get_contract(contract_address)
|
|
tx = contract.functions.setMaxWalletAmount(int(max_amount))
|
|
txb = tx.build().sign(self.private_key_obj)
|
|
result = txb.broadcast()
|
|
return result["txid"]
|
|
|
|
async def set_max_tx(self, contract_address: str, max_amount: str) -> str:
|
|
"""Set max transaction limit on TRC-20."""
|
|
contract = self.client.get_contract(contract_address)
|
|
tx = contract.functions.setMaxTxAmount(int(max_amount))
|
|
txb = tx.build().sign(self.private_key_obj)
|
|
result = txb.broadcast()
|
|
return result["txid"]
|
|
|
|
|
|
# ── Factory ─────────────────────────────────────────────────────
|
|
|
|
|
|
class TokenDeployerFactory:
|
|
"""
|
|
Factory to create the right deployer for a given chain.
|
|
Reads private keys and RPC URLs from environment variables.
|
|
"""
|
|
|
|
CHAIN_CONFIG: ClassVar[dict] =
|
|
{
|
|
"ethereum": {
|
|
"deployer_class": EVMDeployer,
|
|
"rpc_env": "ETH_RPC_URL",
|
|
"key_env": "ETH_DEPLOYER_KEY",
|
|
"default_rpc": "https://eth.llamarpc.com",
|
|
},
|
|
"base": {
|
|
"deployer_class": EVMDeployer,
|
|
"rpc_env": "BASE_RPC_URL",
|
|
"key_env": "BASE_DEPLOYER_KEY",
|
|
"default_rpc": "https://base.llamarpc.com",
|
|
},
|
|
"bsc": {
|
|
"deployer_class": EVMDeployer,
|
|
"rpc_env": "BSC_RPC_URL",
|
|
"key_env": "BSC_DEPLOYER_KEY",
|
|
"default_rpc": "https://bsc-dataseed.binance.org",
|
|
},
|
|
"solana": {
|
|
"deployer_class": SolanaDeployer,
|
|
"rpc_env": "SOLANA_RPC_URL",
|
|
"key_env": "SOLANA_DEPLOYER_KEY",
|
|
"default_rpc": "https://api.mainnet-beta.solana.com",
|
|
},
|
|
"tron": {
|
|
"deployer_class": TronDeployer,
|
|
"rpc_env": "TRON_RPC_URL",
|
|
"key_env": "TRON_DEPLOYER_KEY",
|
|
"default_rpc": "https://api.trongrid.io",
|
|
},
|
|
}
|
|
|
|
@classmethod
|
|
def get_deployer(cls, chain: str) -> ChainDeployer:
|
|
"""Get deployer for a chain."""
|
|
chain = chain.lower()
|
|
config = cls.CHAIN_CONFIG.get(chain)
|
|
if not config:
|
|
raise ValueError(f"Unsupported chain: {chain}. Supported: {list(cls.CHAIN_CONFIG.keys())}")
|
|
|
|
rpc_url = os.getenv(config["rpc_env"], config["default_rpc"])
|
|
private_key = os.getenv(config["key_env"], "")
|
|
|
|
if not private_key:
|
|
raise ValueError(f"No deployer key configured for {chain}. Set {config['key_env']} in .env")
|
|
|
|
return config["deployer_class"](chain if config["deployer_class"] == EVMDeployer else "", private_key, rpc_url)
|
|
|
|
@classmethod
|
|
def list_supported_chains(cls) -> list[dict[str, str]]:
|
|
"""List all supported chains with their status."""
|
|
chains = []
|
|
for chain, config in cls.CHAIN_CONFIG.items():
|
|
has_key = bool(os.getenv(config["key_env"], ""))
|
|
chains.append(
|
|
{
|
|
"chain": chain,
|
|
"configured": has_key,
|
|
"rpc_env": config["rpc_env"],
|
|
"key_env": config["key_env"],
|
|
}
|
|
)
|
|
return chains
|
|
|
|
|
|
# ── Storage ─────────────────────────────────────────────────────
|
|
|
|
|
|
class DeploymentStorage:
|
|
"""
|
|
Store and retrieve deployment records.
|
|
Uses Redis as primary, Supabase as backup.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.redis = None
|
|
self.supabase = None
|
|
self._init_stores()
|
|
|
|
def _init_stores(self):
|
|
"""Initialize storage backends."""
|
|
try:
|
|
import redis.asyncio as redis_lib
|
|
|
|
self.redis = redis_lib.Redis(
|
|
host=os.getenv("REDIS_HOST", "localhost"),
|
|
port=int(os.getenv("REDIS_PORT", "6379")),
|
|
password=os.getenv("REDIS_PASSWORD", ""),
|
|
decode_responses=True,
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Redis not available for deployment storage: {e}")
|
|
|
|
try:
|
|
from supabase import create_client
|
|
|
|
supabase_url = os.getenv("SUPABASE_URL")
|
|
supabase_key = os.getenv("SUPABASE_SERVICE_KEY")
|
|
if supabase_url and supabase_key:
|
|
self.supabase = create_client(supabase_url, supabase_key)
|
|
except Exception as e:
|
|
logger.warning(f"Supabase not available for deployment storage: {e}")
|
|
|
|
async def save(self, deployment: TokenDeployment) -> bool:
|
|
"""Save deployment record."""
|
|
data = deployment.to_dict()
|
|
key = f"token_deployment:{deployment.deployment_id}"
|
|
|
|
success = False
|
|
|
|
# Redis
|
|
if self.redis:
|
|
try:
|
|
await self.redis.set(key, json.dumps(data))
|
|
await self.redis.sadd("token_deployments:all", deployment.deployment_id)
|
|
await self.redis.sadd(f"token_deployments:chain:{deployment.chain}", deployment.deployment_id)
|
|
success = True
|
|
except Exception as e:
|
|
logger.error(f"Redis save failed: {e}")
|
|
|
|
# Supabase
|
|
if self.supabase:
|
|
try:
|
|
self.supabase.table("token_deployments").upsert(data).execute()
|
|
success = True
|
|
except Exception as e:
|
|
logger.error(f"Supabase save failed: {e}")
|
|
|
|
return success
|
|
|
|
async def get(self, deployment_id: str) -> TokenDeployment | None:
|
|
"""Get deployment by ID."""
|
|
# Try Redis first
|
|
if self.redis:
|
|
try:
|
|
data = await self.redis.get(f"token_deployment:{deployment_id}")
|
|
if data:
|
|
d = json.loads(data)
|
|
return TokenDeployment(**d)
|
|
except Exception as e:
|
|
logger.error(f"Redis get failed: {e}")
|
|
|
|
# Fallback to Supabase
|
|
if self.supabase:
|
|
try:
|
|
result = (
|
|
self.supabase.table("token_deployments").select("*").eq("deployment_id", deployment_id).execute()
|
|
)
|
|
if result.data:
|
|
return TokenDeployment(**result.data[0])
|
|
except Exception as e:
|
|
logger.error(f"Supabase get failed: {e}")
|
|
|
|
return None
|
|
|
|
async def list_all(self, chain: str | None = None, limit: int = 100) -> list[TokenDeployment]:
|
|
"""List deployments, optionally filtered by chain."""
|
|
deployments = []
|
|
|
|
if self.redis:
|
|
try:
|
|
if chain:
|
|
ids = await self.redis.smembers(f"token_deployments:chain:{chain}")
|
|
else:
|
|
ids = await self.redis.smembers("token_deployments:all")
|
|
|
|
for dep_id in list(ids)[:limit]:
|
|
dep = await self.get(dep_id)
|
|
if dep:
|
|
deployments.append(dep)
|
|
except Exception as e:
|
|
logger.error(f"Redis list failed: {e}")
|
|
|
|
if not deployments and self.supabase:
|
|
try:
|
|
query = self.supabase.table("token_deployments").select("*").limit(limit)
|
|
if chain:
|
|
query = query.eq("chain", chain)
|
|
result = query.execute()
|
|
for row in result.data:
|
|
deployments.append(TokenDeployment(**row))
|
|
except Exception as e:
|
|
logger.error(f"Supabase list failed: {e}")
|
|
|
|
return deployments
|
|
|
|
async def update_status(self, deployment_id: str, status: str, extra: dict | None = None) -> bool:
|
|
"""Update deployment status."""
|
|
dep = await self.get(deployment_id)
|
|
if not dep:
|
|
return False
|
|
|
|
dep.status = status
|
|
dep.updated_at = datetime.utcnow().isoformat()
|
|
if extra:
|
|
dep.extra.update(extra)
|
|
|
|
return await self.save(dep)
|
|
|
|
|
|
# ── Singleton instances ───────────────────────────────────────
|
|
|
|
_storage: DeploymentStorage | None = None
|
|
|
|
|
|
async def get_storage() -> DeploymentStorage:
|
|
global _storage
|
|
if _storage is None:
|
|
_storage = DeploymentStorage()
|
|
return _storage
|