"""
Test file demonstrating the reentrancy vulnerability in claim() function
where events can be emitted multiple times due to improper CEI pattern.
"""
class MaliciousToken:
def __init__(self):
self.reentry_count = 0
self.max_reentries = 3
self.airdrop = None
def transfer(self, to: str, amount: int) -> bool:
"""Malicious transfer function that will reenter claim()"""
print(f"\nTransfer called with amount: {amount}")
print(f"Reentry count: {self.reentry_count}")
if self.reentry_count < self.max_reentries:
self.reentry_count += 1
print(f"Reentering claim() - attempt {self.reentry_count}")
self.airdrop.claimed_amount["0x1111111111111111111111111111111111111111"] = 0
self.airdrop.claim(
"0x1111111111111111111111111111111111111111",
1000,
["0x1234"]
)
return True
class MockVestedAirdrop:
def __init__(self, token):
self.token = token
token.airdrop = self
self.claimed_amount = {}
self.vesting_start_time = 100
self.events = []
def verify_proof(self, user, amount, proof) -> bool:
"""Mock proof verification - always returns true for demo"""
return True
def _calculate_vested_amount(self, total_amount: int) -> int:
"""Mock vesting calculation - returns 90% of total for demo"""
return (total_amount * 90) // 100
def claim(self, user: str, total_amount: int, proof: list) -> bool:
"""Vulnerable claim function with incorrect event emission order"""
print(f"\nClaim called for user {user}")
print(f"Current claimed amount: {self.claimed_amount.get(user, 0)}")
assert self.verify_proof(user, total_amount, proof), "Invalid proof"
assert True, "Claiming is not available yet"
claimable = 0
current_amount = self.claimed_amount.get(user, 0)
vested = self._calculate_vested_amount(total_amount)
if vested > current_amount:
claimable = vested - current_amount
assert claimable > 0, "Nothing to claim"
self.claimed_amount[user] = current_amount + claimable
print(f"Emitting Claimed event: user={user}, amount={claimable}")
self.events.append({
"event": "Claimed",
"user": user,
"amount": claimable,
"reentry_depth": self.token.reentry_count
})
success = self.token.transfer(user, claimable)
assert success, "Transfer failed"
return True
def test_reentrancy_event():
"""Test to demonstrate the reentrancy vulnerability with event emission"""
print("\n=== Testing Reentrancy with Event Emission ===")
token = MaliciousToken()
airdrop = MockVestedAirdrop(token)
user = "0x1111111111111111111111111111111111111111"
total_amount = 1000
proof = ["0x1234"]
try:
airdrop.claim(user, total_amount, proof)
except Exception as e:
print(f"Claim failed: {str(e)}")
print("\n=== Results ===")
print(f"Number of Claimed events emitted: {len(airdrop.events)}")
total_claimed = 0
print("\nEvent details:")
for i, event in enumerate(airdrop.events, 1):
print(f"Event {i}:")
print(f" User: {event['user']}")
print(f" Amount: {event['amount']}")
print(f" Reentry depth: {event['reentry_depth']}")
total_claimed += event['amount']
if len(airdrop.events) > 1:
print("\n❌ VULNERABILITY CONFIRMED")
print(f"Multiple Claimed events were emitted ({len(airdrop.events)} events)")
print(f"Total amount claimed through events: {total_claimed}")
print(f"Expected maximum claim: {total_amount}")
print("\nThis vulnerability allows:")
print("1. Multiple event emissions for the same claim")
print("2. Misleading off-chain systems tracking claim events")
print("3. Potential double-counting of claims in external systems")
else:
print("\n✅ No vulnerability detected")
if __name__ == "__main__":
test_reentrancy_event()