23757460166

// Copyright (C) 2013-2020 Blockstack PBC, a public benefit corporation
// Copyright (C) 2020-2023 Stacks Open Internet Foundation
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program.  If not, see <http://www.gnu.org/licenses/>.

use regex::{Captures, Regex};
use serde_json;
use stacks_common::types::chainstate::StacksBlockId;
use stacks_common::types::net::PeerHost;

use crate::chainstate::stacks::db::{ExtendedStacksHeader, StacksChainState};
use crate::chainstate::stacks::Error as ChainError;
use crate::net::http::{
    parse_json, Error, HttpBadRequest, HttpChunkGenerator, HttpContentType, HttpNotFound,
    HttpRequest, HttpRequestContents, HttpRequestPreamble, HttpResponse, HttpResponseContents,
    HttpResponsePayload, HttpResponsePreamble, HttpServerError,
};
use crate::net::httpcore::{
    request, HttpRequestContentsExtensions as _, RPCRequestHandler, StacksHttpRequest,
    StacksHttpResponse,
};
use crate::net::{Error as NetError, StacksNodeState, TipRequest, MAX_HEADERS};
use crate::util_lib::db::{DBConn, Error as DBError};

#[derive(Clone)]
pub struct RPCHeadersRequestHandler {
    pub quantity: Option<u32>,
}

impl RPCHeadersRequestHandler {
    pub fn new() -> Self {
        Self { quantity: None }
    }
}

#[derive(Debug)]
pub struct StacksHeaderStream {
    /// index block hash of the block to download
    pub index_block_hash: StacksBlockId,
    /// offset into whatever is being read (the blob, or the file in the chunk store)
    pub offset: u64,
    /// total number of bytes read.
    pub total_bytes: u64,
    /// number of headers remaining to stream
    pub num_headers: u32,

    /// header buffer data
    pub end_of_stream: bool,
    pub corked: bool,

    /// connection to the underlying chainstate
    chainstate_db: DBConn,
    blocks_path: String,
}

impl StacksHeaderStream {
    pub fn new(
        chainstate: &StacksChainState,
        tip: &StacksBlockId,
        num_headers_requested: u32,
    ) -> Result<Self, ChainError> {
        let header_info = StacksChainState::load_staging_block_info(chainstate.db(), tip)?
            .ok_or(ChainError::NoSuchBlockError)?;

        let num_headers = if header_info.height < (num_headers_requested as u64) {
            header_info.height as u32
        } else {
            num_headers_requested
        };

        let db = chainstate.reopen_db()?;
        let blocks_path = chainstate.blocks_path.clone();

        Ok(StacksHeaderStream {
            index_block_hash: tip.clone(),
            offset: 0,
            total_bytes: 0,
            num_headers,
            end_of_stream: false,
            corked: false,
            chainstate_db: db,
            blocks_path,
        })
    }
}

/// Decode the HTTP request
impl HttpRequest for RPCHeadersRequestHandler {
    fn verb(&self) -> &'static str {
        "GET"
    }

    fn path_regex(&self) -> Regex {
        Regex::new(r#"^/v2/headers/(?P<quantity>[0-9]+)$"#).unwrap()
    }

    fn metrics_identifier(&self) -> &str {
        "/v2/headers/:height"
    }

    /// Try to decode this request.
    /// There's nothing to load here, so just make sure the request is well-formed.
    fn try_parse_request(
        &mut self,
        preamble: &HttpRequestPreamble,
        captures: &Captures,
        query: Option<&str>,
        _body: &[u8],
    ) -> Result<HttpRequestContents, Error> {
        if preamble.get_content_length() != 0 {
            return Err(Error::DecodeError(
                "Invalid Http request: expected 0-length body for GetInfo".to_string(),
            ));
        }

        let quantity = request::get_u32(captures, "quantity")?;
        self.quantity = Some(quantity);

        let contents = HttpRequestContents::new().query_string(query);

        Ok(contents)
    }
}

impl RPCRequestHandler for RPCHeadersRequestHandler {
    /// Reset internal state
    fn restart(&mut self) {
        self.quantity = None;
    }

