1  """Smoke test the x402 service locally (without actual payment)."""
 2  
 3  from __future__ import annotations
 4  
 5  import json
 6  import subprocess
 7  import sys
 8  import time
 9  
10  import requests
11  
12  SAMPLE = r"""
13  // SPDX-License-Identifier: MIT
14  pragma solidity ^0.8.20;
15  
16  contract VulnerableVault {
17      mapping(address => uint256) public balances;
18      address public owner;
19  
20      constructor() {
21          owner = msg.sender;
22      }
23  
24      function deposit() external payable {
25          balances[msg.sender] += msg.value;
26      }
27  
28      // Classic reentrancy: external call before state update
29      function withdraw(uint256 amount) external {
30          require(balances[msg.sender] >= amount, "insufficient");
31          (bool ok,) = msg.sender.call{value: amount}("");
32          require(ok, "send failed");
33          balances[msg.sender] -= amount;
34      }
35  
36      // Access control via tx.origin (known bad practice)
37      function setOwner(address newOwner) external {
38          require(tx.origin == owner, "not owner");
39          owner = newOwner;
40      }
41  }
42  """
43  
44  
45  def start_server():
46      from pathlib import Path
47      proc = subprocess.Popen(
48          [sys.executable, "-m", "uvicorn", "app:app", "--port", "8401", "--log-level", "warning"],
49          cwd=str(Path(__file__).parent),
50          stdout=subprocess.PIPE, stderr=subprocess.PIPE,
51      )
52      # Wait for ready
53      for _ in range(60):
54          try:
55              r = requests.get("http://127.0.0.1:8401/health", timeout=1)
56              if r.ok:
57                  return proc
58          except Exception:
59              pass
60          time.sleep(0.5)
61      out, err = proc.communicate(timeout=5)
62      raise RuntimeError(f"server didn't start. stdout: {out[:500]}\nstderr: {err[:1500]}")
63  
64  
65  def main():
66      proc = start_server()
67      try:
68          # 1. /health
69          r = requests.get("http://127.0.0.1:8401/health", timeout=5)
70          print(f"GET /health -> {r.status_code}: {r.text}")
71  
72          # 2. /info
73          r = requests.get("http://127.0.0.1:8401/info", timeout=10)
74          print(f"GET /info -> {r.status_code}:")
75          print(json.dumps(r.json(), indent=2))
76  
77          # 3. POST /analyze (without payment — should return 402 or similar if x402 wired)
78          r = requests.post(
79              "http://127.0.0.1:8401/analyze",
80              json={"source": SAMPLE, "name": "VulnerableVault"},
81              timeout=240,
82          )
83          print(f"POST /analyze (no payment) -> {r.status_code}:")
84          print(r.text[:6000])
85      finally:
86          proc.terminate()
87          try:
88              proc.wait(timeout=10)
89          except subprocess.TimeoutExpired:
90              proc.kill()
91  
92  
93  if __name__ == "__main__":
94      main()