#!/usr/bin/env python3
"""
Simple AITBC Blockchain Explorer - Demonstrating the issues described in the analysis
"""
import asyncio
import httpx
import re
from datetime import datetime
from typing import Dict, Any, Optional
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse
import uvicorn
app = FastAPI(title="Simple AITBC Explorer", version="0.1.0")
# Configuration
BLOCKCHAIN_RPC_URL = "http://localhost:8025"
# Validation patterns for user inputs to prevent SSRF
TX_HASH_PATTERN = re.compile(r'^[a-fA-F0-9]{64}$') # 64-character hex string for transaction hash
def validate_tx_hash(tx_hash: str) -> bool:
"""Validate transaction hash to prevent SSRF"""
if not tx_hash:
return False
# Check for path traversal or URL manipulation
if any(char in tx_hash for char in ['/', '\\', '..', '\n', '\r', '\t', '?', '&']):
return False
# Validate against hash pattern
return bool(TX_HASH_PATTERN.match(tx_hash))
# HTML Template with the problematic frontend
HTML_TEMPLATE = """
Simple AITBC Explorer
AITBC Blockchain Explorer
"""
# Problem 1: Only /api/chain/head and /api/blocks/{height} defined, missing /api/transactions/{hash}
@app.get("/api/chain/head")
async def get_chain_head():
"""Get current chain head"""
try:
async with httpx.AsyncClient() as client:
response = await client.get(f"{BLOCKCHAIN_RPC_URL}/rpc/head")
if response.status_code == 200:
return response.json()
except Exception as e:
print(f"Error getting chain head: {e}")
return {"height": 0, "hash": "", "timestamp": None}
@app.get("/api/blocks/{height}")
async def get_block(height: int):
"""Get block by height"""
# Validate height is non-negative and reasonable
if height < 0 or height > 10000000:
return {"height": height, "hash": "", "timestamp": None, "transactions": []}
try:
async with httpx.AsyncClient() as client:
response = await client.get(f"{BLOCKCHAIN_RPC_URL}/rpc/blocks/{height}")
if response.status_code == 200:
return response.json()
except Exception as e:
print(f"Error getting block {height}: {e}")
return {"height": height, "hash": "", "timestamp": None, "transactions": []}
@app.get("/api/transactions/{tx_hash}")
async def get_transaction(tx_hash: str):
"""Get transaction by hash - Problem 1: This endpoint was missing"""
if not validate_tx_hash(tx_hash):
return {"hash": tx_hash, "from": "unknown", "to": "unknown", "amount": 0, "timestamp": None}
try:
async with httpx.AsyncClient() as client:
response = await client.get(f"{BLOCKCHAIN_RPC_URL}/rpc/tx/{tx_hash}")
if response.status_code == 200:
tx_data = response.json()
# Problem 2: Map RPC schema to UI schema
return {
"hash": tx_data.get("tx_hash", tx_hash), # tx_hash -> hash
"from": tx_data.get("sender", "unknown"), # sender -> from
"to": tx_data.get("recipient", "unknown"), # recipient -> to
"amount": tx_data.get("payload", {}).get("value", "0"), # payload.value -> amount
"fee": tx_data.get("payload", {}).get("fee", "0"), # payload.fee -> fee
"timestamp": tx_data.get("created_at"), # created_at -> timestamp
"block_height": tx_data.get("block_height", "pending")
}
elif response.status_code == 404:
raise HTTPException(status_code=404, detail="Transaction not found")
except HTTPException:
raise
except Exception as e:
print(f"Error getting transaction {tx_hash}: {e}")
raise HTTPException(status_code=500, detail=f"Failed to fetch transaction: {str(e)}")
# Missing: @app.get("/api/transactions/{tx_hash}") - THIS IS THE PROBLEM
@app.get("/", response_class=HTMLResponse)
async def root():
"""Serve the explorer UI"""
return HTML_TEMPLATE
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8017)