Realtime Web3 security platform monitoring smart contracts for vulnerabilities & attacks
PulseProof - Intelligent Smart Contract Security Platform
PulseProof is not just another wrapper or monitoring bot , itβs a self-learning, AI-powered Web3 security platform designed to actively protect user funds. By continuously analyzing smart contracts, detecting vulnerabilities, and generating verified exploit proofs (PoCs) in real time, PulseProof ensures DeFi protocols and Web3 projects stay ahead of attackers. Multiple contracts can be monitored simultaneously, safeguarding funds across entire ecosystems. Key Differentiators
Not Another Bot: PulseProof actively prevents user fund loss, instead of just sending alerts. Agentic Intelligence: Specialized AI agents orchestrate detection, PoC generation, and verification. Verified PoC Pipeline: Exploit proofs are automatically generated, tested, and validated on mainnet forks. Scalable Protection: Monitor dozens to hundreds of smart contracts simultaneously. Future-Ready: Extendable with custom vulnerability detectors, ML models, and multi-chain support.
Core Purpose
Advanced Early Warning System: Continuously monitors smart contracts to detect suspicious activity and vulnerabilities. Actionable Insights: Provides verified PoCs and risk scores so developers can respond instantly. Scalable Monitoring: Track multiple contracts and protocols simultaneously, reducing attack surface across DeFi ecosystems.
Architecture Frontend Dashboard
Built with Fetch.ai uAgents framework:
Security Detection Monitors 10+ Vulnerability Types:
Risk Assessment Engine:
How PulseProof is Built - Technical Deep Dive
The heart of PulseProof uses Fetch.ai's uAgents framework to create a sophisticated multi-agent system:
# Real agent communication using ctx.send_and_receive
@orchestrator.on_rest_post("/analyze-events", EventAnalysisRequest, EventAnalysisResponse)
async def analyze_events_endpoint(ctx: Context, request: EventAnalysisRequest):
# Delegates to specialized agents using real communication
event_analysis_result = await delegate_to_event_analyzer(ctx, request, request_id)
risk_assessment_result = await delegate_to_risk_assessor(ctx, event_analysis_result, request_id)
EnhancedEventProcessor
for blockchain event analysisctx.send
and ctx.send_and_receive
patternsEnhancedRiskEngine
with real-time price dataevents_vec.push(CandidateEvent {
transaction_hash: tx_hash.clone(),
block_number,
log_index,
contract_address: contract_addr.clone(),
event_signature: TRANSFER_TOPIC.to_string(),
event_type: "Transfer".to_string(),
metadata: format!("{{\"topics\":{:?},\"data\":\"{}\"}}",topics_vec, data_hex),
});
Benefits:
// High-performance event extraction and processing
events_vec.push(CandidateEvent {
transaction_hash: tx_hash.clone(),
block_number,
log_index,
contract_address: contract_addr.clone(),
event_signature: TRANSFER_TOPIC.to_string(),
event_type: "Transfer".to_string(),
metadata: format!("{{\"topics\":{:?},\"data\":\"{}\"}}",topics_vec, data_hex),
});
Benefits:
// PoC storage and retrieval system
const storeProofOfConcept = async (vulnerabilityData) => {
const filecoinHash = await ipfs.add({
path: `poc-${vulnerabilityData.id}.sol`,
content: vulnerabilityData.exploitCode
});
return `https://gateway.pinata.cloud/ipfs/${filecoinHash}`;
};
Benefits:
# Custom communication protocol between agents
response, status = await ctx.send_and_receive(
event_analyzer_address,
analysis_request,
reply_to=orchestrator.address,
timeout_seconds=30
)
if status.delivered and status.acknowledged:
# Process successful agent response
return parse_agent_response(response)
Why it's notable: We implemented synchronous communication between asynchronous agents, allowing complex workflows while maintaining real-time responsiveness.
# Multi-dimensional risk assessment
def assess_comprehensive_risk(self, processed_event: ProcessedEvent) -> RiskAssessment:
risk_components = {
'financial_impact': self._analyze_financial_impact(event),
'behavioral_anomalies': self._detect_behavioral_anomalies(event),
'reputation_risk': self._assess_reputation_risk(event),
'historical_context': self._analyze_historical_context(event)
}
# Weighted scoring with confidence factors
overall_score = sum(
component['score'] * Config.RISK_WEIGHTS[category]
for category, component in risk_components.items()
)
Why it's hacky: We combine real-time blockchain data, external price APIs, and historical patterns into a single risk score, requiring complex data synchronization and caching strategies.
// Intentionally vulnerable contract for AI training
contract VulnerableYieldFarm {
// Multiple vulnerabilities embedded for testing
function withdraw(uint256 amount) external {
require(balances[msg.sender] >= amount, "Insufficient balance");
// Reentrancy vulnerability - external call before state update
(bool success, ) = msg.sender.call{value: amount}("");
require(success, "Transfer failed");
balances[msg.sender] -= amount; // State update after external call
}
}
Why it's notable: We created a comprehensive vulnerable contract suite that mimics real DeFi protocols, allowing our AI agents to train on realistic attack scenarios.
// Browser-based wallet integration for instant deposits/withdrawals
const channelManager = new ChannelManager({
provider: window.ethereum,
rpc: process.env.NEXT_PUBLIC_RPC_URL
});
const handleInstantDeposit = async (amount: string, exchange: string) => {
const channel = await channelManager.openChannel(exchange);
return await channel.deposit(amount);
};
Why it's hacky: We bridged state channels with our monitoring system, allowing users to interact with DeFi while simultaneously monitoring for vulnerabilities in real-time.
The Graph Substreams β Raw Events β Master Orchestrator
β
Event Analyzer (Pattern Detection) β Risk Assessor (Financial Impact)
β
Enhanced Risk Engine β Real-time Dashboard β Filecoin PoC Storage