    /// Make the response
    fn try_handle_request(
        &mut self,
        preamble: HttpRequestPreamble,
        contents: HttpRequestContents,
        node: &mut StacksNodeState,
    ) -> Result<(HttpResponsePreamble, HttpResponseContents), NetError> {
        let quantity = self
            .quantity
            .take()
            .ok_or(NetError::SendError("`quantity` not set".to_string()))?;
        if (quantity as usize) > MAX_HEADERS {
            return StacksHttpResponse::new_error(
                &preamble,
                &HttpBadRequest::new(format!(
                    "Invalid request: requested more than {} headers\n",
                    MAX_HEADERS
                )),
            )
            .try_into_contents()
            .map_err(NetError::from);
        }

        // find requested chain tip
        let tip = match node.load_stacks_chain_tip(&preamble, &contents) {
            Ok(tip) => tip,
            Err(error_resp) => {
                return error_resp.try_into_contents().map_err(NetError::from);
            }
        };

        let stream_res =
            node.with_node_state(|_network, _sortdb, chainstate, _mempool, _rpc_args| {
                StacksHeaderStream::new(chainstate, &tip, quantity)
            });

        // start loading headers
        let stream = match stream_res {
            Ok(stream) => stream,
            Err(ChainError::NoSuchBlockError) => {
                return StacksHttpResponse::new_error(
                    &preamble,
                    &HttpNotFound::new(format!("No such block {:?}\n", &tip)),
                )
                .try_into_contents()
                .map_err(NetError::from)
            }
            Err(e) => {
                // nope -- error trying to check
                let msg = format!("Failed to load block header: {:?}\n", &e);
                warn!("{}", &msg);
                return StacksHttpResponse::new_error(&preamble, &HttpServerError::new(msg))
                    .try_into_contents()
                    .map_err(NetError::from);
            }
        };

        let resp_preamble = HttpResponsePreamble::from_http_request_preamble(
            &preamble,
            200,
            "OK",
            None,
            HttpContentType::JSON,
        );
        Ok((
            resp_preamble,
            HttpResponseContents::from_stream(Box::new(stream)),
        ))
    }
}

/// Decode the HTTP response
impl HttpResponse for RPCHeadersRequestHandler {
    /// Decode this response from a byte stream.  This is called by the client to decode this
    /// message
    fn try_parse_response(
        &self,
        preamble: &HttpResponsePreamble,
        body: &[u8],
    ) -> Result<HttpResponsePayload, Error> {
        let headers: Vec<ExtendedStacksHeader> = parse_json(preamble, body)?;
        Ok(HttpResponsePayload::try_from_json(headers)?)
    }
}

/// Stream implementation for HeaderStreamData
impl HttpChunkGenerator for StacksHeaderStream {
    fn hint_chunk_size(&self) -> usize {
        4096
    }

    #[cfg_attr(test, mutants::skip)]
    fn generate_next_chunk(&mut self) -> Result<Vec<u8>, String> {
        if self.total_bytes == 0 {
            // headers are a JSON array.  Start by writing '[', then write each header, and
            // then write ']'
            test_debug!("Opening header stream");
            self.total_bytes += 1;
            return Ok(vec![b'[']);
        }
        if self.num_headers == 0 {
            test_debug!("End of header stream");
            self.end_of_stream = true;
        }
        if self.total_bytes > 0 && !self.end_of_stream && !self.corked {
            // have more data to send.
            // read next header as JSON
            match StacksChainState::read_extended_header(
                &self.chainstate_db,
                &self.blocks_path,
                &self.index_block_hash,
            ) {
                Ok(extended_header) => {
                    // serialize
                    let mut header_bytes = vec![];
                    serde_json::to_writer(&mut header_bytes, &extended_header).map_err(|e| {
                        let msg = format!("Failed to encoded Stacks header: {:?}", &e);
                        warn!("{}", &msg);
                        msg
                    })?;

                    // advance
                    self.index_block_hash = extended_header.parent_block_id;
                    self.num_headers -= 1;

                    if self.num_headers > 0 {
                        header_bytes.push(b',');
                    } else {
                        self.end_of_stream = true;
                    }

                    self.total_bytes += header_bytes.len() as u64;
                    return Ok(header_bytes);
                }
                Err(ChainError::DBError(DBError::NotFoundError)) => {
                    // end of headers
                    test_debug!("Header not found; ending stream");
                    self.end_of_stream = true;
                }
                Err(e) => {
                    warn!("Header DB error: {:?}", &e);
                    self.end_of_stream = true;
                    return Err(format!(
                        "Failed to read extended header {}: {:?}",
                        &self.index_block_hash, &e
                    ));
                }
            };
        }
        if self.end_of_stream && !self.corked {
            // sent all the headers we're gonna send.
            test_debug!("Corking header stream");
            self.corked = true;
            self.total_bytes += 1;
            return Ok(vec![b']']);
        }

        test_debug!("Header stream terminated");
        // end of stream and corked. we're done!
        return Ok(vec![]);
    }
}

