Algorand AI Contract Creator: Generating Smart Contracts with Natural Language

import openai
from typing import Dict, List

class ContractGenerator:
    def __init__(self, api_key: str):
        openai.api_key = api_key
        self.temperature = 0.2  # Low temp for deterministic output
    
    def generate_contract(self, description: str) -> Dict:
        """
        Generate PyTeal smart contract from natural language
        
        Args:
            description: Natural language contract description
            
        Returns:
            Dict with code, explanation, and metadata
        """
        system_prompt = """You are an expert PyTeal smart contract developer.
        
        Generate secure, production-ready Algorand smart contracts following these rules:
        1. Use PyTeal syntax correctly
        2. Implement proper access control
        3. Validate all inputs
        4. Handle edge cases
        5. Add clear comments
        6. Never hardcode private keys
        7. Use safe arithmetic operations
        8. Follow OWASP security guidelines
        
        Output format:
        - Contract code (PyTeal)
        - Line-by-line explanation
        - Security considerations
        - Test scenarios
        """
        
        user_prompt = f"""Generate a PyTeal smart contract for:

{description}

Include proper error handling, input validation, and security checks."""
        
        response = openai.ChatCompletion.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ],
            temperature=self.temperature,
            max_tokens=2000
        )
        
        return self.parse_response(response.choices[0].message.content)

import re
from typing import List, Tuple

class SecurityValidator:
    """
    Multi-layer security validation for generated contracts
    """
    
    DANGEROUS_PATTERNS = [
        (r'private_key\s*=\s*["\']', "Hardcoded private key detected"),
        (r'mnemonic\s*=\s*["\']', "Hardcoded mnemonic detected"),
        (r'Txn\.sender\(\)\s*==\s*["\']', "Hardcoded address check"),
        (r'\.send_raw_transaction', "Unsafe transaction sending"),
        (r'eval\(', "Code injection vulnerability"),
    ]
    
    REQUIRED_PATTERNS = [
        (r'Assert\(', "Missing input validation"),
        (r'Txn\.type_enum\(\)', "Missing transaction type check"),
        (r'Global\.group_size\(\)', "Missing group transaction validation"),
    ]
    
    def validate(self, code: str) -> Tuple[bool, List[str]]:
        """
        Validate contract code for security issues
        
        Returns:
            (is_safe, list_of_issues)
        """
        issues = []
        
        # Check for dangerous patterns
        for pattern, message in self.DANGEROUS_PATTERNS:
            if re.search(pattern, code, re.IGNORECASE):
                issues.append(f"CRITICAL: {message}")
        
        # Check for missing security features
        for pattern, message in self.REQUIRED_PATTERNS:
            if not re.search(pattern, code):
                issues.append(f"WARNING: {message}")
        
        # Check PyTeal syntax
        syntax_valid, syntax_errors = self.check_syntax(code)
        if not syntax_valid:
            issues.extend(syntax_errors)
        
        return len([i for i in issues if "CRITICAL" in i]) == 0, issues
    
    def check_syntax(self, code: str) -> Tuple[bool, List[str]]:
        """
        Validate PyTeal syntax by attempting compilation
        """
        try:
            # Create temporary module and compile
            from pyteal import *
            exec(code, globals())
            return True, []
        except Exception as e:
            return False, [f"Syntax error: {str(e)}"]

from pyteal import *
from algosdk.v2client import algod

class PyTealCompiler:
    """
    Compile PyTeal to TEAL and deploy to Algorand
    """
    
    def __init__(self, algod_token: str, algod_address: str):
        self.algod_client = algod.AlgodClient(algod_token, algod_address)
    
    def compile_pyteal(self, pyteal_code: str) -> str:
        """
        Compile PyTeal code to TEAL
        
        Returns:
            TEAL assembly code
        """
        # Execute PyTeal code to get AST
        local_vars = {}
        exec(pyteal_code, globals(), local_vars)
        
        # Find the approval program
        approval_program = local_vars.get('approval_program')
        if not approval_program:
            raise ValueError("No approval_program found in PyTeal code")
        
        # Compile to TEAL
        teal_code = compileTeal(approval_program, mode=Mode.Application, version=6)
        
        return teal_code
    
    def compile_teal(self, teal_code: str) -> bytes:
        """
        Compile TEAL to bytecode using Algorand node
        """
        response = self.algod_client.compile(teal_code)
        return base64.b64decode(response['result'])
    
    def deploy_contract(self, bytecode: bytes, sender_address: str, 
                       private_key: str) -> int:
        """
        Deploy compiled contract to Algorand TestNet
        
        Returns:
            Application ID
        """
        # Get suggested parameters
        params = self.algod_client.suggested_params()
        
        # Create application transaction
        txn = transaction.ApplicationCreateTxn(
            sender=sender_address,
            sp=params,
            on_complete=transaction.OnComplete.NoOpOC,
            approval_program=bytecode,
            clear_program=bytecode,  # Simplified for demo
            global_schema=transaction.StateSchema(num_uints=1, num_byte_slices=1),
            local_schema=transaction.StateSchema(num_uints=0, num_byte_slices=0)
        )
        
        # Sign and send
        signed_txn = txn.sign(private_key)
        tx_id = self.algod_client.send_transaction(signed_txn)
        
        # Wait for confirmation
        result = transaction.wait_for_confirmation(self.algod_client, tx_id, 4)
        
        return result['application-index']

