- Exclude already-swept payments from watcher queue to prevent endless processing - Mark swept payments as "swept" status to remove from future cycles - Fix cart deactivation by ensuring shop/user objects are attached to session - Add debug logging for cart deactivation to track behavior - Resolves issue where confirmed payments were processed every 20 seconds indefinitely 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
170 lines
5.6 KiB
Python
170 lines
5.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Debug script to test different Monero RPC authentication methods
|
|
and identify why the Python implementation differs from curl.
|
|
"""
|
|
import urllib.request
|
|
import urllib.error
|
|
import json
|
|
import base64
|
|
import sys
|
|
|
|
|
|
def test_no_auth(rpc_url):
|
|
"""Test without authentication"""
|
|
print("=== Testing No Authentication ===")
|
|
payload = {"jsonrpc": "2.0", "id": 1, "method": "get_height"}
|
|
data = json.dumps(payload).encode()
|
|
|
|
req = urllib.request.Request(
|
|
rpc_url, data=data, headers={"Content-Type": "application/json"}
|
|
)
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
|
result = json.loads(resp.read())
|
|
print("✓ No auth SUCCESS:", result)
|
|
return True
|
|
except urllib.error.HTTPError as e:
|
|
print(f"✗ No auth FAILED: HTTP {e.code} {e.reason}")
|
|
if hasattr(e, "read"):
|
|
try:
|
|
body = e.read().decode()
|
|
print(f"Response body: {body}")
|
|
except:
|
|
pass
|
|
return False
|
|
except Exception as e:
|
|
print(f"✗ No auth ERROR: {e}")
|
|
return False
|
|
|
|
|
|
def test_basic_auth(rpc_url, rpc_user, rpc_pass):
|
|
"""Test Basic authentication"""
|
|
print("=== Testing Basic Authentication ===")
|
|
payload = {"jsonrpc": "2.0", "id": 1, "method": "get_height"}
|
|
data = json.dumps(payload).encode()
|
|
|
|
auth = f"{rpc_user}:{rpc_pass}".encode()
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Authorization": "Basic " + base64.b64encode(auth).decode(),
|
|
}
|
|
|
|
req = urllib.request.Request(rpc_url, data=data, headers=headers)
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
|
result = json.loads(resp.read())
|
|
print("✓ Basic auth SUCCESS:", result)
|
|
return True
|
|
except urllib.error.HTTPError as e:
|
|
print(f"✗ Basic auth FAILED: HTTP {e.code} {e.reason}")
|
|
if hasattr(e, "read"):
|
|
try:
|
|
body = e.read().decode()
|
|
print(f"Response body: {body}")
|
|
except:
|
|
pass
|
|
return False
|
|
except Exception as e:
|
|
print(f"✗ Basic auth ERROR: {e}")
|
|
return False
|
|
|
|
|
|
def test_digest_auth_simple(rpc_url, rpc_user, rpc_pass):
|
|
"""Test Digest authentication - current implementation"""
|
|
print("=== Testing Digest Authentication (Current Implementation) ===")
|
|
payload = {"jsonrpc": "2.0", "id": 1, "method": "get_height"}
|
|
data = json.dumps(payload).encode()
|
|
|
|
# Current implementation from crypto_clients.py
|
|
password_mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
|
|
password_mgr.add_password(None, rpc_url, rpc_user, rpc_pass)
|
|
|
|
auth_handler = urllib.request.HTTPDigestAuthHandler(password_mgr)
|
|
opener = urllib.request.build_opener(auth_handler)
|
|
|
|
req = urllib.request.Request(
|
|
rpc_url, data=data, headers={"Content-Type": "application/json"}
|
|
)
|
|
|
|
try:
|
|
with opener.open(req, timeout=15) as resp:
|
|
result = json.loads(resp.read())
|
|
print("✓ Digest auth SUCCESS:", result)
|
|
return True
|
|
except urllib.error.HTTPError as e:
|
|
print(f"✗ Digest auth FAILED: HTTP {e.code} {e.reason}")
|
|
if hasattr(e, "read"):
|
|
try:
|
|
body = e.read().decode()
|
|
print(f"Response body: {body}")
|
|
except:
|
|
pass
|
|
return False
|
|
except Exception as e:
|
|
print(f"✗ Digest auth ERROR: {e}")
|
|
return False
|
|
|
|
|
|
def test_digest_auth_with_realm(rpc_url, rpc_user, rpc_pass):
|
|
"""Test Digest authentication with specific realm"""
|
|
print("=== Testing Digest Authentication (with realm) ===")
|
|
payload = {"jsonrpc": "2.0", "id": 1, "method": "get_height"}
|
|
data = json.dumps(payload).encode()
|
|
|
|
# Try with specific realm
|
|
password_mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
|
|
# Try different realm possibilities
|
|
for realm in [None, "monero-wallet-rpc", "RPC", ""]:
|
|
password_mgr.add_password(realm, rpc_url, rpc_user, rpc_pass)
|
|
|
|
auth_handler = urllib.request.HTTPDigestAuthHandler(password_mgr)
|
|
opener = urllib.request.build_opener(auth_handler)
|
|
|
|
req = urllib.request.Request(
|
|
rpc_url, data=data, headers={"Content-Type": "application/json"}
|
|
)
|
|
|
|
try:
|
|
with opener.open(req, timeout=15) as resp:
|
|
result = json.loads(resp.read())
|
|
print("✓ Digest auth with realm SUCCESS:", result)
|
|
return True
|
|
except urllib.error.HTTPError as e:
|
|
print(f"✗ Digest auth with realm FAILED: HTTP {e.code} {e.reason}")
|
|
if hasattr(e, "read"):
|
|
try:
|
|
body = e.read().decode()
|
|
print(f"Response body: {body}")
|
|
except:
|
|
pass
|
|
return False
|
|
except Exception as e:
|
|
print(f"✗ Digest auth with realm ERROR: {e}")
|
|
return False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Test configuration - these would be the production values
|
|
# (keeping them in the script for easy testing)
|
|
rpc_url = "http://127.0.0.1:18083/json_rpc"
|
|
rpc_user = "mps_xmr_user_5a4c3592d2b4b52d"
|
|
rpc_pass = "jLOnImOKWT1Vomeje3HvKBzsDjT6VPcxvf7FMtEpjvY4"
|
|
|
|
print("Monero RPC Authentication Debug")
|
|
print("=" * 50)
|
|
print(f"URL: {rpc_url}")
|
|
print(f"User: {rpc_user}")
|
|
print(f"Pass: {'*' * len(rpc_pass)}")
|
|
print()
|
|
|
|
# Test all methods
|
|
test_no_auth(rpc_url)
|
|
print()
|
|
test_basic_auth(rpc_url, rpc_user, rpc_pass)
|
|
print()
|
|
test_digest_auth_simple(rpc_url, rpc_user, rpc_pass)
|
|
print()
|
|
test_digest_auth_with_realm(rpc_url, rpc_user, rpc_pass)
|