Apply black code formatting
Formatted Python code using black for: - crypto_clients.py - crypto_watcher.py - test_crypto_watcher.py Maintains consistent code style across the codebase.
This commit is contained in:
parent
0b5b0d2461
commit
4fbe9fedd4
13 changed files with 921 additions and 28 deletions
21
Makefile
21
Makefile
|
|
@ -46,6 +46,7 @@ help:
|
|||
@echo " make monero-node - Start local Monero node (150GB required)"
|
||||
@echo " make monero-wallet - Start wallet RPC with local node"
|
||||
@echo " make monero-wallet-remote - Start wallet RPC with remote node (dev)"
|
||||
@echo " make monero-wallet-remote-auth - Start wallet RPC with auth (testing)"
|
||||
@echo " make monero-full-stack - Show instructions for complete setup"
|
||||
@echo " make crypto-watcher - Start payment monitoring service"
|
||||
@echo " make crypto-watcher-once - Run payment check once (testing)"
|
||||
|
|
@ -257,6 +258,26 @@ monero-wallet-remote: venv config check-monero
|
|||
--trusted-daemon \
|
||||
--log-level=1
|
||||
|
||||
# Development mode with RPC authentication for testing Digest auth
|
||||
monero-wallet-remote-auth: venv config check-monero
|
||||
@echo "Starting wallet with REMOTE node and RPC AUTHENTICATION..."
|
||||
@echo "This enables RPC login for testing Digest authentication"
|
||||
@echo "Wallet file: $(DATA_DIR)/mps-wallet"
|
||||
@echo "RPC will be available at: http://127.0.0.1:18083"
|
||||
@echo "RPC User: test_user"
|
||||
@echo "RPC Pass: test_pass"
|
||||
@echo ""
|
||||
@echo "Using remote node: opennode.xmr-tw.org:18089"
|
||||
monero-wallet-rpc \
|
||||
--wallet-file=$(DATA_DIR)/mps-wallet \
|
||||
--password-file=$(DATA_DIR)/wallet-password.txt \
|
||||
--rpc-bind-ip=127.0.0.1 \
|
||||
--rpc-bind-port=18083 \
|
||||
--rpc-login=test_user:test_pass \
|
||||
--daemon-address=opennode.xmr-tw.org:18089 \
|
||||
--trusted-daemon \
|
||||
--log-level=1
|
||||
|
||||
|
||||
# Install Monero tools automatically
|
||||
install-monero:
|
||||
|
|
|
|||
70
debug_auth_headers.py
Normal file
70
debug_auth_headers.py
Normal file
|
|
@ -0,0 +1,70 @@
|
|||
#!/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}")
|
||||
134
debug_digest_details.py
Normal file
134
debug_digest_details.py
Normal file
|
|
@ -0,0 +1,134 @@
|
|||
#!/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.")
|
||||
162
debug_monero_auth.py
Normal file
162
debug_monero_auth.py
Normal file
|
|
@ -0,0 +1,162 @@
|
|||
#!/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)
|
||||
|
|
@ -7,6 +7,7 @@ import time
|
|||
try:
|
||||
import requests
|
||||
from requests.auth import HTTPDigestAuth
|
||||
|
||||
HAS_REQUESTS = True
|
||||
except ImportError:
|
||||
HAS_REQUESTS = False
|
||||
|
|
@ -30,7 +31,6 @@ class MoneroClient:
|
|||
self.rpc_pass = rpc_pass
|
||||
self.timeout = timeout
|
||||
|
||||
|
||||
def _call(self, method: str, params: Optional[Dict[str, Any]] = None) -> Any:
|
||||
payload = {
|
||||
"jsonrpc": "2.0",
|
||||
|
|
@ -42,24 +42,27 @@ class MoneroClient:
|
|||
|
||||
# Use requests library for all HTTP calls - much better Digest auth support
|
||||
if not HAS_REQUESTS:
|
||||
raise RuntimeError("requests library required for Monero RPC. Install with: pip install requests")
|
||||
|
||||
auth = HTTPDigestAuth(self.rpc_user, self.rpc_pass) if self.rpc_user and self.rpc_pass else None
|
||||
|
||||
raise RuntimeError(
|
||||
"requests library required for Monero RPC. Install with: pip install requests"
|
||||
)
|
||||
|
||||
auth = (
|
||||
HTTPDigestAuth(self.rpc_user, self.rpc_pass)
|
||||
if self.rpc_user and self.rpc_pass
|
||||
else None
|
||||
)
|
||||
|
||||
try:
|
||||
response = requests.post(
|
||||
self.rpc_url,
|
||||
json=payload,
|
||||
auth=auth,
|
||||
timeout=self.timeout
|
||||
self.rpc_url, json=payload, auth=auth, timeout=self.timeout
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
|
||||
obj = response.json()
|
||||
if "error" in obj and obj["error"]:
|
||||
raise RuntimeError(obj["error"]) # bubble up rpc error
|
||||
return obj.get("result")
|
||||
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise RuntimeError(f"Monero RPC connection error: {e}")
|
||||
|
||||
|
|
@ -177,24 +180,28 @@ class DogecoinClient:
|
|||
|
||||
# Use requests library for all HTTP calls - consistent with MoneroClient
|
||||
if not HAS_REQUESTS:
|
||||
raise RuntimeError("requests library required for Dogecoin RPC. Install with: pip install requests")
|
||||
|
||||
auth = (self.rpc_user, self.rpc_pass) if self.rpc_user and self.rpc_pass else None
|
||||
|
||||
raise RuntimeError(
|
||||
"requests library required for Dogecoin RPC. Install with: pip install requests"
|
||||
)
|
||||
|
||||
auth = (
|
||||
(self.rpc_user, self.rpc_pass) if self.rpc_user and self.rpc_pass else None
|
||||
)
|
||||
|
||||
try:
|
||||
response = requests.post(
|
||||
self.rpc_url,
|
||||
json=payload,
|
||||
auth=auth, # Basic auth for Dogecoin
|
||||
timeout=self.timeout
|
||||
timeout=self.timeout,
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
|
||||
obj = response.json()
|
||||
if "error" in obj and obj["error"]:
|
||||
raise RuntimeError(f"RPC error: {obj['error']}")
|
||||
return obj.get("result")
|
||||
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise RuntimeError(f"Dogecoin RPC connection error: {e}")
|
||||
|
||||
|
|
|
|||
|
|
@ -361,7 +361,9 @@ def finalize_invoice(env_request, crypto_payment: CryptoPayment):
|
|||
if current_active_cart and not current_active_cart.is_empty:
|
||||
# Create a new empty cart and make it active
|
||||
invoice.shop.create_new_cart_for_user(invoice.user)
|
||||
logger.info(f"Created new empty cart for user {invoice.user.id} after crypto payment confirmation")
|
||||
logger.info(
|
||||
f"Created new empty cart for user {invoice.user.id} after crypto payment confirmation"
|
||||
)
|
||||
|
||||
# Emails (configurable)
|
||||
email_enabled = True
|
||||
|
|
|
|||
|
|
@ -312,32 +312,34 @@ class CryptoWatcherIntegrationTests(unittest.TestCase):
|
|||
|
||||
@patch("make_post_sell.lib.crypto_watcher.send_purchase_email")
|
||||
@patch("make_post_sell.lib.crypto_watcher.send_sale_email")
|
||||
def test_finalize_invoice_deactivates_cart(self, mock_sale_email, mock_purchase_email):
|
||||
def test_finalize_invoice_deactivates_cart(
|
||||
self, mock_sale_email, mock_purchase_email
|
||||
):
|
||||
"""Test that finalizing invoice creates new empty cart for user."""
|
||||
with transaction.manager:
|
||||
dbsession = get_tm_session(self.session_factory, transaction.manager)
|
||||
|
||||
|
||||
# Get objects from database
|
||||
user = dbsession.query(User).filter_by(id=self.user_id).first()
|
||||
shop = dbsession.query(Shop).filter_by(id=self.shop_id).first()
|
||||
product = dbsession.query(Product).filter_by(id=self.product_id).first()
|
||||
|
||||
|
||||
# Create and set up an active cart with products
|
||||
old_cart = shop.create_new_cart_for_user(user)
|
||||
old_cart.add_product(product)
|
||||
dbsession.add(old_cart)
|
||||
dbsession.flush()
|
||||
|
||||
|
||||
# Verify cart has products
|
||||
self.assertFalse(old_cart.is_empty)
|
||||
old_cart_id = old_cart.id
|
||||
|
||||
|
||||
# Create invoice and payment
|
||||
invoice = Invoice(user)
|
||||
invoice.shop = shop
|
||||
invoice.new_line_item(product=product, quantity=1)
|
||||
dbsession.add(invoice)
|
||||
|
||||
|
||||
payment = CryptoPayment(
|
||||
invoice=invoice,
|
||||
address="test_cart_address",
|
||||
|
|
@ -350,16 +352,16 @@ class CryptoWatcherIntegrationTests(unittest.TestCase):
|
|||
)
|
||||
dbsession.add(payment)
|
||||
dbsession.flush()
|
||||
|
||||
|
||||
# Create mock request
|
||||
mock_request = MagicMock()
|
||||
mock_request.dbsession = dbsession
|
||||
mock_request.registry.settings = {"app.email.enabled": "true"}
|
||||
|
||||
|
||||
# Finalize the invoice
|
||||
finalize_invoice(mock_request, payment)
|
||||
dbsession.flush()
|
||||
|
||||
|
||||
# Check that a new cart was created
|
||||
new_cart = shop.get_active_cart_for_user(user)
|
||||
self.assertIsNotNone(new_cart)
|
||||
|
|
|
|||
112
test_digest_auth_local.py
Normal file
112
test_digest_auth_local.py
Normal file
|
|
@ -0,0 +1,112 @@
|
|||
#!/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")
|
||||
68
test_dogecoin_client.py
Normal file
68
test_dogecoin_client.py
Normal file
|
|
@ -0,0 +1,68 @@
|
|||
#!/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.")
|
||||
128
test_monero_auth.py
Normal file
128
test_monero_auth.py
Normal file
|
|
@ -0,0 +1,128 @@
|
|||
#!/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")
|
||||
75
test_monero_client.py
Normal file
75
test_monero_client.py
Normal file
|
|
@ -0,0 +1,75 @@
|
|||
#!/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()
|
||||
37
test_requests_lib.py
Normal file
37
test_requests_lib.py
Normal file
|
|
@ -0,0 +1,37 @@
|
|||
#!/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.")
|
||||
75
test_urllib_variants.py
Normal file
75
test_urllib_variants.py
Normal file
|
|
@ -0,0 +1,75 @@
|
|||
#!/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