class ValidationPipeline:
    """
    Three-stage validation process
    """
    
    def validate_contract(self, code: str) -> Dict:
        results = {
            'syntax': False,
            'security': False,
            'compilation': False,
            'issues': []
        }
        
        # Stage 1: Syntax validation
        syntax_valid, syntax_issues = self.check_syntax(code)
        results['syntax'] = syntax_valid
        results['issues'].extend(syntax_issues)
        
        if not syntax_valid:
            return results  # Stop if syntax invalid
        
        # Stage 2: Security scanning
        security_valid, security_issues = self.check_security(code)
        results['security'] = security_valid
        results['issues'].extend(security_issues)
        
        if not security_valid:
            return results  # Stop if security issues
        
        # Stage 3: Compilation test
        try:
            teal = self.compile(code)
            results['compilation'] = True
        except Exception as e:
            results['issues'].append(f"Compilation failed: {e}")
        
        return results

def generate_with_retry(description: str, max_retries: int = 3) -> str:
    """
    Attempt generation with automatic error correction
    """
    for attempt in range(max_retries):
        code = generator.generate(description)
        
        is_valid, issues = validator.validate(code)
        
        if is_valid:
            return code
        
        # Provide feedback to AI for correction
        feedback = f"""The generated code has these issues:
        {chr(10).join(issues)}
        
        Please fix these issues and regenerate the contract."""
        
        description = f"{description}\n\nPrevious attempt issues:\n{feedback}"
    
    raise Exception("Failed to generate valid contract after retries")

def explain_contract(code: str) -> Dict:
    """
    Generate human-readable explanation of contract
    """
    prompt = f"""Explain this PyTeal smart contract in simple terms:

{code}

Provide:
1. High-level purpose
2. Key functions
3. Security features
4. Usage instructions
5. Potential risks"""
    
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3
    )
    
    return parse_explanation(response.choices[0].message.content)

from pyteal import *

def approval_program():
    # Define escrow logic
    buyer = Txn.accounts[1]
    seller = Txn.accounts[2]
    amount = Int(5000000)  # 5 ALGO in microAlgos
    timeout = Int(604800)  # 7 days in seconds
    
    # Check if both parties approved
    both_approved = And(
        App.globalGet(Bytes("buyer_approved")),
        App.globalGet(Bytes("seller_approved"))
    )
    
    # Release funds
    release = Seq([
        Assert(both_approved),
        InnerTxnBuilder.Begin(),
        InnerTxnBuilder.SetFields({
            TxnField.type_enum: TxnType.Payment,
            TxnField.receiver: seller,
            TxnField.amount: amount,
        }),
        InnerTxnBuilder.Submit(),
        Approve()
    ])
    
    # Refund on timeout
    refund = Seq([
        Assert(Global.latest_timestamp() > App.globalGet(Bytes("deadline"))),
        InnerTxnBuilder.Begin(),
        InnerTxnBuilder.SetFields({
            TxnField.type_enum: TxnType.Payment,
            TxnField.receiver: buyer,
            TxnField.amount: amount,
        }),
        InnerTxnBuilder.Submit(),
        Approve()
    ])
    
    # Main logic
    return Cond(
        [Txn.application_id() == Int(0), Approve()],
        [Txn.on_completion() == OnComplete.DeleteApplication, Reject()],
        [Txn.application_args[0] == Bytes("approve"), handle_approval()],
        [Txn.application_args[0] == Bytes("release"), release],
        [Txn.application_args[0] == Bytes("refund"), refund],
    )

# Compile
approval_program()

class AuditLogger:
    """
    Log all contract generation for compliance
    """
    
    def log_generation(self, request: Dict, result: Dict):
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'user': request.get('user_id'),
            'description': request['description'],
            'generated_code': result['code'],
            'validation_result': result['validation'],
            'deployed': result.get('deployed', False),
            'app_id': result.get('app_id'),
            'security_issues': result.get('issues', [])
        }
        
        # Store in audit database
        self.db.insert('audit_log', log_entry)
        
        # Alert on security issues
        if any('CRITICAL' in issue for issue in log_entry['security_issues']):
            self.alert_security_team(log_entry)

import pytest
from algosdk import account

def test_escrow_approval():
    """Test escrow requires both approvals"""
    # Generate contract
    description = "Escrow requiring buyer and seller approval"
    contract = generator.generate(description)
    
    # Deploy to TestNet
    app_id = deployer.deploy(contract)
    
    # Test approval flow
    buyer_key, buyer_addr = account.generate_account()
    seller_key, seller_addr = account.generate_account()
    
    # Fund accounts
    fund_account(buyer_addr, 10_000_000)
    fund_account(seller_addr, 1_000_000)
    
    # Buyer approves
    approve_txn(app_id, buyer_key, "buyer")
    
    # Should not release yet
    with pytest.raises(Exception):
        release_funds(app_id)
    
    # Seller approves
    approve_txn(app_id, seller_key, "seller")
    
    # Now should release
    result = release_funds(app_id)
    assert result['confirmed']