diff --git a/conftest.py b/conftest.py index 616d3a9..68ef821 100644 --- a/conftest.py +++ b/conftest.py @@ -1,6 +1,5 @@ import pytest -from monero import MoneroError def pytest_runtest_call(item: pytest.Item): # get not_supported marked @@ -17,7 +16,7 @@ def pytest_runtest_call(item: pytest.Item): try: # run test item.runtest() - except MoneroError as e: + except RuntimeError as e: e_str = str(e).lower() if "not supported" in e_str or "does not support" in e_str: # Ok diff --git a/src/cpp/py_monero.cpp b/src/cpp/py_monero.cpp index 6a979e9..0c6afe4 100644 --- a/src/cpp/py_monero.cpp +++ b/src/cpp/py_monero.cpp @@ -152,33 +152,24 @@ PYBIND11_MODULE(monero, m) { py::implicitly_convertible>>(); // monero_error - py::register_exception(m, "MoneroError"); + py::exception pyMoneroError(m, "MoneroError"); - py::exec(R"pybind( - class MoneroRpcError(RuntimeError): - def __init__(self, code: int, aMessage: str): + // python subclass + py::exec(R"( + class MoneroRpcError(MoneroError): + def __init__(self, message, code=-1): + super().__init__(message) self.code = code - self.message = aMessage - super().__init__(aMessage) - def get_code(self) -> int: - return self.code - def get_message(self) -> str: - return self.message - )pybind", - m.attr("__dict__"), m.attr("__dict__")); + )", m.attr("__dict__")); py::register_exception_translator([](std::exception_ptr p) { - const auto setPyException = [](const char* pyTypeName, const auto& exc) { - const py::object pyClass = py::module_::import("monero").attr(pyTypeName); - const py::object pyInstance = pyClass(exc.code, exc.what()); - PyErr_SetObject(pyClass.ptr(), pyInstance.ptr()); - }; - try { if (p) std::rethrow_exception(p); } - catch (const PyMoneroRpcError& exc) { - setPyException("MoneroRpcError", exc); + catch (const PyMoneroRpcError& e) { + py::object cls = py::module_::import("monero").attr("MoneroRpcError"); + py::object exc = cls(e.what(), e.code); + PyErr_SetObject(cls.ptr(), exc.ptr()); } }); @@ -1892,13 +1883,13 @@ PYBIND11_MODULE(monero, m) { }, py::arg("tx_hash"), py::arg("tx_key"), py::arg("address")) .def("get_tx_proof", [](PyMoneroWallet& self, const std::string& tx_hash, const std::string& address, const std::string& message) { MONERO_CATCH_AND_RETHROW(self.get_tx_proof(tx_hash, address, message)); - }, py::arg("tx_hash"), py::arg("address"), py::arg("message")) + }, py::arg("tx_hash"), py::arg("address"), py::arg("message") = "") .def("check_tx_proof", [](PyMoneroWallet& self, const std::string& tx_hash, const std::string& address, const std::string& message, const std::string& signature) { MONERO_CATCH_AND_RETHROW(self.check_tx_proof(tx_hash, address, message, signature)); }, py::arg("tx_hash"), py::arg("address"), py::arg("message"), py::arg("signature")) .def("get_spend_proof", [](PyMoneroWallet& self, const std::string& tx_hash, const std::string& message) { MONERO_CATCH_AND_RETHROW(self.get_spend_proof(tx_hash, message)); - }, py::arg("tx_hash"), py::arg("message")) + }, py::arg("tx_hash"), py::arg("message") = "") .def("check_spend_proof", [](PyMoneroWallet& self, const std::string& tx_hash, const std::string& message, const std::string& signature) { MONERO_CATCH_AND_RETHROW(self.check_spend_proof(tx_hash, message, signature)); }, py::arg("tx_hash"), py::arg("message"), py::arg("signature")) diff --git a/src/cpp/wallet/py_monero_wallet_model.cpp b/src/cpp/wallet/py_monero_wallet_model.cpp index 8e80539..6065e3d 100644 --- a/src/cpp/wallet/py_monero_wallet_model.cpp +++ b/src/cpp/wallet/py_monero_wallet_model.cpp @@ -1766,6 +1766,12 @@ void PyMoneroCheckReserve::from_property_tree(const boost::property_tree::ptree& else if (key == std::string("total")) check->m_total_amount = it->second.get_value(); else if (key == std::string("spent")) check->m_unconfirmed_spent_amount = it->second.get_value(); } + + if (!bool_equals_2(true, check->m_is_good)) { + // normalize invalid check reserve + check->m_total_amount = boost::none; + check->m_unconfirmed_spent_amount = boost::none; + } } void PyMoneroCheckTxProof::from_property_tree(const boost::property_tree::ptree& node, const std::shared_ptr& check) { diff --git a/src/cpp/wallet/py_monero_wallet_rpc.cpp b/src/cpp/wallet/py_monero_wallet_rpc.cpp index a0e934f..1a84a25 100644 --- a/src/cpp/wallet/py_monero_wallet_rpc.cpp +++ b/src/cpp/wallet/py_monero_wallet_rpc.cpp @@ -1238,23 +1238,32 @@ monero_message_signature_result PyMoneroWalletRpc::verify_message(const std::str } std::string PyMoneroWalletRpc::get_tx_key(const std::string& tx_hash) const { - auto params = std::make_shared(tx_hash); - PyMoneroJsonRequest request("get_tx_key", params); - auto response = m_rpc->send_json_request(request); - if (response->m_result == boost::none) throw std::runtime_error("Invalid Monero JSONRPC response"); - auto node = response->m_result.get(); - for (auto it = node.begin(); it != node.end(); ++it) { - std::string key = it->first; + try { + auto params = std::make_shared(tx_hash); + PyMoneroJsonRequest request("get_tx_key", params); + auto response = m_rpc->send_json_request(request); + if (response->m_result == boost::none) throw std::runtime_error("Invalid Monero JSONRPC response"); + auto node = response->m_result.get(); + for (auto it = node.begin(); it != node.end(); ++it) { + std::string key = it->first; - if (key == std::string("tx_key")) { - return it->second.data(); + if (key == std::string("tx_key")) { + return it->second.data(); + } } - } - throw std::runtime_error("Could not get tx key"); + throw std::runtime_error("Could not get tx key"); + } catch (const PyMoneroRpcError& ex) { + if (ex.code == -8 && ex.what() == std::string("TX ID has invalid format")) { + // normalize error message + throw PyMoneroRpcError(-8, "TX hash has invalid format"); + } + throw; + } } std::shared_ptr PyMoneroWalletRpc::check_tx_key(const std::string& tx_hash, const std::string& tx_key, const std::string& address) const { + try { auto params = std::make_shared(tx_hash, tx_key, address); PyMoneroJsonRequest request("check_tx_key", params); auto response = m_rpc->send_json_request(request); @@ -1263,47 +1272,90 @@ std::shared_ptr PyMoneroWalletRpc::check_tx_key(const std::stri auto check = std::make_shared(); PyMoneroCheckTxProof::from_property_tree(node, check); return check; + } catch (const PyMoneroRpcError& ex) { + if (ex.code == -8 && ex.what() == std::string("TX ID has invalid format")) { + // normalize error message + throw PyMoneroRpcError(-8, "TX hash has invalid format"); + } + throw; + } } std::string PyMoneroWalletRpc::get_tx_proof(const std::string& tx_hash, const std::string& address, const std::string& message) const { - auto params = std::make_shared(tx_hash, message); - params->m_address = address; - PyMoneroJsonRequest request("get_tx_proof", params); - auto response = m_rpc->send_json_request(request); - if (response->m_result == boost::none) throw std::runtime_error("Invalid Monero JSONRPC response"); - auto node = response->m_result.get(); - return PyMoneroReserveProofSignature::from_property_tree(node); + try { + auto params = std::make_shared(tx_hash, message); + params->m_address = address; + PyMoneroJsonRequest request("get_tx_proof", params); + auto response = m_rpc->send_json_request(request); + if (response->m_result == boost::none) throw std::runtime_error("Invalid Monero JSONRPC response"); + auto node = response->m_result.get(); + return PyMoneroReserveProofSignature::from_property_tree(node); + } catch (const PyMoneroRpcError& ex) { + if (ex.code == -8 && ex.what() == std::string("TX ID has invalid format")) { + // normalize error message + throw PyMoneroRpcError(-8, "TX hash has invalid format"); + } + throw; + } } std::shared_ptr PyMoneroWalletRpc::check_tx_proof(const std::string& tx_hash, const std::string& address, const std::string& message, const std::string& signature) const { - auto params = std::make_shared(tx_hash, address, message, signature); - PyMoneroJsonRequest request("check_tx_proof", params); - auto response = m_rpc->send_json_request(request); - if (response->m_result == boost::none) throw std::runtime_error("Invalid Monero JSONRPC response"); - auto node = response->m_result.get(); - auto check = std::make_shared(); - PyMoneroCheckTxProof::from_property_tree(node, check); - return check; + try { + auto params = std::make_shared(tx_hash, address, message, signature); + PyMoneroJsonRequest request("check_tx_proof", params); + auto response = m_rpc->send_json_request(request); + if (response->m_result == boost::none) throw std::runtime_error("Invalid Monero JSONRPC response"); + auto node = response->m_result.get(); + auto check = std::make_shared(); + PyMoneroCheckTxProof::from_property_tree(node, check); + return check; + } catch (const PyMoneroRpcError& ex) { + if (ex.code == -1 && ex.what() == std::string("basic_string")) { + throw PyMoneroRpcError(-1, "Must provide signature to check tx proof"); + } + if (ex.code == -8 && ex.what() == std::string("TX ID has invalid format")) { + // normalize error message + throw PyMoneroRpcError(-8, "TX hash has invalid format"); + } + throw; + } } std::string PyMoneroWalletRpc::get_spend_proof(const std::string& tx_hash, const std::string& message) const { - auto params = std::make_shared(tx_hash, message); - PyMoneroJsonRequest request("get_spend_proof", params); - auto response = m_rpc->send_json_request(request); - if (response->m_result == boost::none) throw std::runtime_error("Invalid Monero JSONRPC response"); - auto node = response->m_result.get(); - return PyMoneroReserveProofSignature::from_property_tree(node); + try { + auto params = std::make_shared(tx_hash, message); + PyMoneroJsonRequest request("get_spend_proof", params); + auto response = m_rpc->send_json_request(request); + if (response->m_result == boost::none) throw std::runtime_error("Invalid Monero JSONRPC response"); + auto node = response->m_result.get(); + return PyMoneroReserveProofSignature::from_property_tree(node); + } catch (const PyMoneroRpcError& ex) { + if (ex.code == -8 && ex.what() == std::string("TX ID has invalid format")) { + // normalize error message + throw PyMoneroRpcError(-8, "TX hash has invalid format"); + } + throw; + } } bool PyMoneroWalletRpc::check_spend_proof(const std::string& tx_hash, const std::string& message, const std::string& signature) const { - auto params = std::make_shared(tx_hash, message, signature); - PyMoneroJsonRequest request("check_spend_proof", params); - auto response = m_rpc->send_json_request(request); - if (response->m_result == boost::none) throw std::runtime_error("Invalid Monero JSONRPC response"); - auto node = response->m_result.get(); - auto proof = std::make_shared(); - PyMoneroCheckReserve::from_property_tree(node, proof); - return proof->m_is_good; + try { + auto params = std::make_shared(tx_hash, message); + params->m_signature = signature; + PyMoneroJsonRequest request("check_spend_proof", params); + auto response = m_rpc->send_json_request(request); + if (response->m_result == boost::none) throw std::runtime_error("Invalid Monero JSONRPC response"); + auto node = response->m_result.get(); + auto proof = std::make_shared(); + PyMoneroCheckReserve::from_property_tree(node, proof); + return proof->m_is_good; + } catch (const PyMoneroRpcError& ex) { + if (ex.code == -8 && ex.what() == std::string("TX ID has invalid format")) { + // normalize error message + throw PyMoneroRpcError(-8, "TX hash has invalid format"); + } + throw; + } } std::string PyMoneroWalletRpc::get_reserve_proof_wallet(const std::string& message) const { diff --git a/src/python/monero_rpc_error.pyi b/src/python/monero_rpc_error.pyi index 4ec4a3a..e23c6e6 100644 --- a/src/python/monero_rpc_error.pyi +++ b/src/python/monero_rpc_error.pyi @@ -2,19 +2,16 @@ class MoneroRpcError(RuntimeError): """ Exception when interacting with the Monero daemon or wallet RPC API. """ - def __init__(self, code: int, aMessage: str) -> None: - ... - def get_code(self) -> int: - """ - JSON-RPC error code. - :return int: Error code. - """ - ... - def get_message(self) -> str: + code: int + """JSON-RPC error code""" + + def __init__(self, message: str, code: int = -1) -> None: """ - JSON-RPC error message. + Initialize a new monero rpc error - :return str: Error message. + :param str message: rpc error message + :param int code: rpc error code """ ... + diff --git a/src/python/monero_ssl_options.pyi b/src/python/monero_ssl_options.pyi index feeca09..98e0477 100644 --- a/src/python/monero_ssl_options.pyi +++ b/src/python/monero_ssl_options.pyi @@ -1,9 +1,9 @@ class MoneroSslOptions: - ssl_private_key_path: str + ssl_private_key_path: str | None """Path to private ssl key""" - ssl_certificate_path: str + ssl_certificate_path: str | None """Path to private ssl certificate""" - ssl_ca_file: str + ssl_ca_file: str | None """Path to ssl CA file""" ssl_allowed_fingerprints: list[str] """Allowed ssl fingerprints""" diff --git a/src/python/monero_wallet.pyi b/src/python/monero_wallet.pyi index 0adea69..43e75de 100644 --- a/src/python/monero_wallet.pyi +++ b/src/python/monero_wallet.pyi @@ -546,7 +546,7 @@ class MoneroWallet: :return str: the language of the wallet's mnemonic phrase or seed. """ ... - def get_spend_proof(self, tx_hash: str, message: str) -> str: + def get_spend_proof(self, tx_hash: str, message: str = "") -> str: """ Generate a signature to prove a spend. Unlike proving a transaction, it does not require the destination public address. @@ -625,7 +625,7 @@ class MoneroWallet: :returns list[str]: notes for the transactions """ ... - def get_tx_proof(self, tx_hash: str, address: str, message: str) -> str: + def get_tx_proof(self, tx_hash: str, address: str, message: str = "") -> str: """ Get a transaction signature to prove it. diff --git a/tests/test_monero_common.py b/tests/test_monero_common.py new file mode 100644 index 0000000..fd4eacd --- /dev/null +++ b/tests/test_monero_common.py @@ -0,0 +1,34 @@ +import pytest +import logging + +from monero import ( + MoneroError, MoneroRpcError +) + +logger: logging.Logger = logging.getLogger("TestMoneroCommon") + + +@pytest.mark.unit +class TestMoneroCommon: + """Monero common unit tests""" + + @pytest.fixture(autouse=True) + def setup_and_teardown(self, request: pytest.FixtureRequest): + logger.info(f"Before {request.node.name}") # type: ignore + yield + logger.info(f"After {request.node.name}") # type: ignore + + # test monero error inheritance + def test_monero_error(self) -> None: + monero_err: MoneroError = MoneroError("Test monero error") + monero_rpc_err: MoneroRpcError = MoneroRpcError("Test monero rpc error") + + # test monero error + assert isinstance(monero_err, Exception) + assert str(monero_err) == "Test monero error" + + # test monero rpc error + assert isinstance(monero_rpc_err, Exception) + assert isinstance(monero_rpc_err, MoneroError) + assert str(monero_rpc_err) == "Test monero rpc error" + assert monero_rpc_err.code == -1 diff --git a/tests/test_monero_wallet_common.py b/tests/test_monero_wallet_common.py index 7b95a85..686e9da 100644 --- a/tests/test_monero_wallet_common.py +++ b/tests/test_monero_wallet_common.py @@ -16,7 +16,7 @@ MoneroOutputQuery, MoneroTransfer, MoneroIncomingTransfer, MoneroOutgoingTransfer, MoneroTxWallet, MoneroOutputWallet, MoneroTx, MoneroAccount, MoneroSubaddress, MoneroMessageSignatureType, MoneroTxPriority, MoneroFeeEstimate, - MoneroIntegratedAddress + MoneroIntegratedAddress, MoneroCheckTx, MoneroCheckReserve ) from utils import ( TestUtils, WalletEqualityUtils, @@ -837,7 +837,7 @@ def test_get_address_indices(self, wallet: MoneroWallet) -> None: wallet.get_address_index("this is definitely not an address") raise Exception("Should have thrown exception") except Exception as e: - AssertUtils.assert_equals("Invalid address", str(e)) + TxUtils.test_invalid_address_error(e) # Can decode an integrated address @pytest.mark.skipif(TestUtils.TEST_NON_RELAYS is False, reason="TEST_NON_RELAYS disabled") @@ -851,7 +851,7 @@ def test_decode_integrated_address(self, wallet: MoneroWallet) -> None: wallet.decode_integrated_address("bad address") raise Exception("Should have failed decoding bad address") except Exception as e: - AssertUtils.assert_equals("Invalid address", str(e)) + TxUtils.test_invalid_address_error(e) # Can sync (without progress) # TODO test syncing from start height @@ -2248,6 +2248,382 @@ def test_accounting(self, wallet: MoneroWallet) -> None: if subaddress_sum != subaddress.balance: assert has_unconfirmed_tx, "Subaddress balance must equal sum of its unspent outputs if no unconfirmed txs" + # Can check a transfer using the transaction's secret key and the destination + @pytest.mark.skipif(TestUtils.TEST_NON_RELAYS is False, reason="TEST_NON_RELAYS disabled") + def test_check_tx_key(self, wallet: MoneroWallet) -> None: + # get random txs that are confirmed and have outgoing destinations + txs: list[MoneroTxWallet] = [] + try: + query: MoneroTxQuery = MoneroTxQuery() + query.is_confirmed = True + query.transfer_query = MoneroTransferQuery() + query.transfer_query.has_destinations = True + txs = TxUtils.get_random_transactions(wallet, query, 1, WalletUtils.MAX_TX_PROOFS) + except AssertionError as e: + if "found with" in str(e): + raise Exception("No txs with outgoing destinations found; run send tests") + raise + + # test good checks + assert len(txs) > 0, "No transactions found with outgoing destinations" + for tx in txs: + assert tx.hash is not None + key: str = wallet.get_tx_key(tx.hash) + assert tx.outgoing_transfer is not None + assert len(tx.outgoing_transfer.destinations) > 0 + for destination in tx.outgoing_transfer.destinations: + assert destination.address is not None + check: MoneroCheckTx = wallet.check_tx_key(tx.hash, key, destination.address) + assert destination.amount is not None + if destination.amount > 0: + # TODO monero-wallet-rpc: indicates amount received amount is 0 despite transaction with transfer to this address + # TODO monero-wallet-rpc: returns 0-4 errors, not consistent + assert check.received_amount is not None + if check.received_amount == 0: + msg: str = f"WARNING: key proof indicates no funds received despite transfer (txid={tx.hash}, key={key}, address={destination.address} amount={destination.amount})" + logger.warning(msg) + else: + assert check.received_amount == 0 + TxUtils.test_check_tx(tx, check) + + # test get tx key with invalid hash + try: + wallet.get_tx_key("invalid_tx_id") + raise Exception("Should throw exception for invalid key") + except Exception as e: + TxUtils.test_invalid_tx_hash_error(e) + + # test check with invalid tx hash + tx: MoneroTxWallet = txs[0] + assert tx.hash is not None + key: str = wallet.get_tx_key(tx.hash) + assert tx.outgoing_transfer is not None + destination: MoneroDestination = tx.outgoing_transfer.destinations[0] + assert destination.address is not None + try: + wallet.check_tx_key("invalid_tx_id", key, destination.address) + raise Exception("Should have thrown exception") + except Exception as e: + TxUtils.test_invalid_tx_hash_error(e) + + # test check with invalid key + try: + wallet.check_tx_key(tx.hash, "invalid_tx_key", destination.address) + raise Exception("Should have thrown exception") + except Exception as e: + TxUtils.test_invalid_tx_key_error(e) + + # test check with invalid address + try: + wallet.check_tx_key(tx.hash, key, "invalid_tx_address") + raise Exception("Should have thrown exception") + except Exception as e: + TxUtils.test_invalid_address_error(e) + + # test check with different address + different_address: Optional[str] = None + for a_tx in wallet.get_txs(): + if a_tx.outgoing_transfer is None or len(a_tx.outgoing_transfer.destinations) == 0: + continue + for a_destination in a_tx.outgoing_transfer.destinations: + assert a_destination.address is not None + if a_destination.address != destination.address: + different_address = a_destination.address + break + + assert different_address is not None, "Could not get a different outgoing address to test; run send tests" + check: MoneroCheckTx = wallet.check_tx_key(tx.hash, key, different_address) + assert check.is_good is True + assert check.received_amount is not None + assert check.received_amount >= 0 + TxUtils.test_check_tx(tx, check) + + # Can prove a transaction by getting its signature + @pytest.mark.skipif(TestUtils.TEST_NON_RELAYS is False, reason="TEST_NON_RELAYS disabled") + def test_check_tx_proof(self, wallet: MoneroWallet) -> None: + # get random txs with outgoing destinations + txs: list[MoneroTxWallet] = [] + try: + query: MoneroTxQuery = MoneroTxQuery() + query.transfer_query = MoneroTransferQuery() + query.transfer_query.has_destinations = True + txs = TxUtils.get_random_transactions(wallet, query, 2, WalletUtils.MAX_TX_PROOFS) + except Exception as e: + if "found with" in str(e): + raise Exception("No txs with outgoing destinations found; run send tests") + raise + + # test good checks with messages + for tx in txs: + assert tx.hash is not None + assert tx.outgoing_transfer is not None + for destination in tx.outgoing_transfer.destinations: + assert destination.address is not None + signature: str = wallet.get_tx_proof(tx.hash, destination.address, "This transaction definitely happened.") + check: MoneroCheckTx = wallet.check_tx_proof(tx.hash, destination.address, "This transaction definitely happened.", signature) + TxUtils.test_check_tx(tx, check) + + # test good check without message + tx: MoneroTx = txs[0] + assert tx.hash is not None + assert tx.outgoing_transfer is not None + assert len(tx.outgoing_transfer.destinations) > 0 + destination: MoneroDestination = tx.outgoing_transfer.destinations[0] + assert destination.address is not None + signature: str = wallet.get_tx_proof(tx.hash, destination.address) + check: MoneroCheckTx = wallet.check_tx_proof(tx.hash, destination.address, '', signature) + TxUtils.test_check_tx(tx, check) + + # test get proof with invalid hash + try: + wallet.get_tx_proof("invalid_tx_id", destination.address) + raise Exception("Should throw exception for invalid key") + except Exception as e: + TxUtils.test_invalid_tx_hash_error(e) + + # test check tx proof with invalid tx hash + try: + wallet.check_tx_proof("invalid_tx_id", destination.address, '', signature) + raise Exception("Should have thrown exception") + except Exception as e: + TxUtils.test_invalid_tx_hash_error(e) + + # test check with invalid address + try: + wallet.check_tx_proof(tx.hash, "invalid_tx_address", '', signature) + raise Exception("Should have throw exception") + except Exception as e: + TxUtils.test_invalid_address_error(e) + + # test check with wrong message + signature = wallet.get_tx_proof(tx.hash, destination.address, "This is the right message") + check = wallet.check_tx_proof(tx.hash, destination.address, "This is the wrong message", signature) + assert check.is_good is False + TxUtils.test_check_tx(tx, check) + + # test check with wrong signature + other_tx: MoneroTxWallet = txs[1] + assert other_tx.hash is not None + assert other_tx.outgoing_transfer is not None + assert len(other_tx.outgoing_transfer.destinations) > 0 + address: Optional[str] = other_tx.outgoing_transfer.destinations[0].address + assert address is not None + wrong_signature: str = wallet.get_tx_proof(other_tx.hash, address, "This is the right message") + try: + check = wallet.check_tx_proof(tx.hash, destination.address, "This is the right message", wrong_signature) + assert check.is_good is False + except Exception as e: + TxUtils.test_invalid_signature_error(e) + + # test check with empty signature + try: + check = wallet.check_tx_proof(tx.hash, destination.address, "This is the right message", "") + assert check.is_good is False + except Exception as e: + assert str(e) == "Must provide signature to check tx proof" + + # Can prove a spend using a generated signature and no destination public address + @pytest.mark.skipif(TestUtils.TEST_NON_RELAYS is False, reason="TEST_NON_RELAYS disablde") + def test_check_spend_proof(self, wallet: MoneroWallet) -> None: + # get random confirmed outgoing txs + query: MoneroTxQuery = MoneroTxQuery() + query.is_incoming = False + query.in_tx_pool = False + query.is_failed = False + txs: list[MoneroTxWallet] = TxUtils.get_random_transactions(wallet, query, 2, WalletUtils.MAX_TX_PROOFS) + for tx in txs: + assert tx.is_confirmed is True + assert len(tx.incoming_transfers) == 0 + assert tx.outgoing_transfer is not None + + # test good checks with messages + for tx in txs: + assert tx.hash is not None + logger.debug(f"Getting tx hash {tx.hash} proof") + signature: str = wallet.get_spend_proof(tx.hash, "I am a message.") + logger.debug(f"Checking tx hash {tx.hash} proof") + result: bool = wallet.check_spend_proof(tx.hash, "I am a message.", signature) + assert result is True + + # test good check without message + tx: MoneroTxWallet = txs[0] + assert tx.hash is not None + signature: str = wallet.get_spend_proof(tx.hash) + result: bool = wallet.check_spend_proof(tx.hash, '', signature) + assert result is True + + # test get proof with invalid hash + try: + wallet.get_spend_proof("invalid_tx_id") + raise Exception("Should throw exception for invalid key") + except Exception as e: + TxUtils.test_invalid_tx_hash_error(e) + + # test check with invalid tx hash + try: + wallet.check_spend_proof("invalid_tx_id", '', signature) + raise Exception("Should have thrown exception") + except Exception as e: + TxUtils.test_invalid_tx_hash_error(e) + + # test check with invalid message + signature = wallet.get_spend_proof(tx.hash, "This is the right message") + result = wallet.check_spend_proof(tx.hash, "This is the wrong message", signature) + assert result is False + + # test check with wrong signature + other_tx: MoneroTxWallet = txs[1] + assert other_tx.hash is not None + signature = wallet.get_spend_proof(other_tx.hash, "This is the right message") + result = wallet.check_spend_proof(tx.hash, "This is the right message", signature) + assert result is False + + # Can prove reserves in the wallet + @pytest.mark.skipif(TestUtils.TEST_NON_RELAYS is False, reason="TEST_NON_RELAYS disablde") + def test_get_reserve_proof_wallet(self, wallet: MoneroWallet) -> None: + # get proof of entire wallet + signature: str = wallet.get_reserve_proof_wallet("Test message") + + # check proof of entire wallet + check: MoneroCheckReserve = wallet.check_reserve_proof(wallet.get_primary_address(), "Test message", signature) + assert check.is_good is True + TxUtils.test_check_reserve(check) + balance: int = wallet.get_balance() + if balance != check.total_amount: + # TODO monero-wallet-rpc: this check fails with unconfirmed txs + query: MoneroTxQuery = MoneroTxQuery() + query.in_tx_pool = True + unconfirmed_txs: list[MoneroTxWallet] = wallet.get_txs(query) + assert len(unconfirmed_txs) > 0, "Reserve amount must equal balance unless wallet has unconfirmed txs" + + # test different wallet address + different_address: str = WalletUtils.get_external_wallet_address() + try: + wallet.check_reserve_proof(different_address, "Test message", signature) + raise Exception("Should have thrown exception") + except Exception as e: + TxUtils.test_no_subaddress_error(e) + + # test subaddress + try: + address: Optional[str] = wallet.get_subaddress(0, 1).address + assert address is not None + wallet.check_reserve_proof(address, "Test message", signature) + raise Exception("Should have thrown exception") + except Exception as e: + TxUtils.test_no_subaddress_error(e) + + # test wrong message + check = wallet.check_reserve_proof(wallet.get_primary_address(), "Wrong message", signature) + # TODO: specifically test reserve checks, probably separate objects + assert check.is_good is False + TxUtils.test_check_reserve(check) + + # test wrong signature + try: + wallet.check_reserve_proof(wallet.get_primary_address(), "Test message", "wrong signature") + raise Exception("Should have thrown exception") + except Exception as e: + TxUtils.test_signature_header_error(e) + + # Can prove reserves in an account + @pytest.mark.skipif(TestUtils.TEST_NON_RELAYS is False, reason="TEST_NON_RELAYS disablde") + def test_get_reserve_proof_account(self, wallet: MoneroWallet) -> None: + # test proofs of accounts + num_non_zero_tests: int = 0 + msg: str = "Test message" + accounts: list[MoneroAccount] = wallet.get_accounts() + signature: str = '' + for account in accounts: + assert account.balance is not None + assert account.index is not None + if account.balance > 0: + check_amount: int = int(account.balance/2) + signature = wallet.get_reserve_proof_account(account.index, check_amount, msg) + check: MoneroCheckReserve = wallet.check_reserve_proof(wallet.get_primary_address(), msg, signature) + assert check.is_good is True + TxUtils.test_check_reserve(check) + assert check.total_amount is not None + assert check.total_amount >= 0 + num_non_zero_tests += 1 + else: + try: + wallet.get_reserve_proof_account(account.index, account.balance, msg) + raise Exception("Should have thrown exception") + except Exception as e: + err_msg: str = str(e) + logger.debug(err_msg) + assert "Should have thrown exception" != err_msg, err_msg + + try: + wallet.get_reserve_proof_account(account.index, TxUtils.MAX_FEE, msg) + raise Exception("Should have thrown exception") + except Exception as e: + err_msg: str = str(e) + logger.debug(err_msg) + assert "Should have thrown exception" != err_msg, err_msg + + assert num_non_zero_tests > 1, "Must have more than one account with non-zero balance; run send-to-multiple tests" + + # test error when not enough balance for requested minimum reserve amount + try: + account: MoneroAccount = accounts[0] + assert account.balance is not None + amount: int = account.balance + TxUtils.MAX_FEE + proof: str = wallet.get_reserve_proof_account(0, amount, "Test message") + logger.info(f"Account balance: {wallet.get_balance(0)}") + logger.info(f"First account balance {account.balance}") + reserve: MoneroCheckReserve = wallet.check_reserve_proof(wallet.get_primary_address(), "Test message", proof) + try: + wallet.get_reserve_proof_account(0, amount, "Test message") + raise Exception("expecting this to succeed") + except Exception as e: + err_msg: str = str(e) + assert "expecting this to succeed" == err_msg, err_msg + + logger.info(f"Check reserve proof: {reserve.serialize()}") + raise Exception("Should have thrown exception but got reserve proof: https://github.com/monero-project/monero/issues/6595") + except Exception as e: + err_msg: str = str(e) + logger.debug(err_msg) + #assert "Should have thrown exception" not in err_msg, err_msg + + # test different wallet address + different_address: str = WalletUtils.get_external_wallet_address() + try: + wallet.check_reserve_proof(different_address, "Test message", signature) + raise Exception("Should have thrown exception") + except Exception as e: + err_msg: str = str(e) + logger.debug(err_msg) + assert "Should have thrown exception" != err_msg, err_msg + + # test subaddress + try: + address: Optional[str] = wallet.get_subaddress(0, 1).address + assert address is not None + wallet.check_reserve_proof(address, "Test message", signature) + raise Exception("Should have thrown exception") + except Exception as e: + err_msg: str = str(e) + logger.debug(err_msg) + assert "Should have thrown exception" != err_msg, err_msg + + # test wrong message + check: MoneroCheckReserve = wallet.check_reserve_proof(wallet.get_primary_address(), "Wrong message", signature) + # TODO: specifically test reserve checks, probably separate objects + assert check.is_good is False + TxUtils.test_check_reserve(check) + + # test wrong signature + try: + wallet.check_reserve_proof(wallet.get_primary_address(), "Test message", "wrong signature") + raise Exception("Should have thrown exception") + except Exception as e: + err_msg: str = str(e) + logger.debug(err_msg) + assert "Should have thrown exception" != err_msg, err_msg + # Can get and set a transaction note @pytest.mark.skipif(TestUtils.TEST_NON_RELAYS is False, reason="TEST_NON_RELAYS disabled") def test_set_tx_note(self, wallet: MoneroWallet) -> None: diff --git a/tests/test_monero_wallet_keys.py b/tests/test_monero_wallet_keys.py index 96b4112..53f8f3d 100644 --- a/tests/test_monero_wallet_keys.py +++ b/tests/test_monero_wallet_keys.py @@ -6,7 +6,7 @@ from monero import ( MoneroWalletKeys, MoneroWalletConfig, MoneroWallet, MoneroUtils, MoneroAccount, MoneroSubaddress, - MoneroError, MoneroDaemonRpc, MoneroDaemon + MoneroDaemonRpc, MoneroDaemon ) from utils import TestUtils as Utils, AssertUtils, WalletUtils, WalletType @@ -262,7 +262,7 @@ def test_get_subaddresses_by_indices(self, wallet: MoneroWallet) -> None: def test_create_subaddress(self, wallet: MoneroWallet) -> None: return super().test_create_subaddress(wallet) - @pytest.mark.xfail(raises=MoneroError, reason="Keys-only wallet does not have enumerable set of subaddresses") + @pytest.mark.xfail(raises=RuntimeError, reason="Keys-only wallet does not have enumerable set of subaddresses") @override def test_set_subaddress_label(self, wallet: MoneroWallet) -> None: return super().test_set_subaddress_label(wallet) @@ -432,6 +432,31 @@ def test_get_transfers_with_query(self, wallet: MoneroWallet) -> None: def test_rescan_blockchain(self, wallet: MoneroWallet) -> None: return super().test_rescan_blockchain(wallet) + @pytest.mark.not_supported + @override + def test_check_tx_key(self, wallet: MoneroWallet) -> None: + return super().test_check_tx_key(wallet) + + @pytest.mark.not_supported + @override + def test_check_tx_proof(self, wallet: MoneroWallet) -> None: + return super().test_check_tx_proof(wallet) + + @pytest.mark.not_supported + @override + def test_check_spend_proof(self, wallet: MoneroWallet) -> None: + return super().test_check_spend_proof(wallet) + + @pytest.mark.not_supported + @override + def test_get_reserve_proof_wallet(self, wallet: MoneroWallet) -> None: + return super().test_get_reserve_proof_wallet(wallet) + + @pytest.mark.not_supported + @override + def test_get_reserve_proof_account(self, wallet: MoneroWallet) -> None: + return super().test_get_reserve_proof_account(wallet) + #endregion #region Tests diff --git a/tests/test_monero_wallet_rpc.py b/tests/test_monero_wallet_rpc.py index 22abce4..e5ff515 100644 --- a/tests/test_monero_wallet_rpc.py +++ b/tests/test_monero_wallet_rpc.py @@ -354,4 +354,12 @@ def test_get_txs_with_payment_ids(self, wallet: MoneroWallet) -> None: def test_subtract_fee_from(self, wallet: MoneroWallet) -> None: return super().test_subtract_fee_from(wallet) + @pytest.mark.skip(reason="TODO wallet rpc can't find destinations in outgoing transfers") + def test_check_tx_key(self, wallet: MoneroWallet) -> None: + return super().test_check_tx_key(wallet) + + @pytest.mark.skip(reason="TODO wallet rpc can't find destinations in outgoing transfers") + def test_check_tx_proof(self, wallet: MoneroWallet) -> None: + return super().test_check_tx_proof(wallet) + #endregion diff --git a/tests/utils/integration_test_utils.py b/tests/utils/integration_test_utils.py index 2db92af..107ed74 100644 --- a/tests/utils/integration_test_utils.py +++ b/tests/utils/integration_test_utils.py @@ -40,15 +40,15 @@ def setup(cls, wallet_type: WalletType) -> None: logger.warning("Only RPC and FULL wallet are supported for integration tests") return - num_wallet_txs: int = len(wallet.get_txs()) + wallet_txs: list[MoneroTxWallet] = wallet.get_txs() + num_wallet_txs: int = len(wallet_txs) # fund wallet with mined coins and wait for unlocked balance txs = cls.fund_wallet_and_wait_for_unlocked(wallet) - num_txs: int = len(txs) # setup regtest first receive height - if TestUtils.REGTEST and num_wallet_txs == 0: - assert num_txs > 0 - tx_height: int | None = txs[0].get_height() + if TestUtils.REGTEST: + tx: MoneroTxWallet = txs[0] if num_wallet_txs == 0 else wallet_txs[0] + tx_height: int | None = tx.get_height() assert tx_height is not None TestUtils.FIRST_RECEIVE_HEIGHT = tx_height logger.debug(f"Set FIRST_RECEIVE_HEIGHT = {tx_height}") diff --git a/tests/utils/test_utils.py b/tests/utils/test_utils.py index ac7dbb9..1c70b9a 100644 --- a/tests/utils/test_utils.py +++ b/tests/utils/test_utils.py @@ -403,7 +403,7 @@ def get_wallet_rpc(cls) -> MoneroWalletRpc: cls._WALLET_RPC.open_wallet(cls.WALLET_NAME, cls.WALLET_PASSWORD) except MoneroRpcError as e: # -1 returned when wallet does not exist or fails to open e.g. it's already open by another application - if e.get_code() == -1: + if e.code == -1: # create wallet config = MoneroWalletConfig() config.path = cls.WALLET_NAME diff --git a/tests/utils/to_multiple_tx_sender.py b/tests/utils/to_multiple_tx_sender.py index a0390d9..8364bc0 100644 --- a/tests/utils/to_multiple_tx_sender.py +++ b/tests/utils/to_multiple_tx_sender.py @@ -3,7 +3,7 @@ from typing import Optional from monero import ( MoneroWallet, MoneroAccount, MoneroSubaddress, MoneroTxConfig, - MoneroTxPriority, MoneroDestination, MoneroTxWallet, MoneroError + MoneroTxPriority, MoneroDestination, MoneroTxWallet ) from utils import TxUtils, AssertUtils, TestUtils, TxContext @@ -188,7 +188,7 @@ def send(self) -> None: txs: list[MoneroTxWallet] = [] try: txs = self._wallet.create_txs(config) - except MoneroError as e: + except Exception as e: # test error applying subtractFromFee with split txs if self._subtract_fee_from_destinations and len(txs) == 0: if str(e) == "subtractfeefrom transfers cannot be split over multiple transactions yet": diff --git a/tests/utils/tx_utils.py b/tests/utils/tx_utils.py index e56f0d9..0ff7ab7 100644 --- a/tests/utils/tx_utils.py +++ b/tests/utils/tx_utils.py @@ -10,7 +10,7 @@ MoneroUtils, MoneroOutputWallet, MoneroTx, MoneroOutput, MoneroKeyImage, MoneroDaemon, MoneroTxConfig, MoneroTxSet, MoneroTransferQuery, - MoneroOutputQuery + MoneroOutputQuery, MoneroCheckTx, MoneroCheckReserve ) from .tx_context import TxContext @@ -717,6 +717,112 @@ def test_spend_tx(cls, spend_tx: Optional[MoneroTxWallet]) -> None: assert input_wallet.key_image.hex is not None assert len(input_wallet.key_image.hex) > 0 + @classmethod + def test_check_tx(cls, tx: Optional[MoneroTxWallet], check: MoneroCheckTx) -> None: + """ + Test check tx + + :param MoneroTxWallet | None tx: transaction to test + :param MoneroCheckTx check: transaction check to test + """ + assert tx is not None + assert check.is_good is not None + if check.is_good is True: + assert check.num_confirmations is not None + assert check.num_confirmations >= 0 + assert check.in_tx_pool is not None + GenUtils.test_unsigned_big_integer(check.received_amount) + if check.in_tx_pool is True: + assert check.num_confirmations == 0 + else: + # TODO (monero-wall-rpc) this fails (confirmations is 0) for (at least one) transaction + # that has 1 confirmation on test_check_tx_key() + assert check.num_confirmations > 0 + else: + assert check.in_tx_pool is None + assert check.num_confirmations is None + assert check.received_amount is None + + @classmethod + def test_check_reserve(cls, check: MoneroCheckReserve) -> None: + """ + Test wallet check reserve + + :param MoneroCheckReserve check: reserve check to test + """ + assert check.is_good is not None + if check.is_good is True: + assert check.total_amount is not None + GenUtils.test_unsigned_big_integer(check.total_amount) + assert check.total_amount >= 0 + + assert check.unconfirmed_spent_amount is not None + GenUtils.test_unsigned_big_integer(check.unconfirmed_spent_amount) + assert check.unconfirmed_spent_amount >= 0 + else: + assert check.total_amount is None + assert check.unconfirmed_spent_amount is None + + @classmethod + def test_invalid_address_error(cls, ex: Exception) -> None: + """ + Test exception is invalid address + + :param Exception ex: exception to test + """ + msg: str = str(ex) + assert msg == "Invalid address", msg + + @classmethod + def test_invalid_tx_hash_error(cls, ex: Exception) -> None: + """ + Test exception is invalid hash format + + :param Exception ex: exception to test + """ + msg: str = str(ex) + assert msg == "TX hash has invalid format", msg + + @classmethod + def test_invalid_tx_key_error(cls, ex: Exception) -> None: + """ + Test exception is invalid key error + + :param Exception ex: exception to test + """ + msg: str = str(ex) + assert msg == "Tx key has invalid format", msg + + @classmethod + def test_invalid_signature_error(cls, ex: Exception) -> None: + """ + Test exception is invalid signature error + + :param Exception ex: exception to test + """ + msg: str = str(ex) + assert msg == "Signature size mismatch with additional tx pubkeys", msg + + @classmethod + def test_no_subaddress_error(cls, ex: Exception) -> None: + """ + Test exception is no subaddress error + + :param Exception ex: exception to test + """ + msg: str = str(ex) + assert msg == "Address must not be a subaddress", msg + + @classmethod + def test_signature_header_error(cls, ex: Exception) -> None: + """ + Test exception is signature header error + + :param Exception ex: exception to test + """ + msg: str = str(ex) + assert msg == "Signature header check error", msg + @classmethod def get_and_test_txs(cls, wallet: MoneroWallet, query: Optional[MoneroTxQuery], ctx: Optional[TxContext], is_expected: Optional[bool], regtest: bool) -> list[MoneroTxWallet]: """Get and test txs from wallet""" diff --git a/tests/utils/wallet_utils.py b/tests/utils/wallet_utils.py index a9fee82..a7e0602 100644 --- a/tests/utils/wallet_utils.py +++ b/tests/utils/wallet_utils.py @@ -24,6 +24,9 @@ class WalletUtils(ABC): """Wallet test utilities""" + MAX_TX_PROOFS: Optional[int] = 25 + """maximum number of transactions to check for each proof, undefined to check all""" + #region Test Utils @classmethod