impl StacksHttpRequest {
    pub fn new_getheaders(host: PeerHost, quantity: u64, tip_req: TipRequest) -> StacksHttpRequest {
        StacksHttpRequest::new_for_peer(
            host,
            "GET".into(),
            format!("/v2/headers/{}", quantity),
            HttpRequestContents::new().for_tip(tip_req),
        )
        .expect("FATAL: failed to construct request from infallible data")
    }
}

impl StacksHttpResponse {
    pub fn decode_stacks_headers(self) -> Result<Vec<ExtendedStacksHeader>, NetError> {
        let contents = self.get_http_payload_ok()?;
        let response_json: serde_json::Value = contents.try_into()?;
        let headers: Vec<ExtendedStacksHeader> = serde_json::from_value(response_json)
            .map_err(|_e| Error::DecodeError("Failed to decode JSON".to_string()))?;
        Ok(headers)
    }
}

1	// Copyright (C) 2013-2020 Blockstack PBC, a public benefit corporation
2	// Copyright (C) 2020-2023 Stacks Open Internet Foundation
3	//
4	// This program is free software: you can redistribute it and/or modify
5	// it under the terms of the GNU General Public License as published by
6	// the Free Software Foundation, either version 3 of the License, or
7	// (at your option) any later version.
8	//
9	// This program is distributed in the hope that it will be useful,
10	// but WITHOUT ANY WARRANTY; without even the implied warranty of
11	// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12	// GNU General Public License for more details.
13	//
14	// You should have received a copy of the GNU General Public License
15	// along with this program. If not, see <http://www.gnu.org/licenses/>.
16
17	use regex::{Captures, Regex};
18	use serde_json;
19	use stacks_common::types::chainstate::StacksBlockId;
20	use stacks_common::types::net::PeerHost;
21
22	use crate::chainstate::stacks::db::{ExtendedStacksHeader, StacksChainState};
23	use crate::chainstate::stacks::Error as ChainError;
24	use crate::net::http::{
25	parse_json, Error, HttpBadRequest, HttpChunkGenerator, HttpContentType, HttpNotFound,
26	HttpRequest, HttpRequestContents, HttpRequestPreamble, HttpResponse, HttpResponseContents,
27	HttpResponsePayload, HttpResponsePreamble, HttpServerError,
28	};
29	use crate::net::httpcore::{
30	request, HttpRequestContentsExtensions as _, RPCRequestHandler, StacksHttpRequest,
31	StacksHttpResponse,
32	};
33	use crate::net::{Error as NetError, StacksNodeState, TipRequest, MAX_HEADERS};
34	use crate::util_lib::db::{DBConn, Error as DBError};
35
36	#[derive(Clone)]
37	pub struct RPCHeadersRequestHandler {
38	pub quantity: Option<u32>,
39	}
40
41	impl RPCHeadersRequestHandler {
42	pub fn new() -> Self {	985,120✔
43	Self { quantity: None }	985,120✔
44	}	985,120✔
45	}
46
47	#[derive(Debug)]
48	pub struct StacksHeaderStream {
49	/// index block hash of the block to download
50	pub index_block_hash: StacksBlockId,
51	/// offset into whatever is being read (the blob, or the file in the chunk store)
52	pub offset: u64,
53	/// total number of bytes read.
54	pub total_bytes: u64,
55	/// number of headers remaining to stream
56	pub num_headers: u32,
57
58	/// header buffer data
59	pub end_of_stream: bool,
60	pub corked: bool,
61
62	/// connection to the underlying chainstate
63	chainstate_db: DBConn,
64	blocks_path: String,
65	}
66
67	impl StacksHeaderStream {
68	pub fn new(	×
69	chainstate: &StacksChainState,	×
70	tip: &StacksBlockId,	×
71	num_headers_requested: u32,	×
72	) -> Result<Self, ChainError> {	×
73	let header_info = StacksChainState::load_staging_block_info(chainstate.db(), tip)?	×
74	.ok_or(ChainError::NoSuchBlockError)?;	×
75
76	let num_headers = if header_info.height < (num_headers_requested as u64) {	×
77	header_info.height as u32	×
78	} else {
79	num_headers_requested	×
80	};
81
82	let db = chainstate.reopen_db()?;	×
83	let blocks_path = chainstate.blocks_path.clone();	×
84
85	Ok(StacksHeaderStream {	×
86	index_block_hash: tip.clone(),	×
87	offset: 0,	×
88	total_bytes: 0,	×
89	num_headers,	×
90	end_of_stream: false,	×
91	corked: false,	×
92	chainstate_db: db,	×
93	blocks_path,	×
94	})	×
95	}	×
96	}
97
98	/// Decode the HTTP request
99	impl HttpRequest for RPCHeadersRequestHandler {
100	fn verb(&self) -> &'static str {	985,120✔
101	"GET"	985,120✔
102	}	985,120✔
103
104	fn path_regex(&self) -> Regex {	1,970,240✔
105	Regex::new(r#"^/v2/headers/(?P<quantity>[0-9]+)$"#).unwrap()	1,970,240✔
106	}	1,970,240✔
107
108	fn metrics_identifier(&self) -> &str {	×
109	"/v2/headers/:height"	×
110	}	×
111
112	/// Try to decode this request.
113	/// There's nothing to load here, so just make sure the request is well-formed.
114	fn try_parse_request(	×
115	&mut self,	×
116	preamble: &HttpRequestPreamble,	×
117	captures: &Captures,	×
118	query: Option<&str>,	×
119	_body: &[u8],	×
120	) -> Result<HttpRequestContents, Error> {	×
121	if preamble.get_content_length() != 0 {	×
122	return Err(Error::DecodeError(	×
123	"Invalid Http request: expected 0-length body for GetInfo".to_string(),	×
124	));	×
125	}	×
126
127	let quantity = request::get_u32(captures, "quantity")?;	×
128	self.quantity = Some(quantity);	×
129
130	let contents = HttpRequestContents::new().query_string(query);	×
131
132	Ok(contents)	×
133	}	×
134	}
135
136	impl RPCRequestHandler for RPCHeadersRequestHandler {
137	/// Reset internal state
138	fn restart(&mut self) {	×
139	self.quantity = None;	×
140	}	×
141
142	/// Make the response
143	fn try_handle_request(	×
144	&mut self,	×
145	preamble: HttpRequestPreamble,	×
146	contents: HttpRequestContents,	×
147	node: &mut StacksNodeState,	×
148	) -> Result<(HttpResponsePreamble, HttpResponseContents), NetError> {	×
149	let quantity = self	×
150	.quantity	×
151	.take()	×
152	.ok_or(NetError::SendError("`quantity` not set".to_string()))?;	×
153	if (quantity as usize) > MAX_HEADERS {	×
154	return StacksHttpResponse::new_error(	×
155	&preamble,	×
156	&HttpBadRequest::new(format!(	×
157	"Invalid request: requested more than {} headers\n",	×
158	MAX_HEADERS	×
159	)),	×
160	)
161	.try_into_contents()	×
162	.map_err(NetError::from);	×
163	}	×
164
165	// find requested chain tip
166	let tip = match node.load_stacks_chain_tip(&preamble, &contents) {	×
167	Ok(tip) => tip,	×
168	Err(error_resp) => {	×
169	return error_resp.try_into_contents().map_err(NetError::from);	×
170	}
171	};
172
173	let stream_res =	×
174	node.with_node_state(\|_network, _sortdb, chainstate, _mempool, _rpc_args\| {	×
175	StacksHeaderStream::new(chainstate, &tip, quantity)	×
176	});	×
177
178	// start loading headers
179	let stream = match stream_res {	×
180	Ok(stream) => stream,	×
181	Err(ChainError::NoSuchBlockError) => {
182	return StacksHttpResponse::new_error(	×
183	&preamble,	×
184	&HttpNotFound::new(format!("No such block {:?}\n", &tip)),	×
185	)
186	.try_into_contents()	×
187	.map_err(NetError::from)	×
188	}
189	Err(e) => {	×
190	// nope -- error trying to check
191	let msg = format!("Failed to load block header: {:?}\n", &e);	×
192	warn!("{}", &msg);	×
193	return StacksHttpResponse::new_error(&preamble, &HttpServerError::new(msg))	×
194	.try_into_contents()	×
195	.map_err(NetError::from);	×
196	}
197	};
198
199	let resp_preamble = HttpResponsePreamble::from_http_request_preamble(	×
200	&preamble,	×
201	200,
202	"OK",	×
203	None,	×
204	HttpContentType::JSON,	×
205	);
206	Ok((	×
207	resp_preamble,	×
208	HttpResponseContents::from_stream(Box::new(stream)),	×
209	))	×
210	}	×
211	}
212
213	/// Decode the HTTP response
214	impl HttpResponse for RPCHeadersRequestHandler {
215	/// Decode this response from a byte stream. This is called by the client to decode this
216	/// message
217	fn try_parse_response(	×
218	&self,	×
219	preamble: &HttpResponsePreamble,	×
220	body: &[u8],	×
221	) -> Result<HttpResponsePayload, Error> {	×
222	let headers: Vec<ExtendedStacksHeader> = parse_json(preamble, body)?;	×
223	Ok(HttpResponsePayload::try_from_json(headers)?)	×
224	}	×
225	}
226
227	/// Stream implementation for HeaderStreamData
228	impl HttpChunkGenerator for StacksHeaderStream {
229	fn hint_chunk_size(&self) -> usize {	×
230	4096	×
231	}	×
232
233	#[cfg_attr(test, mutants::skip)]
234	fn generate_next_chunk(&mut self) -> Result<Vec<u8>, String> {	×
235	if self.total_bytes == 0 {	×
236	// headers are a JSON array. Start by writing '[', then write each header, and
237	// then write ']'
238	test_debug!("Opening header stream");	×
239	self.total_bytes += 1;	×
240	return Ok(vec![b'[']);	×
241	}	×
242	if self.num_headers == 0 {	×
243	test_debug!("End of header stream");	×
244	self.end_of_stream = true;	×
245	}	×
246	if self.total_bytes > 0 && !self.end_of_stream && !self.corked {	×
247	// have more data to send.
248	// read next header as JSON
249	match StacksChainState::read_extended_header(	×
250	&self.chainstate_db,	×
251	&self.blocks_path,	×
252	&self.index_block_hash,	×
253	) {
254	Ok(extended_header) => {	×
255	// serialize
256	let mut header_bytes = vec![];	×
257	serde_json::to_writer(&mut header_bytes, &extended_header).map_err(\|e\| {	×
258	let msg = format!("Failed to encoded Stacks header: {:?}", &e);	×
259	warn!("{}", &msg);	×
260	msg	×
261	})?;	×
262
263	// advance
264	self.index_block_hash = extended_header.parent_block_id;	×
265	self.num_headers -= 1;	×
266
267	if self.num_headers > 0 {	×
268	header_bytes.push(b',');	×
269	} else {	×
270	self.end_of_stream = true;	×
271	}	×
272
273	self.total_bytes += header_bytes.len() as u64;	×
274	return Ok(header_bytes);	×
275	}
276	Err(ChainError::DBError(DBError::NotFoundError)) => {
277	// end of headers
278	test_debug!("Header not found; ending stream");	×
279	self.end_of_stream = true;	×
280	}
281	Err(e) => {	×
282	warn!("Header DB error: {:?}", &e);	×
283	self.end_of_stream = true;	×
284	return Err(format!(	×
285	"Failed to read extended header {}: {:?}",	×
286	&self.index_block_hash, &e	×
287	));	×
288	}
289	};
290	}	×
291	if self.end_of_stream && !self.corked {	×
292	// sent all the headers we're gonna send.
293	test_debug!("Corking header stream");	×
294	self.corked = true;	×
295	self.total_bytes += 1;	×
296	return Ok(vec![b']']);	×
297	}	×
298
299	test_debug!("Header stream terminated");	×
300	// end of stream and corked. we're done!
301	return Ok(vec![]);	×
302	}	×
303	}
304
305	impl StacksHttpRequest {
306	pub fn new_getheaders(host: PeerHost, quantity: u64, tip_req: TipRequest) -> StacksHttpRequest {	×
307	StacksHttpRequest::new_for_peer(	×
308	host,	×
309	"GET".into(),	×
310	format!("/v2/headers/{}", quantity),	×
311	HttpRequestContents::new().for_tip(tip_req),	×
312	)
313	.expect("FATAL: failed to construct request from infallible data")	×
314	}	×
315	}
316
317	impl StacksHttpResponse {
318	pub fn decode_stacks_headers(self) -> Result<Vec<ExtendedStacksHeader>, NetError> {	×
319	let contents = self.get_http_payload_ok()?;	×
320	let response_json: serde_json::Value = contents.try_into()?;	×
321	let headers: Vec<ExtendedStacksHeader> = serde_json::from_value(response_json)	×
322	.map_err(\|_e\| Error::DecodeError("Failed to decode JSON".to_string()))?;	×
323	Ok(headers)	×
324	}	×
325	}

stacks-network / stacks-core / 23757460166

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous