Merge pull request #5253 from stacks-network/fix/5249

jcnelson · web-flow · commit 84a0f9a92ee9 · 2024-10-03T21:32:06.000Z
Fix/5249
diff --git a/stackslib/src/chainstate/nakamoto/mod.rs b/stackslib/src/chainstate/nakamoto/mod.rs
@@ -2533,6 +2533,18 @@ impl NakamotoChainState {
         Ok(result)
     }
 
+    /// Load a consensus hash for a Nakamoto header
+    pub fn get_block_header_nakamoto_tenure_id(
+        chainstate_conn: &Connection,
+        index_block_hash: &StacksBlockId,
+    ) -> Result<Option<ConsensusHash>, ChainstateError> {
+        let sql = "SELECT consensus_hash FROM nakamoto_block_headers WHERE index_block_hash = ?1";
+        let result = query_row_panic(chainstate_conn, sql, &[&index_block_hash], || {
+            "FATAL: multiple rows for the same block hash".to_string()
+        })?;
+        Ok(result)
+    }
+
     /// Load an epoch2 header
     pub fn get_block_header_epoch2(
         chainstate_conn: &Connection,
diff --git a/stackslib/src/chainstate/nakamoto/staging_blocks.rs b/stackslib/src/chainstate/nakamoto/staging_blocks.rs
@@ -271,7 +271,7 @@ impl<'a> NakamotoStagingBlocksConnRef<'a> {
             .optional()?)
     }
 
