/* This file is part of DarkFi (https://dark.fi) * * Copyright (C) 2020-2026 Dyne.org foundation * * This program is free software: you can redistribute it and/or modify * it under the terms of the GNU Affero 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 Affero General Public License for more details. * * You should have received a copy of the GNU Affero General Public License * along with this program. If not, see . */ use std::{io, time::Duration}; use smol::io::{AsyncReadExt, AsyncWriteExt, BufReader, ReadHalf, WriteHalf}; use tracing::error; use super::jsonrpc::*; use crate::net::transport::PtStream; pub(super) const INIT_BUF_SIZE: usize = 4096; // 4K pub(super) const MAX_BUF_SIZE: usize = 1024 * 1024 * 16; // 16M pub(super) const READ_TIMEOUT: Duration = Duration::from_secs(30); /// Internal read function that reads from the active stream into a buffer. /// Performs HTTP POST request parsing. Returns the request body length. pub(super) async fn http_read_from_stream_request( reader: &mut BufReader>>, buf: &mut Vec, ) -> io::Result { let mut total_read = 0; // Intermediate buffer we use to read byte-by-byte. let mut tmpbuf = [0_u8]; while total_read < MAX_BUF_SIZE { buf.resize(total_read + INIT_BUF_SIZE, 0u8); match reader.read(&mut tmpbuf).await { Ok(0) if total_read == 0 => return Err(io::ErrorKind::ConnectionAborted.into()), Ok(0) => break, // Finished reading Ok(_) => { // Copy the read byte to the destination buffer. buf[total_read] = tmpbuf[0]; total_read += 1; // In HTTP, when we reach '\r\n\r\n' we know we've read the headers. // The rest is the body. Headers should contain Content-Length which // tells us the remaining amount of bytes to read. if total_read > 4 && buf[total_read - 4..total_read] == *b"\r\n\r\n" { break } } Err(e) => return Err(e), } } // Here we parse the HTTP for correctness and find Content-Length let mut headers = [httparse::EMPTY_HEADER; 8]; let mut req = httparse::Request::new(&mut headers); match req.parse(buf) { Ok(httparse::Status::Complete(_)) => { /* Do nothing */ } Ok(httparse::Status::Partial) => { error!("[RPC] Failed parsing HTTP request: partial headers"); return Err(io::ErrorKind::InvalidData.into()) } Err(e) => { error!("[RPC] Failed parsing HTTP request: {e}"); return Err(io::ErrorKind::InvalidData.into()) } } let mut content_length: usize = 0; for header in headers { if header.name.to_lowercase() == "content-length" { let s = String::from_utf8_lossy(header.value); content_length = match s.parse() { Ok(v) => v, Err(_) => return Err(io::ErrorKind::InvalidData.into()), }; } } if content_length == 0 || content_length > MAX_BUF_SIZE { return Err(io::ErrorKind::InvalidData.into()) } // Now we know the request body size. Read it into the buffer. buf.clear(); buf.resize(content_length, 0_u8); reader.read_exact(buf).await?; Ok(content_length) } /// Internal read function that reads from the active stream into a buffer. /// Performs HTTP POST response parsing. Returns the response body length. pub(super) async fn http_read_from_stream_response( reader: &mut BufReader>>, buf: &mut Vec, ) -> io::Result { let mut total_read = 0; // Intermediate buffer we use to read byte-by-byte. let mut tmpbuf = [0_u8]; while total_read < MAX_BUF_SIZE { buf.resize(total_read + INIT_BUF_SIZE, 0u8); match reader.read(&mut tmpbuf).await { Ok(0) if total_read == 0 => return Err(io::ErrorKind::ConnectionAborted.into()), Ok(0) => break, // Finished reading Ok(_) => { // Copy the read byte to the destination buffer. buf[total_read] = tmpbuf[0]; total_read += 1; // In HTTP, when we reach '\r\n\r\n' we know we've read the headers. // The rest is the body. Headers should contain Content-Length which // tells us the remaining amount of bytes to read. if total_read > 4 && buf[total_read - 4..total_read] == *b"\r\n\r\n" { break } } Err(e) => return Err(e), } } // Here we parse the HTTP for correctness and find Content-Length let mut headers = [httparse::EMPTY_HEADER; 8]; let mut resp = httparse::Response::new(&mut headers); match resp.parse(buf) { Ok(httparse::Status::Complete(_)) => { /* Do nothing */ } Ok(httparse::Status::Partial) => { error!("[RPC] Failed parsing HTTP response: partial headers"); return Err(io::ErrorKind::InvalidData.into()) } Err(e) => { error!("[RPC] Failed parsing HTTP response: {e}"); return Err(io::ErrorKind::InvalidData.into()) } } let mut content_length: usize = 0; for header in headers { if header.name.to_lowercase() == "content-length" { let s = String::from_utf8_lossy(header.value); content_length = match s.parse() { Ok(v) => v, Err(_) => return Err(io::ErrorKind::InvalidData.into()), }; } } if content_length == 0 || content_length > MAX_BUF_SIZE { return Err(io::ErrorKind::InvalidData.into()) } // Now we know the response body size. Read it into the buffer. buf.clear(); buf.resize(content_length, 0_u8); reader.read_exact(buf).await?; Ok(content_length) } /// Internal read function that reads from the active stream into a buffer. /// Reading stops upon reaching CRLF or LF, or when `MAX_BUF_SIZE` is reached. pub(super) async fn read_from_stream( reader: &mut BufReader>>, buf: &mut Vec, ) -> io::Result { let mut total_read = 0; // Intermediate buffer we use to read byte-by-byte. let mut tmpbuf = [0_u8]; while total_read < MAX_BUF_SIZE { buf.resize(total_read + INIT_BUF_SIZE, 0u8); match reader.read(&mut tmpbuf).await { Ok(0) if total_read == 0 => return Err(io::ErrorKind::ConnectionAborted.into()), Ok(0) => break, // Finished reading Ok(_) => { // When we reach '\n', pop a possible '\r' from the buffer and bail. if tmpbuf[0] == b'\n' { if total_read > 0 && buf[total_read - 1] == b'\r' { buf.pop(); total_read -= 1; } buf.truncate(total_read); return Ok(total_read) } // Copy the read byte to the destination buffer. buf[total_read] = tmpbuf[0]; total_read += 1; } Err(e) => return Err(e), } } // Loop exited without finding \n. if total_read >= MAX_BUF_SIZE { return Err(io::Error::new( io::ErrorKind::InvalidData, "line exceeded maximum buf size without terminator", )) } // Truncate buffer to actual data size buf.truncate(total_read); Ok(total_read) } /// Internal write function that writes a JSON-RPC object to the active stream. /// Sent as an HTTP response. pub(super) async fn http_write_to_stream( writer: &mut WriteHalf>, object: &JsonResult, ) -> io::Result<()> { let (status_line, object_str) = match object { JsonResult::Notification(v) => ("HTTP/1.1 200 OK", v.stringify().unwrap()), JsonResult::Response(v) => ("HTTP/1.1 200 OK", v.stringify().unwrap()), JsonResult::Error(v) => ("HTTP/1.1 400 Bad Request", v.stringify().unwrap()), JsonResult::Request(v) => ("POST /json_rpc HTTP/1.1", v.stringify().unwrap()), _ => unreachable!(), }; let length = object_str.len(); let data = format!("{status_line}\r\nContent-Length: {length}\r\nContent-Type: application/json\r\n\r\n{object_str}"); writer.write_all(data.as_bytes()).await?; writer.flush().await?; Ok(()) } /// Internal write function that writes a JSON-RPC object to the active stream. pub(super) async fn write_to_stream( writer: &mut WriteHalf>, object: &JsonResult, ) -> io::Result<()> { let object_str = match object { JsonResult::Notification(v) => v.stringify().unwrap(), JsonResult::Response(v) => v.stringify().unwrap(), JsonResult::Error(v) => v.stringify().unwrap(), JsonResult::Request(v) => v.stringify().unwrap(), _ => unreachable!(), }; // As we're a line-based protocol, we append CRLF to the end of the JSON string. for i in [object_str.as_bytes(), b"\r\n"] { writer.write_all(i).await? } writer.flush().await?; Ok(()) }