rpc: submit_block + /send_raw_transaction (#515)

* enable `submit_block` and `/send_raw_transaction`

* endpoint

* map

* not_relayed

* book

* log

* Update binaries/cuprated/src/rpc/service/tx_handler.rs

Co-authored-by: Boog900 <boog900@tutanota.com>

* review

* fix

---------

Co-authored-by: Boog900 <boog900@tutanota.com>

binaries/cuprated/src/main.rs

@@ -162,6 +162,7 @@ fn main() {
             blockchain_read_handle,
             context_svc.clone(),
             txpool_read_handle,
+            tx_handler,
         );
 
         // Start the command listener.

binaries/cuprated/src/p2p/request_handler.rs

@@ -401,7 +401,12 @@ where
         .ready()
         .await
         .expect(PANIC_CRITICAL_SERVICE_ERROR)
-        .call(IncomingTxs { txs, state })
+        .call(IncomingTxs {
+            txs,
+            state,
+            drop_relay_rule_errors: true,
+            do_not_relay: false,
+        })
         .await;
 
     match res {

binaries/cuprated/src/rpc/handlers/json_rpc.rs

@@ -5,6 +5,7 @@
 //! <https://github.com/Cuprate/cuprate/pull/355>
 
 use std::{
+    collections::HashMap,
     net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4},
     num::NonZero,
     time::{Duration, Instant},
@@ -58,6 +59,7 @@ use cuprate_types::{
 };
 
 use crate::{
+    blockchain::interface as blockchain_interface,
     constants::VERSION_BUILD,
     rpc::{
         constants::{FIELD_NOT_SUPPORTED, UNSUPPORTED_RPC_CALL},
@@ -80,7 +82,7 @@ pub async fn map_request(
         Req::GetBlockTemplate(r) => Resp::GetBlockTemplate(not_available()?),
         Req::GetBlockCount(r) => Resp::GetBlockCount(get_block_count(state, r).await?),
         Req::OnGetBlockHash(r) => Resp::OnGetBlockHash(on_get_block_hash(state, r).await?),
-        Req::SubmitBlock(r) => Resp::SubmitBlock(not_available()?),
+        Req::SubmitBlock(r) => Resp::SubmitBlock(submit_block(state, r).await?),
         Req::GenerateBlocks(r) => Resp::GenerateBlocks(not_available()?),
         Req::GetLastBlockHeader(r) => {
             Resp::GetLastBlockHeader(get_last_block_header(state, r).await?)
@@ -234,7 +236,13 @@ async fn submit_block(
     let block_id = Hex(block.hash());
 
     // Attempt to relay the block.
-    blockchain_manager::relay_block(todo!(), Box::new(block)).await?;
+    blockchain_interface::handle_incoming_block(
+        block,
+        HashMap::new(), // this function reads the txpool
+        &mut state.blockchain_read,
+        &mut state.txpool_read,
+    )
+    .await?;
 
     Ok(SubmitBlockResponse {
         base: helper::response_base(false),

binaries/cuprated/src/rpc/handlers/other_json.rs

@@ -16,6 +16,7 @@ use cuprate_constants::rpc::{
     MAX_RESTRICTED_GLOBAL_FAKE_OUTS_COUNT, RESTRICTED_SPENT_KEY_IMAGES_COUNT,
     RESTRICTED_TRANSACTIONS_COUNT,
 };
+use cuprate_dandelion_tower::TxState;
 use cuprate_helper::cast::usize_to_u64;
 use cuprate_hex::{Hex, HexVec};
 use cuprate_p2p_core::{client::handshaker::builder::DummyAddressBook, ClearNet};
@@ -49,11 +50,17 @@ use cuprate_types::{
 use crate::{
     rpc::{
         constants::UNSUPPORTED_RPC_CALL,
-        handlers::{helper, shared, shared::not_available},
-        service::{address_book, blockchain, blockchain_context, blockchain_manager, txpool},
+        handlers::{
+            helper,
+            shared::{self, not_available},
+        },
+        service::{
+            address_book, blockchain, blockchain_context, blockchain_manager, tx_handler, txpool,
+        },
         CupratedRpcHandler,
     },
     statics::START_INSTANT_UNIX,
+    txpool::IncomingTxs,
 };
 
 /// Map a [`OtherRequest`] to the function that will lead to a [`OtherResponse`].
@@ -69,7 +76,9 @@ pub async fn map_request(
         Req::GetTransactions(r) => Resp::GetTransactions(not_available()?),
         Req::GetAltBlocksHashes(r) => Resp::GetAltBlocksHashes(not_available()?),
         Req::IsKeyImageSpent(r) => Resp::IsKeyImageSpent(not_available()?),
-        Req::SendRawTransaction(r) => Resp::SendRawTransaction(not_available()?),
+        Req::SendRawTransaction(r) => {
+            Resp::SendRawTransaction(send_raw_transaction(state, r).await?)
+        }
         Req::SaveBc(r) => Resp::SaveBc(not_available()?),
         Req::GetPeerList(r) => Resp::GetPeerList(not_available()?),
         Req::SetLogLevel(r) => Resp::SetLogLevel(not_available()?),
@@ -442,14 +451,32 @@ async fn send_raw_transaction(
         }
     }
 
-    // TODO: handle to txpool service.
-    let tx_relay_checks =
-        txpool::check_maybe_relay_local(todo!(), tx, !request.do_not_relay).await?;
+    if state.is_restricted() && request.do_not_relay {
+        // FIXME: implement something like `/check_tx` in `cuprated/monerod`.
+        // boog900:
+        // > making nodes hold txs in their pool that don't get passed
+        // > around the network can cause issues, like targeted tx pool double spends
+        // > there is also no reason to have this for public RPC
+        return Err(anyhow!("do_not_relay is not supported on restricted RPC"));
+    }
+
+    let txs = vec![tx.serialize().into()];
+
+    let mut txs = IncomingTxs {
+        txs,
+        state: TxState::Local,
+        drop_relay_rule_errors: false,
+        do_not_relay: request.do_not_relay,
+    };
+
+    let tx_relay_checks = tx_handler::handle_incoming_txs(&mut state.tx_handler, txs).await?;
 
     if tx_relay_checks.is_empty() {
         return Ok(resp);
     }
 
+    resp.not_relayed = true;
+
     // <https://github.com/monero-project/monero/blob/cc73fe71162d564ffda8e549b79a350bca53c454/src/rpc/core_rpc_server.cpp#L124>
     fn add_reason(reasons: &mut String, reason: &'static str) {
         if !reasons.is_empty() {

binaries/cuprated/src/rpc/rpc_handler.rs

@@ -19,7 +19,7 @@ use cuprate_rpc_types::{
 use cuprate_txpool::service::TxpoolReadHandle;
 use cuprate_types::BlockTemplate;
 
-use crate::rpc::handlers;
+use crate::{rpc::handlers, txpool::IncomingTxHandler};
 
 /// TODO: use real type when public.
 #[derive(Clone)]
@@ -169,7 +169,8 @@ pub struct CupratedRpcHandler {
 
     /// Read handle to the transaction pool database.
     pub txpool_read: TxpoolReadHandle,
-    // TODO: handle to txpool service.
+
+    pub tx_handler: IncomingTxHandler,
 }
 
 impl CupratedRpcHandler {
@@ -179,12 +180,14 @@ impl CupratedRpcHandler {
         blockchain_read: BlockchainReadHandle,
         blockchain_context: BlockchainContextService,
         txpool_read: TxpoolReadHandle,
+        tx_handler: IncomingTxHandler,
     ) -> Self {
         Self {
             restricted,
             blockchain_read,
             blockchain_context,
             txpool_read,
+            tx_handler,
         }
     }
 }

binaries/cuprated/src/rpc/server.rs

@@ -19,6 +19,7 @@ use cuprate_txpool::service::TxpoolReadHandle;
 use crate::{
     config::RpcConfig,
     rpc::{rpc_handler::BlockchainManagerHandle, CupratedRpcHandler},
+    txpool::IncomingTxHandler,
 };
 
 /// Initialize the RPC server(s).
@@ -33,6 +34,7 @@ pub fn init_rpc_servers(
     blockchain_read: BlockchainReadHandle,
     blockchain_context: BlockchainContextService,
     txpool_read: TxpoolReadHandle,
+    tx_handler: IncomingTxHandler,
 ) {
     for ((enable, addr, request_byte_limit), restricted) in [
         (
@@ -76,6 +78,7 @@ pub fn init_rpc_servers(
             blockchain_read.clone(),
             blockchain_context.clone(),
             txpool_read.clone(),
+            tx_handler.clone(),
         );
 
         tokio::task::spawn(async move {
@@ -107,6 +110,8 @@ async fn run_rpc_server(
     let router = RouterBuilder::new()
         .json_rpc()
         .other_get_height()
+        .other_send_raw_transaction()
+        .other_sendrawtransaction()
         .fallback()
         .build()
         .with_state(rpc_handler);

binaries/cuprated/src/rpc/service.rs

@@ -16,4 +16,5 @@ pub(super) mod address_book;
 pub(super) mod blockchain;
 pub(super) mod blockchain_context;
 pub(super) mod blockchain_manager;
+pub(super) mod tx_handler;
 pub(super) mod txpool;

binaries/cuprated/src/rpc/service/tx_handler.rs

@@ -0,0 +1,69 @@
+use anyhow::{anyhow, Error};
+use cuprate_consensus::ExtendedConsensusError;
+use cuprate_consensus_rules::{transactions::TransactionError, ConsensusError};
+use tower::{Service, ServiceExt};
+
+use cuprate_types::TxRelayChecks;
+
+use crate::txpool::{IncomingTxError, IncomingTxHandler, IncomingTxs, RelayRuleError};
+
+pub async fn handle_incoming_txs(
+    tx_handler: &mut IncomingTxHandler,
+    incoming_txs: IncomingTxs,
+) -> Result<TxRelayChecks, Error> {
+    let resp = tx_handler
+        .ready()
+        .await
+        .map_err(|e| anyhow!(e))?
+        .call(incoming_txs)
+        .await;
+
+    Ok(match resp {
+        Ok(()) => TxRelayChecks::empty(),
+        Err(e) => match e {
+            IncomingTxError::Consensus(ExtendedConsensusError::ConErr(
+                ConsensusError::Transaction(e),
+            )) => match e {
+                TransactionError::TooBig => TxRelayChecks::TOO_BIG,
+                TransactionError::KeyImageSpent => TxRelayChecks::DOUBLE_SPEND,
+
+                TransactionError::OutputNotValidPoint
+                | TransactionError::OutputTypeInvalid
+                | TransactionError::ZeroOutputForV1
+                | TransactionError::NonZeroOutputForV2
+                | TransactionError::OutputsOverflow
+                | TransactionError::OutputsTooHigh => TxRelayChecks::INVALID_OUTPUT,
+
+                TransactionError::MoreThanOneMixableInputWithUnmixable
+                | TransactionError::InvalidNumberOfOutputs
+                | TransactionError::InputDoesNotHaveExpectedNumbDecoys
+                | TransactionError::IncorrectInputType
+                | TransactionError::InputsAreNotOrdered
+                | TransactionError::InputsOverflow
+                | TransactionError::NoInputs => TxRelayChecks::INVALID_INPUT,
+
+                TransactionError::KeyImageIsNotInPrimeSubGroup
+                | TransactionError::AmountNotDecomposed
+                | TransactionError::DuplicateRingMember
+                | TransactionError::OneOrMoreRingMembersLocked
+                | TransactionError::RingMemberNotFoundOrInvalid
+                | TransactionError::RingSignatureIncorrect
+                | TransactionError::TransactionVersionInvalid
+                | TransactionError::RingCTError(_) => return Err(anyhow!("unreachable")),
+            },
+            IncomingTxError::Parse(_) | IncomingTxError::Consensus(_) => {
+                return Err(anyhow!("unreachable"))
+            }
+            IncomingTxError::RelayRule(RelayRuleError::NonZeroTimelock) => {
+                TxRelayChecks::NONZERO_UNLOCK_TIME
+            }
+            IncomingTxError::RelayRule(RelayRuleError::ExtraFieldTooLarge) => {
+                TxRelayChecks::TX_EXTRA_TOO_BIG
+            }
+            IncomingTxError::RelayRule(RelayRuleError::FeeBelowMinimum) => {
+                TxRelayChecks::FEE_TOO_LOW
+            }
+            IncomingTxError::DuplicateTransaction => TxRelayChecks::DOUBLE_SPEND,
+        },
+    })
+}

binaries/cuprated/src/rpc/service/txpool.rs

@@ -1,6 +1,10 @@
 //! Functions to send [`TxpoolReadRequest`]s.
 
-use std::{collections::HashSet, convert::Infallible, num::NonZero};
+use std::{
+    collections::{HashMap, HashSet},
+    convert::Infallible,
+    num::NonZero,
+};
 
 use anyhow::{anyhow, Error};
 use monero_serai::transaction::Transaction;
@@ -17,7 +21,7 @@ use cuprate_txpool::{
 };
 use cuprate_types::{
     rpc::{PoolInfo, PoolInfoFull, PoolInfoIncremental, PoolTxInfo, TxpoolStats},
-    TxInPool, TxRelayChecks,
+    TransactionVerificationData, TxInPool, TxRelayChecks,
 };
 
 // FIXME: use `anyhow::Error` over `tower::BoxError` in txpool.
@@ -222,6 +226,25 @@ pub async fn all_hashes(
     Ok(hashes)
 }
 
+/// [`TxpoolReadRequest::TxsForBlock`]
+pub async fn txs_for_block(
+    txpool_read: &mut TxpoolReadHandle,
+    tx_hashes: Vec<[u8; 32]>,
+) -> Result<(HashMap<[u8; 32], TransactionVerificationData>, Vec<usize>), Error> {
+    let TxpoolReadResponse::TxsForBlock { txs, missing } = txpool_read
+        .ready()
+        .await
+        .map_err(|e| anyhow!(e))?
+        .call(TxpoolReadRequest::TxsForBlock(tx_hashes))
+        .await
+        .map_err(|e| anyhow!(e))?
+    else {
+        unreachable!();
+    };
+
+    Ok((txs, missing))
+}
+
 /// TODO: impl txpool manager.
 pub async fn flush(txpool_manager: &mut Infallible, tx_hashes: Vec<[u8; 32]>) -> Result<(), Error> {
     todo!();
@@ -233,12 +256,3 @@ pub async fn relay(txpool_manager: &mut Infallible, tx_hashes: Vec<[u8; 32]>) ->
     todo!();
     Ok(())
 }
-
-/// TODO: impl txpool manager.
-pub async fn check_maybe_relay_local(
-    txpool_manager: &mut Infallible,
-    tx: Transaction,
-    relay: bool,
-) -> Result<TxRelayChecks, Error> {
-    Ok(todo!())
-}

binaries/cuprated/src/txpool.rs

@@ -12,3 +12,4 @@ mod relay_rules;
 mod txs_being_handled;
 
 pub use incoming_tx::{IncomingTxError, IncomingTxHandler, IncomingTxs};
+pub use relay_rules::RelayRuleError;

binaries/cuprated/src/txpool/incoming_tx.rs

@@ -40,7 +40,7 @@ use crate::{
     signals::REORG_LOCK,
     txpool::{
         dandelion,
-        relay_rules::check_tx_relay_rules,
+        relay_rules::{check_tx_relay_rules, RelayRuleError},
         txs_being_handled::{TxsBeingHandled, TxsBeingHandledLocally},
     },
 };
@@ -54,6 +54,8 @@ pub enum IncomingTxError {
     Consensus(ExtendedConsensusError),
     #[error("Duplicate tx in message")]
     DuplicateTransaction,
+    #[error("Relay rule was broken: {0}")]
+    RelayRule(RelayRuleError),
 }
 
 /// Incoming transactions.
@@ -62,6 +64,13 @@ pub struct IncomingTxs {
     pub txs: Vec<Bytes>,
     /// The routing state of the transactions.
     pub state: TxState<CrossNetworkInternalPeerId>,
+    /// If [`true`], transactions breaking relay
+    /// rules will be ignored and processing will continue,
+    /// otherwise the service will return an early error.
+    pub drop_relay_rule_errors: bool,
+    /// If [`true`], only checks will be done,
+    /// the transaction will not be relayed.
+    pub do_not_relay: bool,
 }
 
 ///  The transaction type used for dandelion++.
@@ -148,7 +157,12 @@ impl Service<IncomingTxs> for IncomingTxHandler {
 
 /// Handles the incoming txs.
 async fn handle_incoming_txs(
-    IncomingTxs { txs, state }: IncomingTxs,
+    IncomingTxs {
+        txs,
+        state,
+        drop_relay_rule_errors,
+        do_not_relay,
+    }: IncomingTxs,
     txs_being_handled: TxsBeingHandled,
     mut blockchain_context_cache: BlockchainContextService,
     blockchain_read_handle: ConsensusBlockchainReadHandle,
@@ -183,29 +197,36 @@ async fn handle_incoming_txs(
         // TODO: this could be a DoS, if someone spams us with txs that violate these rules?
         // Maybe we should remember these invalid txs for some time to prevent them getting repeatedly sent.
         if let Err(e) = check_tx_relay_rules(&tx, context) {
-            tracing::debug!(err = %e, tx = hex::encode(tx.tx_hash), "Tx failed relay check, skipping.");
+            if drop_relay_rule_errors {
+                tracing::debug!(err = %e, tx = hex::encode(tx.tx_hash), "Tx failed relay check, skipping.");
+                continue;
+            }
 
-            continue;
+            return Err(IncomingTxError::RelayRule(e));
         }
 
-        handle_valid_tx(
-            tx,
-            state.clone(),
-            &mut txpool_write_handle,
-            &mut dandelion_pool_manager,
-        )
-        .await;
+        if !do_not_relay {
+            handle_valid_tx(
+                tx,
+                state.clone(),
+                &mut txpool_write_handle,
+                &mut dandelion_pool_manager,
+            )
+            .await;
+        }
     }
 
     // Re-relay any txs we got in the block that were already in our stem pool.
-    for stem_tx in stem_pool_txs {
-        rerelay_stem_tx(
-            &stem_tx,
-            state.clone(),
-            &mut txpool_read_handle,
-            &mut dandelion_pool_manager,
-        )
-        .await;
+    if !do_not_relay {
+        for stem_tx in stem_pool_txs {
+            rerelay_stem_tx(
+                &stem_tx,
+                state.clone(),
+                &mut txpool_read_handle,
+                &mut dandelion_pool_manager,
+            )
+            .await;
+        }
     }
 
     Ok(())

books/user/src/rpc.md

@@ -60,7 +60,7 @@ This section contains the development status of endpoints/methods in `cuprated`.
 | `prune_blockchain`             | ⚫     |
 | `relay_tx`                     | ⚪     |
 | `set_bans`                     | ⚪     |
-| `submit_block`                 | ⚪     |
+| `submit_block`                 | 🟠     |
 | `sync_info`                    | ⚪     |
 
 ## JSON endpoints
@@ -83,7 +83,7 @@ This section contains the development status of endpoints/methods in `cuprated`.
 | `/out_peers`                   | ⚪     |
 | `/pop_blocks`                  | ⚪     |
 | `/save_bc`                     | ⚪     |
-| `/send_raw_transaction`        | ⚪     |
+| `/send_raw_transaction`        | 🟠     |
 | `/set_bootstrap_daemon`        | ⚪     | Requires bootstrap implementation
 | `/set_limit`                   | ⚪     |
 | `/set_log_categories`          | ⚪     | Could be re-purposed to use `tracing` filters