-    /// Get the rowid of a Nakamoto block
+    /// Get the rowid of a staging Nakamoto block
     pub fn get_nakamoto_block_rowid(
         &self,
         index_block_hash: &StacksBlockId,
@@ -282,6 +282,26 @@ impl<'a> NakamotoStagingBlocksConnRef<'a> {
         Ok(res)
     }
 
+    /// Get the tenure and parent block ID of a staging block.
+    /// Used for downloads
+    pub fn get_tenure_and_parent_block_id(
+        &self,
+        index_block_hash: &StacksBlockId,
+    ) -> Result<Option<(ConsensusHash, StacksBlockId)>, ChainstateError> {
+        let sql = "SELECT consensus_hash,parent_block_id FROM nakamoto_staging_blocks WHERE index_block_hash = ?1";
+        let args = params![index_block_hash];
+
+        let mut stmt = self.deref().prepare(sql)?;
+        Ok(stmt
+            .query_row(args, |row| {
+                let ch: ConsensusHash = row.get(0)?;
+                let parent_id: StacksBlockId = row.get(1)?;
+
+                Ok((ch, parent_id))
+            })
+            .optional()?)
+    }
+
     /// Get a Nakamoto block by index block hash, as well as its size.
     /// Verifies its integrity.
     /// Returns Ok(Some(block, size)) if the block was present
diff --git a/stackslib/src/main.rs b/stackslib/src/main.rs
@@ -36,6 +36,8 @@ use std::collections::{BTreeMap, HashMap, HashSet};
 use std::fs::File;
 use std::io::prelude::*;
 use std::io::BufReader;
+use std::net::{SocketAddr, TcpStream, ToSocketAddrs};
+use std::time::Duration;
 use std::{env, fs, io, process, thread};
 
 use blockstack_lib::burnchains::bitcoin::{spv, BitcoinNetworkType};
@@ -62,10 +64,12 @@ use blockstack_lib::clarity::vm::ClarityVersion;
 use blockstack_lib::core::{MemPoolDB, *};
 use blockstack_lib::cost_estimates::metrics::UnitMetric;
 use blockstack_lib::cost_estimates::UnitEstimator;
+use blockstack_lib::net::api::getinfo::RPCPeerInfoData;
 use blockstack_lib::net::db::LocalPeer;
+use blockstack_lib::net::httpcore::{send_http_request, StacksHttpRequest};
 use blockstack_lib::net::p2p::PeerNetwork;
 use blockstack_lib::net::relay::Relayer;
-use blockstack_lib::net::StacksMessage;
+use blockstack_lib::net::{GetNakamotoInvData, HandshakeData, StacksMessage, StacksMessageType};
 use blockstack_lib::util_lib::db::sqlite_open;
 use blockstack_lib::util_lib::strings::UrlString;
 use blockstack_lib::{clarity_cli, cli};
@@ -76,7 +80,7 @@ use stacks_common::codec::{read_next, StacksMessageCodec};
 use stacks_common::types::chainstate::{
     BlockHeaderHash, BurnchainHeaderHash, StacksAddress, StacksBlockId,
 };
-use stacks_common::types::net::PeerAddress;
+use stacks_common::types::net::{PeerAddress, PeerHost};
 use stacks_common::types::sqlite::NO_PARAMS;
 use stacks_common::types::MempoolCollectionBehavior;
 use stacks_common::util::hash::{hex_bytes, to_hex, Hash160};
@@ -85,6 +89,164 @@ use stacks_common::util::secp256k1::{Secp256k1PrivateKey, Secp256k1PublicKey};
 use stacks_common::util::vrf::VRFProof;
 use stacks_common::util::{get_epoch_time_ms, sleep_ms};
 
+struct P2PSession {
+    pub local_peer: LocalPeer,
+    peer_info: RPCPeerInfoData,
+    burn_block_hash: BurnchainHeaderHash,
+    stable_burn_block_hash: BurnchainHeaderHash,
+    tcp_socket: TcpStream,
+    seq: u32,
+}
+
+impl P2PSession {
+    /// Make a StacksMessage.  Sign it and set a sequence number.
+    fn make_peer_message(&mut self, payload: StacksMessageType) -> Result<StacksMessage, String> {
+        let mut msg = StacksMessage::new(
+            self.peer_info.peer_version,
+            self.peer_info.network_id,
+            self.peer_info.burn_block_height,
+            &self.burn_block_hash,
+            self.peer_info.stable_burn_block_height,
+            &self.stable_burn_block_hash,
+            payload,
+        );
+
+        msg.sign(self.seq, &self.local_peer.private_key)
+            .map_err(|e| format!("Failed to sign message {:?}: {:?}", &msg, &e))?;
+        self.seq = self.seq.wrapping_add(1);
+
+        Ok(msg)
+    }
+
+    /// Send a p2p message.
+    /// Returns error text on failure.
+    fn send_peer_message(&mut self, msg: StacksMessage) -> Result<(), String> {
+        msg.consensus_serialize(&mut self.tcp_socket)
+            .map_err(|e| format!("Failed to send message {:?}: {:?}", &msg, &e))
+    }
+
+    /// Receive a p2p message.
+    /// Returns error text on failure.
+    fn recv_peer_message(&mut self) -> Result<StacksMessage, String> {
+        let msg: StacksMessage = read_next(&mut self.tcp_socket)
+            .map_err(|e| format!("Failed to receive message: {:?}", &e))?;
+        Ok(msg)
+    }
+
+    /// Begin a p2p session.
+    /// Synthesizes a LocalPeer from the remote peer's responses to /v2/info and /v2/pox.
+    /// Performs the initial handshake for you.
+    ///
+    /// Returns the session handle on success.
+    /// Returns error text on failure.
+    pub fn begin(peer_addr: SocketAddr, data_port: u16) -> Result<Self, String> {
+        let mut data_addr = peer_addr.clone();
+        data_addr.set_port(data_port);
+
+        // get /v2/info
+        let peer_info = send_http_request(
+            &format!("{}", data_addr.ip()),
+            data_addr.port(),
+            StacksHttpRequest::new_getinfo(PeerHost::from(data_addr.clone()), None)
+                .with_header("Connection".to_string(), "close".to_string()),
+            Duration::from_secs(60),
+        )
+        .map_err(|e| format!("Failed to query /v2/info: {:?}", &e))?
+        .decode_peer_info()
+        .map_err(|e| format!("Failed to decode response from /v2/info: {:?}", &e))?;
+
+        // convert `pox_consensus` and `stable_pox_consensus` into their respective burn block
+        // hashes
+        let sort_info = send_http_request(
+            &format!("{}", data_addr.ip()),
+            data_addr.port(),
+            StacksHttpRequest::new_get_sortition_consensus(
+                PeerHost::from(data_addr.clone()),
+                &peer_info.pox_consensus,
+            )
+            .with_header("Connection".to_string(), "close".to_string()),
+            Duration::from_secs(60),
+        )
+        .map_err(|e| format!("Failed to query /v3/sortitions: {:?}", &e))?
+        .decode_sortition_info()
+        .map_err(|e| format!("Failed to decode response from /v3/sortitions: {:?}", &e))?
+        .pop()
+        .ok_or_else(|| format!("No sortition returned for {}", &peer_info.pox_consensus))?;
+
+        let stable_sort_info = send_http_request(
+            &format!("{}", data_addr.ip()),
+            data_addr.port(),
+            StacksHttpRequest::new_get_sortition_consensus(
+                PeerHost::from(data_addr.clone()),
+                &peer_info.stable_pox_consensus,
+            )
+            .with_header("Connection".to_string(), "close".to_string()),
+            Duration::from_secs(60),
+        )
+        .map_err(|e| format!("Failed to query stable /v3/sortitions: {:?}", &e))?
+        .decode_sortition_info()
+        .map_err(|e| {
+            format!(
+                "Failed to decode response from stable /v3/sortitions: {:?}",
+                &e
+            )
+        })?
+        .pop()
+        .ok_or_else(|| {
+            format!(
+                "No sortition returned for {}",
+                &peer_info.stable_pox_consensus
+            )
+        })?;
+
+        let burn_block_hash = sort_info.burn_block_hash;
+        let stable_burn_block_hash = stable_sort_info.burn_block_hash;
+
+        let local_peer = LocalPeer::new(
+            peer_info.network_id,
+            peer_info.parent_network_id,
+            PeerAddress::from_socketaddr(&peer_addr),
+            peer_addr.port(),
+            Some(StacksPrivateKey::new()),
+            u64::MAX,
+            UrlString::try_from(format!("http://127.0.0.1:{}", data_port).as_str()).unwrap(),
+            vec![],
+        );
+
+        let tcp_socket = TcpStream::connect(&peer_addr)
+            .map_err(|e| format!("Failed to open {:?}: {:?}", &peer_addr, &e))?;
+
+        let mut session = Self {
+            local_peer,
+            peer_info,
+            burn_block_hash,
+            stable_burn_block_hash,
+            tcp_socket,
+            seq: 0,
+        };
+
+        // perform the handshake
+        let handshake_data =
+            StacksMessageType::Handshake(HandshakeData::from_local_peer(&session.local_peer));
+        let handshake = session.make_peer_message(handshake_data)?;
+        session.send_peer_message(handshake)?;
+
+        let resp = session.recv_peer_message()?;
+        match resp.payload {
+            StacksMessageType::HandshakeAccept(..)
+            | StacksMessageType::StackerDBHandshakeAccept(..) => {}
+            x => {
+                return Err(format!(
+                    "Peer returned unexpected message (expected HandshakeAccept variant): {:?}",
+                    &x
+                ));
+            }
+        }
+
+        Ok(session)
+    }
+}
+
 #[cfg_attr(test, mutants::skip)]
 fn main() {
     let mut argv: Vec<String> = env::args().collect();
@@ -974,6 +1136,36 @@ simulating a miner.
         process::exit(1);
     }
 
+    if argv[1] == "getnakamotoinv" {
+        if argv.len() < 5 {
+            eprintln!(
+                "Usage: {} getnakamotoinv HOST:PORT DATA_PORT CONSENSUS_HASH",
+                &argv[0]
+            );
+            process::exit(1);
+        }
+
+        let peer_addr: SocketAddr = argv[2].to_socket_addrs().unwrap().next().unwrap();
+        let data_port: u16 = argv[3].parse().unwrap();
+        let ch = ConsensusHash::from_hex(&argv[4]).unwrap();
+
+        let mut session = P2PSession::begin(peer_addr, data_port).unwrap();
+
+        // send getnakamotoinv
+        let get_nakamoto_inv =
+            StacksMessageType::GetNakamotoInv(GetNakamotoInvData { consensus_hash: ch });
+
+        let msg = session.make_peer_message(get_nakamoto_inv).unwrap();
+        session.send_peer_message(msg).unwrap();
+        let resp = session.recv_peer_message().unwrap();
+
+        let StacksMessageType::NakamotoInv(inv) = &resp.payload else {
+            panic!("Got spurious message: {:?}", &resp);
+        };
+
+        println!("{:?}", inv);
+    }
+
     if argv[1] == "replay-chainstate" {
         if argv.len() < 7 {
             eprintln!("Usage: {} OLD_CHAINSTATE_PATH OLD_SORTITION_DB_PATH OLD_BURNCHAIN_DB_PATH NEW_CHAINSTATE_PATH NEW_BURNCHAIN_DB_PATH", &argv[0]);
diff --git a/stackslib/src/net/api/getblock_v3.rs b/stackslib/src/net/api/getblock_v3.rs
@@ -181,20 +181,13 @@ impl RPCRequestHandler for RPCNakamotoBlockRequestHandler {
 
         let stream_res =
             node.with_node_state(|_network, _sortdb, chainstate, _mempool, _rpc_args| {
-                let Some(header) =
-                    NakamotoChainState::get_block_header_nakamoto(chainstate.db(), &block_id)?
+                let Some((tenure_id, parent_block_id)) = chainstate
+                    .nakamoto_blocks_db()
+                    .get_tenure_and_parent_block_id(&block_id)?
                 else {
                     return Err(ChainError::NoSuchBlockError);
                 };
-                let Some(nakamoto_header) = header.anchored_header.as_stacks_nakamoto() else {
-                    return Err(ChainError::NoSuchBlockError);
-                };
-                NakamotoBlockStream::new(
-                    chainstate,
-                    block_id.clone(),
-                    nakamoto_header.consensus_hash.clone(),
-                    nakamoto_header.parent_block_id.clone(),
-                )
+                NakamotoBlockStream::new(chainstate, block_id.clone(), tenure_id, parent_block_id)
             });
 
         // start loading up the block
diff --git a/stackslib/src/net/api/getsortition.rs b/stackslib/src/net/api/getsortition.rs
@@ -381,3 +381,34 @@ impl HttpResponse for GetSortitionHandler {
         Ok(HttpResponsePayload::try_from_json(sortition_info)?)
     }
 }
+
+impl StacksHttpRequest {
+    /// Make a new getsortition request to this endpoint
+    pub fn new_get_sortition(
+        host: PeerHost,
+        sort_key: &str,
+        sort_value: &str,
+    ) -> StacksHttpRequest {
+        StacksHttpRequest::new_for_peer(
+            host,
+            "GET".into(),
+            format!("{}/{}/{}", RPC_SORTITION_INFO_PATH, sort_key, sort_value),
+            HttpRequestContents::new(),
+        )
+        .expect("FATAL: failed to construct request from infallible data")
+    }
+
+    pub fn new_get_sortition_consensus(host: PeerHost, ch: &ConsensusHash) -> StacksHttpRequest {
+        Self::new_get_sortition(host, "consensus", &ch.to_string())
+    }
+}
+
+impl StacksHttpResponse {
+    pub fn decode_sortition_info(self) -> Result<Vec<SortitionInfo>, NetError> {
+        let contents = self.get_http_payload_ok()?;
+        let response_json: serde_json::Value = contents.try_into()?;
+        let response: Vec<SortitionInfo> = serde_json::from_value(response_json)
+            .map_err(|_e| Error::DecodeError(format!("Failed to decode JSON: {:?}", &_e)))?;
+        Ok(response)
+    }
+}
diff --git a/stackslib/src/net/chat.rs b/stackslib/src/net/chat.rs
@@ -1744,7 +1744,8 @@ impl ConversationP2P {
             &tip,
             sortdb,
             chainstate,
-            &network.stacks_tip.block_id(),
+            &network.stacks_tip.consensus_hash,
+            &network.stacks_tip.block_hash,
             reward_cycle,
         )?;
         let nakamoto_inv = NakamotoInvData::try_from(&bitvec_bools).map_err(|e| {
diff --git a/stackslib/src/net/download/nakamoto/tenure_downloader.rs b/stackslib/src/net/download/nakamoto/tenure_downloader.rs
@@ -376,8 +376,20 @@ impl NakamotoTenureDownloader {
         let mut expected_block_id = block_cursor;
         let mut count = 0;
         for block in tenure_blocks.iter() {
-            if &block.header.block_id() != expected_block_id {
+            // must be from this tenure
+            // This may not always be the case, since a remote peer could have processed a
+            // different Stacks micro-fork.  The consequence of erroring here (or below) is that we
+            // disconnect from the peer that served this to us.
+            if block.header.consensus_hash != self.tenure_id_consensus_hash {
                 warn!("Unexpected Nakamoto block -- not part of tenure";
+                      "block.header.consensus_hash" => %block.header.consensus_hash,
+                      "self.tenure_id_consensus_hash" => %self.tenure_id_consensus_hash,
+                      "state" => %self.state);
+                return Err(NetError::InvalidMessage);
+            }
+
+            if &block.header.block_id() != expected_block_id {
+                warn!("Unexpected Nakamoto block -- does not match cursor";
                       "expected_block_id" => %expected_block_id,
                       "block_id" => %block.header.block_id(),
                       "state" => %self.state);
diff --git a/stackslib/src/net/httpcore.rs b/stackslib/src/net/httpcore.rs
@@ -532,6 +532,12 @@ impl StacksHttpRequest {
         self.preamble.add_header(hdr, value);
     }
 
+    /// Constructor to add headers
+    pub fn with_header(mut self, hdr: String, value: String) -> Self {
+        self.add_header(hdr, value);
+        self
+    }
+
     /// Get a ref to all request headers
     pub fn get_headers(&self) -> &BTreeMap<String, String> {
         &self.preamble.headers
diff --git a/stackslib/src/net/inv/nakamoto.rs b/stackslib/src/net/inv/nakamoto.rs
diff --git a/stackslib/src/net/p2p.rs b/stackslib/src/net/p2p.rs
diff --git a/stackslib/src/net/tests/download/nakamoto.rs b/stackslib/src/net/tests/download/nakamoto.rs
diff --git a/stackslib/src/net/tests/inv/nakamoto.rs b/stackslib/src/net/tests/inv/nakamoto.rs
