deleted: debug_auth_headers.py
deleted: debug_digest_details.py deleted: debug_monero_auth.py renamed: CRYPTO_WATCHER_FLOW.md -> docs/CRYPTO_WATCHER_FLOW.md deleted: test_digest_auth_local.py deleted: test_dogecoin_client.py deleted: test_monero_auth.py deleted: test_monero_client.py deleted: test_requests_lib.py deleted: test_urllib_variants.py
This commit is contained in:
parent
e3b022e2c5
commit
c1a936539e
10 changed files with 0 additions and 883 deletions
|
|
@ -1,76 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Debug script to examine the exact WWW-Authenticate header
|
||||
that Monero wallet RPC sends and how urllib handles it.
|
||||
"""
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
import json
|
||||
|
||||
|
||||
def examine_auth_challenge():
|
||||
"""Look at the WWW-Authenticate header from the server"""
|
||||
print("=== Examining Authentication Challenge ===")
|
||||
|
||||
rpc_url = "http://127.0.0.1:18083/json_rpc"
|
||||
payload = {"jsonrpc": "2.0", "id": 1, "method": "get_height"}
|
||||
data = json.dumps(payload).encode()
|
||||
|
||||
# Make request without auth to get the challenge
|
||||
req = urllib.request.Request(
|
||||
rpc_url, data=data, headers={"Content-Type": "application/json"}
|
||||
)
|
||||
|
||||
try:
|
||||
with urllib.request.urlopen(req) as resp:
|
||||
print("Unexpected: No auth required")
|
||||
return None
|
||||
except urllib.error.HTTPError as e:
|
||||
if e.code == 401:
|
||||
www_auth = e.headers.get("WWW-Authenticate", "")
|
||||
print(f"WWW-Authenticate header: {www_auth!r}")
|
||||
|
||||
# Parse it
|
||||
if www_auth.startswith("Digest "):
|
||||
challenge = www_auth[7:].strip()
|
||||
print(f"Challenge details: {challenge}")
|
||||
|
||||
# Parse individual components
|
||||
import re
|
||||
|
||||
pattern = r'(\w+)=(?:"([^"]*)"|([^,\s]+))'
|
||||
matches = re.findall(pattern, challenge)
|
||||
|
||||
print("Parsed components:")
|
||||
for key, quoted_val, unquoted_val in matches:
|
||||
val = quoted_val if quoted_val else unquoted_val
|
||||
print(f" {key} = {val!r}")
|
||||
|
||||
return dict(
|
||||
(key, quoted_val if quoted_val else unquoted_val)
|
||||
for key, quoted_val, unquoted_val in matches
|
||||
)
|
||||
else:
|
||||
print(f"Not Digest auth: {www_auth}")
|
||||
return None
|
||||
else:
|
||||
print(f"Unexpected error: {e.code} {e.reason}")
|
||||
return None
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
challenge = examine_auth_challenge()
|
||||
|
||||
if challenge:
|
||||
print()
|
||||
print("=== Analysis ===")
|
||||
print(f"Realm: {challenge.get('realm', 'NOT SET')}")
|
||||
print(f"Nonce: {challenge.get('nonce', 'NOT SET')}")
|
||||
print(f"QOP: {challenge.get('qop', 'NOT SET')}")
|
||||
print(f"Algorithm: {challenge.get('algorithm', 'NOT SET (defaults to MD5)')}")
|
||||
|
||||
# Check for any unusual parameters
|
||||
standard_params = {"realm", "nonce", "qop", "algorithm", "opaque"}
|
||||
extra_params = set(challenge.keys()) - standard_params
|
||||
if extra_params:
|
||||
print(f"Extra parameters: {extra_params}")
|
||||
|
|
@ -1,138 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Debug script to examine the exact Digest authentication process
|
||||
and identify potential issues with the Python urllib implementation.
|
||||
"""
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
import http.client
|
||||
import json
|
||||
import hashlib
|
||||
import secrets
|
||||
|
||||
|
||||
def debug_digest_auth_process():
|
||||
"""
|
||||
Manually trace through the Digest authentication process to understand
|
||||
what might be different from curl's implementation.
|
||||
"""
|
||||
print("=== Digest Authentication Debug Analysis ===")
|
||||
print()
|
||||
|
||||
# Enable HTTP debugging to see what's being sent
|
||||
http.client.HTTPConnection.debuglevel = 1
|
||||
|
||||
rpc_url = "http://127.0.0.1:18083/json_rpc"
|
||||
rpc_user = "mps_xmr_user_5a4c3592d2b4b52d"
|
||||
rpc_pass = "jLOnImOKWT1Vomeje3HvKBzsDjT6VPcxvf7FMtEpjvY4"
|
||||
|
||||
payload = {"jsonrpc": "2.0", "id": 1, "method": "get_height"}
|
||||
data = json.dumps(payload).encode()
|
||||
|
||||
print("Step 1: Attempting initial request (should get 401 with WWW-Authenticate)")
|
||||
|
||||
# Make initial request without auth to get the challenge
|
||||
req = urllib.request.Request(
|
||||
rpc_url, data=data, headers={"Content-Type": "application/json"}
|
||||
)
|
||||
|
||||
try:
|
||||
with urllib.request.urlopen(req) as resp:
|
||||
print("Unexpected: Got response without auth required")
|
||||
return
|
||||
except urllib.error.HTTPError as e:
|
||||
if e.code == 401:
|
||||
print(f"✓ Got 401 as expected: {e.reason}")
|
||||
auth_header = e.headers.get("WWW-Authenticate")
|
||||
if auth_header:
|
||||
print(f"WWW-Authenticate header: {auth_header}")
|
||||
|
||||
# Parse the challenge
|
||||
if auth_header.startswith("Digest "):
|
||||
print("✓ Server supports Digest authentication")
|
||||
challenge_parts = auth_header[7:] # Remove 'Digest '
|
||||
print(f"Challenge details: {challenge_parts}")
|
||||
else:
|
||||
print(f"⚠ Server wants {auth_header.split()[0]} auth, not Digest")
|
||||
else:
|
||||
print("✗ No WWW-Authenticate header found")
|
||||
else:
|
||||
print(f"Unexpected HTTP error: {e.code} {e.reason}")
|
||||
|
||||
print("\nStep 2: Now trying with Python's built-in Digest handler")
|
||||
|
||||
# Reset debug level for cleaner output
|
||||
http.client.HTTPConnection.debuglevel = 0
|
||||
|
||||
# Try with the urllib digest handler
|
||||
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) as resp:
|
||||
result = json.loads(resp.read())
|
||||
print("✓ Digest auth SUCCESS:", result)
|
||||
except urllib.error.HTTPError as e:
|
||||
print(f"✗ Digest auth FAILED: {e.code} {e.reason}")
|
||||
# Try to get more details
|
||||
if hasattr(e, "read"):
|
||||
body = e.read().decode()
|
||||
print(f"Response body: {body}")
|
||||
|
||||
|
||||
def compare_with_requests_library():
|
||||
"""
|
||||
Compare urllib with the requests library to see if there's a difference.
|
||||
This helps identify if it's a urllib-specific issue.
|
||||
"""
|
||||
print("\n=== Comparison with requests library ===")
|
||||
|
||||
try:
|
||||
import requests
|
||||
from requests.auth import HTTPDigestAuth
|
||||
|
||||
rpc_url = "http://127.0.0.1:18083/json_rpc"
|
||||
rpc_user = "mps_xmr_user_5a4c3592d2b4b52d"
|
||||
rpc_pass = "jLOnImOKWT1Vomeje3HvKBzsDjT6VPcxvf7FMtEpjvY4"
|
||||
|
||||
payload = {"jsonrpc": "2.0", "id": 1, "method": "get_height"}
|
||||
|
||||
print("Testing with requests library...")
|
||||
|
||||
response = requests.post(
|
||||
rpc_url, json=payload, auth=HTTPDigestAuth(rpc_user, rpc_pass), timeout=15
|
||||
)
|
||||
|
||||
if response.status_code == 200:
|
||||
result = response.json()
|
||||
print("✓ requests library SUCCESS:", result)
|
||||
else:
|
||||
print(
|
||||
f"✗ requests library FAILED: {response.status_code} {response.reason}"
|
||||
)
|
||||
print(f"Response body: {response.text}")
|
||||
|
||||
except ImportError:
|
||||
print("requests library not available for comparison")
|
||||
except Exception as e:
|
||||
print(f"requests library error: {e}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
debug_digest_auth_process()
|
||||
compare_with_requests_library()
|
||||
|
||||
print("\n=== Summary ===")
|
||||
print("This debug script helps identify:")
|
||||
print("1. What WWW-Authenticate challenge the server sends")
|
||||
print("2. Whether urllib's Digest implementation works correctly")
|
||||
print("3. How different HTTP libraries handle the same credentials")
|
||||
print()
|
||||
print("Run this script on the production server where RPC auth is required.")
|
||||
|
|
@ -1,170 +0,0 @@
|
|||
#!/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)
|
||||
|
|
@ -1,115 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test Digest authentication locally using monero-wallet-remote-auth target.
|
||||
|
||||
This script tests our Digest auth implementation against a real Monero wallet RPC
|
||||
with authentication enabled, rather than the no-auth version.
|
||||
"""
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Add the current directory to path so we can import the module
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
from make_post_sell.lib.crypto_clients import MoneroClient
|
||||
|
||||
|
||||
def test_no_auth():
|
||||
"""Test that requests fail without authentication"""
|
||||
print("=== Testing No Authentication (should fail) ===")
|
||||
|
||||
client = MoneroClient("http://127.0.0.1:18083/json_rpc")
|
||||
|
||||
try:
|
||||
height = client.get_height()
|
||||
print(f"✗ UNEXPECTED SUCCESS: {height}")
|
||||
print("(This means RPC authentication is not enabled)")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"✓ EXPECTED FAILURE: {e}")
|
||||
return True
|
||||
|
||||
|
||||
def test_wrong_credentials():
|
||||
"""Test that requests fail with wrong credentials"""
|
||||
print("=== Testing Wrong Credentials (should fail) ===")
|
||||
|
||||
client = MoneroClient("http://127.0.0.1:18083/json_rpc", "wrong_user", "wrong_pass")
|
||||
|
||||
try:
|
||||
height = client.get_height()
|
||||
print(f"✗ UNEXPECTED SUCCESS: {height}")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"✓ EXPECTED FAILURE: {e}")
|
||||
return True
|
||||
|
||||
|
||||
def test_correct_credentials():
|
||||
"""Test that requests succeed with correct credentials"""
|
||||
print("=== Testing Correct Credentials (should work) ===")
|
||||
|
||||
client = MoneroClient("http://127.0.0.1:18083/json_rpc", "test_user", "test_pass")
|
||||
|
||||
try:
|
||||
height = client.get_height()
|
||||
print(f"✓ SUCCESS: Current height = {height}")
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"✗ FAILED: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def test_curl_comparison():
|
||||
"""Test the same credentials with curl to compare"""
|
||||
print("=== Testing with curl for comparison ===")
|
||||
|
||||
import subprocess
|
||||
|
||||
curl_cmd = [
|
||||
"curl",
|
||||
"-X",
|
||||
"POST",
|
||||
"http://127.0.0.1:18083/json_rpc",
|
||||
"-H",
|
||||
"Content-Type: application/json",
|
||||
"-d",
|
||||
'{"jsonrpc":"2.0","id":1,"method":"get_height"}',
|
||||
"--digest",
|
||||
"-u",
|
||||
"test_user:test_pass",
|
||||
"--silent",
|
||||
]
|
||||
|
||||
try:
|
||||
result = subprocess.run(curl_cmd, capture_output=True, text=True, timeout=10)
|
||||
if result.returncode == 0:
|
||||
print(f"✓ curl SUCCESS: {result.stdout.strip()}")
|
||||
else:
|
||||
print(f"✗ curl FAILED: {result.stderr.strip()}")
|
||||
except Exception as e:
|
||||
print(f"✗ curl ERROR: {e}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("Local Digest Authentication Test")
|
||||
print("=" * 50)
|
||||
print()
|
||||
print("IMPORTANT: Start the authenticated wallet RPC first:")
|
||||
print(" make monero-wallet-remote-auth")
|
||||
print()
|
||||
|
||||
# Run all tests
|
||||
test_no_auth()
|
||||
print()
|
||||
test_wrong_credentials()
|
||||
print()
|
||||
test_correct_credentials()
|
||||
print()
|
||||
test_curl_comparison()
|
||||
|
||||
print("\n" + "=" * 50)
|
||||
print("If authentication is working correctly:")
|
||||
print("- No auth and wrong credentials should fail")
|
||||
print("- Correct credentials should succeed")
|
||||
print("- curl and Python should behave the same")
|
||||
|
|
@ -1,67 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test the DogecoinClient with requests library.
|
||||
"""
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Add the current directory to path so we can import the module
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
from make_post_sell.lib.crypto_clients import DogecoinClient
|
||||
|
||||
|
||||
def test_dogecoin_mock():
|
||||
"""Test DogecoinClient with mock implementation (no real RPC needed)"""
|
||||
print("=== Testing DogecoinClient (Mock mode) ===")
|
||||
|
||||
# This uses the MockDogecoinClient which doesn't need a real server
|
||||
from make_post_sell.lib.crypto_clients import MockDogecoinClient
|
||||
|
||||
client = MockDogecoinClient()
|
||||
|
||||
try:
|
||||
# Test basic operations
|
||||
address = client.getnewaddress("test")
|
||||
print(f"✓ Generated address: {address}")
|
||||
|
||||
balance = client.getbalance()
|
||||
print(f"✓ Balance: {balance} DOGE")
|
||||
|
||||
height = client.getblockcount()
|
||||
print(f"✓ Block height: {height}")
|
||||
|
||||
print("✓ MockDogecoinClient working correctly")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"✗ MockDogecoinClient failed: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def test_dogecoin_real():
|
||||
"""Test DogecoinClient with real RPC (would need running dogecoind)"""
|
||||
print("\n=== Testing DogecoinClient (Real RPC - will likely fail) ===")
|
||||
|
||||
# This would need a real Dogecoin RPC running
|
||||
client = DogecoinClient("http://127.0.0.1:22555", "test_user", "test_pass")
|
||||
|
||||
try:
|
||||
info = client.getnetworkinfo()
|
||||
print(f"✓ Network info: {info}")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"✗ EXPECTED FAILURE (no Dogecoin RPC running): {e}")
|
||||
return False
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("Dogecoin Client Test")
|
||||
print("=" * 40)
|
||||
|
||||
test_dogecoin_mock()
|
||||
test_dogecoin_real()
|
||||
|
||||
print("\nNote: Real RPC test expected to fail unless dogecoind is running")
|
||||
print("The important thing is that the mock client works correctly.")
|
||||
|
|
@ -1,122 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test script to debug Monero RPC authentication issues.
|
||||
This helps isolate whether the problem is with credentials or authentication method.
|
||||
"""
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
import json
|
||||
import base64
|
||||
import os
|
||||
|
||||
|
||||
def test_basic_auth(rpc_url, rpc_user, rpc_pass):
|
||||
"""Test Basic authentication (old method)"""
|
||||
print("Testing Basic Authentication...")
|
||||
|
||||
payload = {"jsonrpc": "2.0", "id": 1, "method": "get_height"}
|
||||
data = json.dumps(payload).encode()
|
||||
|
||||
# Basic auth header
|
||||
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}")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"✗ Basic auth ERROR: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def test_digest_auth(rpc_url, rpc_user, rpc_pass):
|
||||
"""Test Digest authentication (new method)"""
|
||||
print("Testing Digest Authentication...")
|
||||
|
||||
payload = {"jsonrpc": "2.0", "id": 1, "method": "get_height"}
|
||||
data = json.dumps(payload).encode()
|
||||
|
||||
# Digest auth setup
|
||||
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}")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"✗ Digest auth ERROR: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def test_no_auth(rpc_url):
|
||||
"""Test no 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}")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"✗ No auth ERROR: {e}")
|
||||
return False
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Configuration from environment or defaults
|
||||
rpc_url = os.environ.get("MPS_MONERO_RPC_URL", "http://127.0.0.1:18083/json_rpc")
|
||||
rpc_user = os.environ.get("MPS_MONERO_RPC_USER", "")
|
||||
rpc_pass = os.environ.get("MPS_MONERO_RPC_PASS", "")
|
||||
|
||||
print("=== Monero RPC Authentication Test ===")
|
||||
print(f"URL: {rpc_url}")
|
||||
print(f"User: {rpc_user!r}")
|
||||
print(f"Pass: {'*' * len(rpc_pass) if rpc_pass else '(empty)'}")
|
||||
print()
|
||||
|
||||
if not rpc_user or not rpc_pass:
|
||||
print("⚠️ No credentials found in environment variables")
|
||||
print("Set MPS_MONERO_RPC_USER and MPS_MONERO_RPC_PASS")
|
||||
print()
|
||||
|
||||
# Test all three methods
|
||||
test_no_auth(rpc_url)
|
||||
print()
|
||||
|
||||
if rpc_user and rpc_pass:
|
||||
test_basic_auth(rpc_url, rpc_user, rpc_pass)
|
||||
print()
|
||||
test_digest_auth(rpc_url, rpc_user, rpc_pass)
|
||||
else:
|
||||
print("Skipping credential tests - no username/password provided")
|
||||
|
|
@ -1,78 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test the actual MoneroClient class to identify the authentication issue.
|
||||
"""
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Add the current directory to path so we can import the module
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
from make_post_sell.lib.crypto_clients import MoneroClient
|
||||
|
||||
|
||||
def test_monero_client():
|
||||
"""Test the MoneroClient with the production credentials"""
|
||||
print("=== Testing MoneroClient ===")
|
||||
|
||||
# Production credentials from the remote server
|
||||
rpc_url = "http://127.0.0.1:18083/json_rpc"
|
||||
rpc_user = "mps_xmr_user_5a4c3592d2b4b52d"
|
||||
rpc_pass = "jLOnImOKWT1Vomeje3HvKBzsDjT6VPcxvf7FMtEpjvY4"
|
||||
|
||||
print(f"URL: {rpc_url}")
|
||||
print(f"User: {rpc_user}")
|
||||
print(f"Pass: {'*' * len(rpc_pass)}")
|
||||
print()
|
||||
|
||||
client = MoneroClient(rpc_url, rpc_user, rpc_pass)
|
||||
|
||||
# Test basic connection
|
||||
try:
|
||||
print("Testing get_height()...")
|
||||
height = client.get_height()
|
||||
print(f"✓ SUCCESS: Current height = {height}")
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"✗ FAILED: {e}")
|
||||
print(f"Error type: {type(e).__name__}")
|
||||
|
||||
# Try to get more detailed error info
|
||||
if hasattr(e, "__cause__") and e.__cause__:
|
||||
print(f"Caused by: {e.__cause__}")
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def test_monero_client_localhost():
|
||||
"""Test with localhost connection - might not work but useful for debugging"""
|
||||
print("=== Testing MoneroClient with localhost (fallback) ===")
|
||||
|
||||
# This won't work since we're not running the RPC locally, but good for testing
|
||||
rpc_url = "http://127.0.0.1:18083/json_rpc"
|
||||
rpc_user = "test_user"
|
||||
rpc_pass = "test_pass"
|
||||
|
||||
client = MoneroClient(rpc_url, rpc_user, rpc_pass)
|
||||
|
||||
try:
|
||||
print("Testing get_height() on localhost...")
|
||||
height = client.get_height()
|
||||
print(f"✓ SUCCESS: Current height = {height}")
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"✗ EXPECTED FAILURE (no local RPC): {e}")
|
||||
return False
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("Monero Client Test")
|
||||
print("=" * 40)
|
||||
print()
|
||||
|
||||
# Test with production creds (won't work locally but shows the code path)
|
||||
test_monero_client()
|
||||
print()
|
||||
|
||||
# Test with localhost (also won't work but different error pattern)
|
||||
test_monero_client_localhost()
|
||||
|
|
@ -1,37 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test if the requests library works better than urllib for Digest auth.
|
||||
If requests works, we can see what it does differently.
|
||||
"""
|
||||
|
||||
try:
|
||||
import requests
|
||||
from requests.auth import HTTPDigestAuth
|
||||
|
||||
def test_with_requests():
|
||||
print("=== Testing with requests library ===")
|
||||
|
||||
url = "http://127.0.0.1:18083/json_rpc"
|
||||
payload = {"jsonrpc": "2.0", "id": 1, "method": "get_height"}
|
||||
auth = HTTPDigestAuth("test_user", "test_pass")
|
||||
|
||||
try:
|
||||
response = requests.post(url, json=payload, auth=auth, timeout=15)
|
||||
if response.status_code == 200:
|
||||
result = response.json()
|
||||
print(f"✓ requests SUCCESS: {result}")
|
||||
return True
|
||||
else:
|
||||
print(f"✗ requests FAILED: {response.status_code} {response.reason}")
|
||||
print(f"Response: {response.text}")
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"✗ requests ERROR: {e}")
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_with_requests()
|
||||
|
||||
except ImportError:
|
||||
print("requests library not installed. Install with: pip install requests")
|
||||
print("This would help us compare urllib vs requests for Digest auth.")
|
||||
|
|
@ -1,80 +0,0 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Test different urllib approaches to identify the exact issue.
|
||||
"""
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
import json
|
||||
import http.client
|
||||
|
||||
|
||||
def test_urllib_debug():
|
||||
"""Test urllib with full HTTP debugging enabled"""
|
||||
print("=== urllib with HTTP debugging ===")
|
||||
|
||||
# Enable debugging to see what's being sent
|
||||
http.client.HTTPConnection.debuglevel = 1
|
||||
|
||||
rpc_url = "http://127.0.0.1:18083/json_rpc"
|
||||
rpc_user = "test_user"
|
||||
rpc_pass = "test_pass"
|
||||
|
||||
payload = {"jsonrpc": "2.0", "id": 1, "method": "get_height"}
|
||||
data = json.dumps(payload).encode()
|
||||
|
||||
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) as resp:
|
||||
result = json.loads(resp.read())
|
||||
print(f"SUCCESS: {result}")
|
||||
except Exception as e:
|
||||
print(f"FAILED: {e}")
|
||||
finally:
|
||||
http.client.HTTPConnection.debuglevel = 0
|
||||
|
||||
|
||||
def test_urllib_different_url():
|
||||
"""Test if the issue is with the /json_rpc path"""
|
||||
print("\n=== Testing different URL formats ===")
|
||||
|
||||
base_url = "http://127.0.0.1:18083"
|
||||
paths = ["/json_rpc", "/json_rpc/", ""]
|
||||
|
||||
for path in paths:
|
||||
print(f"\nTesting with path: {path!r}")
|
||||
test_url = base_url + path if path else base_url
|
||||
|
||||
payload = {"jsonrpc": "2.0", "id": 1, "method": "get_height"}
|
||||
data = json.dumps(payload).encode()
|
||||
|
||||
password_mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
|
||||
password_mgr.add_password(None, test_url, "test_user", "test_pass")
|
||||
|
||||
auth_handler = urllib.request.HTTPDigestAuthHandler(password_mgr)
|
||||
opener = urllib.request.build_opener(auth_handler)
|
||||
|
||||
req = urllib.request.Request(
|
||||
test_url, data=data, headers={"Content-Type": "application/json"}
|
||||
)
|
||||
|
||||
try:
|
||||
with opener.open(req) as resp:
|
||||
result = json.loads(resp.read())
|
||||
print(f" ✓ SUCCESS: {result}")
|
||||
break
|
||||
except Exception as e:
|
||||
print(f" ✗ FAILED: {e}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_urllib_debug()
|
||||
test_urllib_different_url()
|
||||
Loading…
Add table
Add a link
Reference in a new issue