summaryrefslogtreecommitdiff
path: root/crates
diff options
context:
space:
mode:
Diffstat (limited to 'crates')
-rw-r--r--crates/tor-proto/src/circuit.rs15
-rw-r--r--crates/tor-proto/src/circuit/halfstream.rs112
-rw-r--r--crates/tor-proto/src/circuit/reactor.rs49
-rw-r--r--crates/tor-proto/src/circuit/streammap.rs41
-rw-r--r--crates/tor-proto/src/stream.rs3
-rw-r--r--crates/tor-proto/src/stream/cmdcheck.rs55
-rw-r--r--crates/tor-proto/src/stream/data.rs60
-rw-r--r--crates/tor-proto/src/stream/resolve.rs40
8 files changed, 273 insertions, 102 deletions
diff --git a/crates/tor-proto/src/circuit.rs b/crates/tor-proto/src/circuit.rs
index f3dae893e..0868bac23 100644
--- a/crates/tor-proto/src/circuit.rs
+++ b/crates/tor-proto/src/circuit.rs
@@ -56,7 +56,10 @@ use crate::circuit::reactor::{
};
pub use crate::circuit::unique_id::UniqId;
use crate::crypto::cell::{HopNum, InboundClientCrypt, OutboundClientCrypt};
-use crate::stream::{DataStream, ResolveStream, StreamParameters, StreamReader};
+use crate::stream::{
+ AnyCmdChecker, DataCmdChecker, DataStream, ResolveCmdChecker, ResolveStream, StreamParameters,
+ StreamReader,
+};
use crate::{Error, ResolveError, Result};
use tor_cell::{
chancell::{self, msg::AnyChanMsg, CircId},
@@ -404,6 +407,7 @@ impl ClientCirc {
async fn begin_stream_impl(
&self,
begin_msg: AnyRelayMsg,
+ cmd_checker: AnyCmdChecker,
) -> Result<(StreamReader, StreamTarget)> {
// TODO: Possibly this should take a hop, rather than just
// assuming it's the last hop.
@@ -428,6 +432,7 @@ impl ClientCirc {
sender,
rx: msg_rx,
done: tx,
+ cmd_checker,
})
.map_err(|_| Error::CircuitClosed)?;
@@ -453,7 +458,9 @@ impl ClientCirc {
/// Start a DataStream (anonymized connection) to the given
/// address and port, using a BEGIN cell.
async fn begin_data_stream(&self, msg: AnyRelayMsg, optimistic: bool) -> Result<DataStream> {
- let (reader, target) = self.begin_stream_impl(msg).await?;
+ let (reader, target) = self
+ .begin_stream_impl(msg, DataCmdChecker::new_any())
+ .await?;
let mut stream = DataStream::new(reader, target);
if !optimistic {
stream.wait_for_connection().await?;
@@ -539,7 +546,9 @@ impl ClientCirc {
/// Helper: Send the resolve message, and read resolved message from
/// resolve stream.
async fn try_resolve(&self, msg: Resolve) -> Result<Resolved> {
- let (reader, _) = self.begin_stream_impl(msg.into()).await?;
+ let (reader, _) = self
+ .begin_stream_impl(msg.into(), ResolveCmdChecker::new_any())
+ .await?;
let mut resolve_stream = ResolveStream::new(reader);
resolve_stream.read_msg().await
}
diff --git a/crates/tor-proto/src/circuit/halfstream.rs b/crates/tor-proto/src/circuit/halfstream.rs
index f6204a171..b0cbf322f 100644
--- a/crates/tor-proto/src/circuit/halfstream.rs
+++ b/crates/tor-proto/src/circuit/halfstream.rs
@@ -3,10 +3,10 @@
//! A half-closed stream is one that we've sent an END on, but where
//! we might still receive some cells.
-use crate::circuit::sendme::{StreamRecvWindow, StreamSendWindow};
+use crate::circuit::sendme::{cmd_counts_towards_windows, StreamRecvWindow, StreamSendWindow};
+use crate::stream::{AnyCmdChecker, StreamStatus};
use crate::{Error, Result};
-use tor_cell::relaycell::UnparsedRelayCell;
-use tor_cell::restricted_msg;
+use tor_cell::relaycell::{RelayCmd, UnparsedRelayCell};
/// Type to track state of half-closed streams.
///
@@ -23,23 +23,8 @@ pub(super) struct HalfStream {
/// Receive window for this stream. Used to detect whether we get too
/// many data cells.
recvw: StreamRecvWindow,
- /// If true, accept a connected cell on this stream.
- connected_ok: bool,
-}
-
-restricted_msg! {
- enum HalfStreamMsg : RelayMsg {
- Sendme, Data, Connected, End, Resolved
- }
-}
-
-/// A status value returned by [`HalfStream::handle_msg`].
-#[derive(Debug, Copy, Clone, Eq, PartialEq)]
-pub(crate) enum HalfStreamStatus {
- /// The stream has been closed successfully and can now be dropped.
- Closed,
- /// The stream is still half,open, and must still be tracked.
- Open,
+ /// Object to tell us which cells to accept on this stream.
+ cmd_checker: AnyCmdChecker,
}
impl HalfStream {
@@ -47,12 +32,12 @@ impl HalfStream {
pub(super) fn new(
sendw: StreamSendWindow,
recvw: StreamRecvWindow,
- connected_ok: bool,
+ cmd_checker: AnyCmdChecker,
) -> Self {
HalfStream {
sendw,
recvw,
- connected_ok,
+ cmd_checker,
}
}
@@ -62,37 +47,25 @@ impl HalfStream {
/// The caller must handle END cells; it is an internal error to pass
/// END cells to this method.
/// no ends here.
- pub(super) fn handle_msg(&mut self, msg: UnparsedRelayCell) -> Result<HalfStreamStatus> {
- use HalfStreamMsg::*;
- use HalfStreamStatus::*;
- let msg = msg
- .decode::<HalfStreamMsg>()
- .map_err(|e| Error::from_bytes_err(e, "message on half-closed stream"))?
- .into_msg();
- match msg {
- Sendme(_) => {
- self.sendw.put(Some(()))?;
- Ok(Open)
- }
- Data(_) => {
- self.recvw.take()?;
- Ok(Open)
- }
- Connected(_) => {
- if self.connected_ok {
- self.connected_ok = false;
- Ok(Open)
- } else {
- Err(Error::CircProto(
- "Bad CONNECTED cell on a closed stream!".into(),
- ))
- }
- }
- End(_) => Ok(Closed),
- // TODO XXXX: We should only allow a Resolved() on streams where we sent
- // Resolve. My intended solution for #774 will fix this too.
- Resolved(_) => Ok(Closed),
+ pub(super) fn handle_msg(&mut self, msg: UnparsedRelayCell) -> Result<StreamStatus> {
+ use tor_cell::relaycell::msg::Sendme;
+ use StreamStatus::*;
+ if msg.cmd() == RelayCmd::SENDME {
+ // We handle SENDME separately, and don't give it to the checker.
+ let _ = msg
+ .decode::<Sendme>()
+ .map_err(|e| Error::from_bytes_err(e, "SENDME on half-closed stream"))?;
+ self.sendw.put(Some(()))?;
+ return Ok(Open);
+ }
+
+ if cmd_counts_towards_windows(msg.cmd()) {
+ self.recvw.take()?;
}
+
+ let status = self.cmd_checker.check_msg(&msg)?;
+ self.cmd_checker.consume_checked_msg(msg)?;
+ Ok(status)
}
}
@@ -109,7 +82,10 @@ mod test {
#![allow(clippy::unchecked_duration_subtraction)]
//! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
use super::*;
- use crate::circuit::sendme::{StreamRecvWindow, StreamSendWindow};
+ use crate::{
+ circuit::sendme::{StreamRecvWindow, StreamSendWindow},
+ stream::DataCmdChecker,
+ };
use rand::{CryptoRng, Rng};
use tor_basic_utils::test_rng::testing_rng;
use tor_cell::relaycell::{
@@ -132,7 +108,7 @@ mod test {
let mut sendw = StreamSendWindow::new(101);
sendw.take(&())?; // Make sure that it will accept one sendme.
- let mut hs = HalfStream::new(sendw, StreamRecvWindow::new(20), true);
+ let mut hs = HalfStream::new(sendw, StreamRecvWindow::new(20), DataCmdChecker::new_any());
// one sendme is fine
let m = msg::Sendme::new_empty();
@@ -152,7 +128,11 @@ mod test {
}
fn hs_new() -> HalfStream {
- HalfStream::new(StreamSendWindow::new(20), StreamRecvWindow::new(20), true)
+ HalfStream::new(
+ StreamSendWindow::new(20),
+ StreamRecvWindow::new(20),
+ DataCmdChecker::new_any(),
+ )
}
#[test]
@@ -160,6 +140,10 @@ mod test {
let mut hs = hs_new();
let mut rng = testing_rng();
+ // we didn't give a connected cell during setup, so do it now.
+ hs.handle_msg(to_unparsed(&mut rng, msg::Connected::new_empty().into()))
+ .unwrap();
+
// 20 data cells are okay.
let m = msg::Data::new(&b"this offer is unrepeatable"[..]).unwrap();
for _ in 0_u8..20 {
@@ -193,16 +177,26 @@ mod test {
.handle_msg(to_unparsed(&mut rng, m.clone().into()))
.is_err());
- // If we try that again with connected_ok == false, we won't
+ // If we try that again _after getting a connected_,
// accept any.
- let mut hs = HalfStream::new(StreamSendWindow::new(20), StreamRecvWindow::new(20), false);
+ let mut cmd_checker = DataCmdChecker::new_any();
+ {
+ cmd_checker
+ .check_msg(&to_unparsed(&mut rng, msg::Connected::new_empty().into()))
+ .unwrap();
+ }
+ let mut hs = HalfStream::new(
+ StreamSendWindow::new(20),
+ StreamRecvWindow::new(20),
+ cmd_checker,
+ );
let e = hs
.handle_msg(to_unparsed(&mut rng, m.into()))
.err()
.unwrap();
assert_eq!(
format!("{}", e),
- "Circuit protocol violation: Bad CONNECTED cell on a closed stream!"
+ "Stream protocol violation: Received CONNECTED twice on a stream."
);
}
@@ -217,7 +211,7 @@ mod test {
.unwrap();
assert_eq!(
format!("{}", e),
- "Unable to parse message on half-closed stream"
+ "Stream protocol violation: Unexpected EXTENDED2 on a data stream!"
);
}
}
diff --git a/crates/tor-proto/src/circuit/reactor.rs b/crates/tor-proto/src/circuit/reactor.rs
index 6ece08ed5..87040a79f 100644
--- a/crates/tor-proto/src/circuit/reactor.rs
+++ b/crates/tor-proto/src/circuit/reactor.rs
@@ -1,5 +1,20 @@
//! Code to handle incoming cells on a circuit.
-use super::halfstream::HalfStreamStatus;
+//!
+//! ## On message validation
+//!
+//! There are three steps for validating an incoming message on a stream:
+//!
+//! 1. Is the message contextually appropriate? (e.g., no more than one
+//! `CONNECTED` message per stream.) This is handled by calling
+//! [`CmdChecker::check_msg`](crate::stream::CmdChecker::check_msg).
+//! 2. Does the message comply with flow-control rules? (e.g., no more data than
+//! we've gotten SENDMEs for.) For open streams, the stream itself handles
+//! this; for half-closed streams, the reactor handles it using the
+//! `halfstream` module.
+//! 3. Does the message have an acceptable command type, and is the message
+//! well-formed? For open streams, the streams themselves handle this check.
+//! For half-closed streams, the reactor handles it by calling
+//! `consume_checked_msg()`.
use super::streammap::{ShouldSendEnd, StreamEnt};
use crate::circuit::celltypes::{ClientCircChanMsg, CreateResponse};
use crate::circuit::unique_id::UniqId;
@@ -10,6 +25,7 @@ use crate::crypto::cell::{
ClientLayer, CryptInit, HopNum, InboundClientCrypt, InboundClientLayer, OutboundClientCrypt,
OutboundClientLayer, RelayCellBody, Tor1RelayCrypto,
};
+use crate::stream::{AnyCmdChecker, StreamStatus};
use crate::util::err::{ChannelClosed, ReactorError};
use crate::{Error, Result};
use std::collections::VecDeque;
@@ -124,6 +140,8 @@ pub(super) enum CtrlMsg {
rx: mpsc::Receiver<AnyRelayMsg>,
/// Oneshot channel to notify on completion, with the allocated stream ID.
done: ReactorResultChannel<StreamId>,
+ /// A `CmdChecker` to keep track of which message types are acceptable.
+ cmd_checker: AnyCmdChecker,
},
/// Send a SENDME cell (used to ask for more data to be sent) on the given stream.
SendSendme {
@@ -1104,8 +1122,9 @@ impl Reactor {
sender,
rx,
done,
+ cmd_checker,
} => {
- let ret = self.begin_stream(cx, hop_num, message, sender, rx);
+ let ret = self.begin_stream(cx, hop_num, message, sender, rx, cmd_checker);
let _ = done.send(ret); // don't care if sender goes away
}
CtrlMsg::SendSendme { stream_id, hop_num } => {
@@ -1169,12 +1188,13 @@ impl Reactor {
message: AnyRelayMsg,
sender: mpsc::Sender<UnparsedRelayCell>,
rx: mpsc::Receiver<AnyRelayMsg>,
+ cmd_checker: AnyCmdChecker,
) -> Result<StreamId> {
let hop = self
.hop_mut(hopnum)
.ok_or_else(|| Error::from(internal!("No such hop {:?}", hopnum)))?;
let send_window = StreamSendWindow::new(SEND_WINDOW_INIT);
- let r = hop.map.add_ent(sender, rx, send_window)?;
+ let r = hop.map.add_ent(sender, rx, send_window, cmd_checker)?;
let cell = AnyRelayCell::new(r, message);
self.send_relay_cell(cx, hopnum, false, cell)?;
Ok(r)
@@ -1310,7 +1330,7 @@ impl Reactor {
sink,
send_window,
dropped,
- ref mut received_connected,
+ cmd_checker,
..
}) => {
// The stream for this message exists, and is open.
@@ -1327,18 +1347,7 @@ impl Reactor {
return Ok(CellStatus::Continue);
}
- if msg.cmd() == RelayCmd::CONNECTED {
- // Remember that we've received a Connected cell, and can't get another,
- // even if we become a HalfStream. (This rule is enforced separately at
- // DataStreamReader.)
-
- // TODO: This is problematic; see #774.
- *received_connected = true;
- }
-
- // Remember whether this was an end cell: if so we should
- // close the stream.
- let is_end_cell = msg.cmd() == RelayCmd::END;
+ let message_closes_stream = cmd_checker.check_msg(&msg)? == StreamStatus::Closed;
if let Err(e) = sink.try_send(msg) {
if e.is_full() {
@@ -1357,16 +1366,16 @@ impl Reactor {
*dropped += 1;
}
}
- if is_end_cell {
- hop.map.end_received(streamid)?;
+ if message_closes_stream {
+ hop.map.ending_msg_received(streamid)?;
}
}
Some(StreamEnt::EndSent(halfstream)) => {
// We sent an end but maybe the other side hasn't heard.
match halfstream.handle_msg(msg)? {
- HalfStreamStatus::Open => {}
- HalfStreamStatus::Closed => hop.map.end_received(streamid)?,
+ StreamStatus::Open => {}
+ StreamStatus::Closed => hop.map.ending_msg_received(streamid)?,
}
}
_ => {
diff --git a/crates/tor-proto/src/circuit/streammap.rs b/crates/tor-proto/src/circuit/streammap.rs
index 87245e41a..3b2b36c3a 100644
--- a/crates/tor-proto/src/circuit/streammap.rs
+++ b/crates/tor-proto/src/circuit/streammap.rs
@@ -2,6 +2,7 @@
use crate::circuit::halfstream::HalfStream;
use crate::circuit::sendme;
+use crate::stream::AnyCmdChecker;
use crate::{Error, Result};
use tor_cell::relaycell::UnparsedRelayCell;
/// Mapping from stream ID to streams.
@@ -33,9 +34,8 @@ pub(super) enum StreamEnt {
/// Number of cells dropped due to the stream disappearing before we can
/// transform this into an `EndSent`.
dropped: u16,
- /// True iff we've received a CONNECTED cell on this stream.
- /// (This is redundant with `DataStreamReader::connected`.)
- received_connected: bool,
+ /// A `CmdChecker` used to tell whether cells on this stream are valid.
+ cmd_checker: AnyCmdChecker,
},
/// A stream for which we have received an END cell, but not yet
/// had the stream object get dropped.
@@ -109,17 +109,14 @@ impl StreamMap {
sink: mpsc::Sender<UnparsedRelayCell>,
rx: mpsc::Receiver<AnyRelayMsg>,
send_window: sendme::StreamSendWindow,
+ cmd_checker: AnyCmdChecker,
) -> Result<StreamId> {
let stream_ent = StreamEnt::Open {
sink,
rx,
send_window,
dropped: 0,
- // TODO: This is true for all streams at this point, but it is
- // problematic to accept even one CONNECTED for RESOLVE/RESOLVED
- // streams, and will become more so once UDP streams are
- // implemented. This is #774.
- received_connected: false,
+ cmd_checker,
};
// This "65536" seems too aggressive, but it's what tor does.
//
@@ -146,10 +143,11 @@ impl StreamMap {
self.m.get_mut(&id)
}
- /// Note that we received an END cell on the stream with `id`.
+ /// Note that we received an END message (or other message indicating the end of
+ /// the stream) on the stream with `id`.
///
/// Returns true if there was really a stream there.
- pub(super) fn end_received(&mut self, id: StreamId) -> Result<()> {
+ pub(super) fn ending_msg_received(&mut self, id: StreamId) -> Result<()> {
// Check the hashmap for the right stream. Bail if not found.
// Also keep the hashmap handle so that we can do more efficient inserts/removals
let mut stream_entry = match self.m.entry(id) {
@@ -194,7 +192,7 @@ impl StreamMap {
StreamEnt::Open {
send_window,
dropped,
- received_connected,
+ cmd_checker,
// notably absent: the channels for sink and stream, which will get dropped and
// closed (meaning reads/writes from/to this stream will now fail)
..
@@ -205,9 +203,7 @@ impl StreamMap {
let mut recv_window = StreamRecvWindow::new(RECV_WINDOW_INIT);
recv_window.decrement_n(dropped)?;
// TODO: would be nice to avoid new_ref.
- // If we haven't gotten a CONNECTED already, we accept one on the half-stream.
- let connected_ok = !received_connected;
- let halfstream = HalfStream::new(send_window, recv_window, connected_ok);
+ let halfstream = HalfStream::new(send_window, recv_window, cmd_checker);
self.m.insert(id, StreamEnt::EndSent(halfstream));
Ok(ShouldSendEnd::Send)
}
@@ -234,7 +230,7 @@ mod test {
#![allow(clippy::unchecked_duration_subtraction)]
//! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
use super::*;
- use crate::circuit::sendme::StreamSendWindow;
+ use crate::{circuit::sendme::StreamSendWindow, stream::DataCmdChecker};
#[test]
fn streammap_basics() -> Result<()> {
@@ -246,7 +242,12 @@ mod test {
for _ in 0..128 {
let (sink, _) = mpsc::channel(128);
let (_, rx) = mpsc::channel(2);
- let id = map.add_ent(sink, rx, StreamSendWindow::new(500))?;
+ let id = map.add_ent(
+ sink,
+ rx,
+ StreamSendWindow::new(500),
+ DataCmdChecker::new_any(),
+ )?;
let expect_id: StreamId = next_id.into();
assert_eq!(expect_id, id);
next_id = next_id.wrapping_add(1);
@@ -262,10 +263,10 @@ mod test {
assert!(map.get_mut(nonesuch_id).is_none());
// Test end_received
- assert!(map.end_received(nonesuch_id).is_err());
- assert!(map.end_received(ids[1]).is_ok());
+ assert!(map.ending_msg_received(nonesuch_id).is_err());
+ assert!(map.ending_msg_received(ids[1]).is_ok());
assert!(matches!(map.get_mut(ids[1]), Some(StreamEnt::EndReceived)));
- assert!(map.end_received(ids[1]).is_err());
+ assert!(map.ending_msg_received(ids[1]).is_err());
// Test terminate
assert!(map.terminate(nonesuch_id).is_err());
@@ -275,7 +276,7 @@ mod test {
assert!(matches!(map.get_mut(ids[1]), None));
// Try receiving an end after a terminate.
- assert!(map.end_received(ids[2]).is_ok());
+ assert!(map.ending_msg_received(ids[2]).is_ok());
assert!(matches!(map.get_mut(ids[2]), None));
Ok(())
diff --git a/crates/tor-proto/src/stream.rs b/crates/tor-proto/src/stream.rs
index 6ccab73e8..f4f3ab55c 100644
--- a/crates/tor-proto/src/stream.rs
+++ b/crates/tor-proto/src/stream.rs
@@ -9,6 +9,7 @@
//!
//! There is no fairness, rate-limiting, or flow control.
+mod cmdcheck;
mod data;
#[cfg(feature = "onion-service")]
mod incoming;
@@ -16,6 +17,7 @@ mod params;
mod raw;
mod resolve;
+pub(crate) use cmdcheck::{AnyCmdChecker, CmdChecker, StreamStatus};
pub use data::{DataReader, DataStream, DataWriter};
#[cfg(feature = "onion-service")]
#[cfg_attr(docsrs, doc(cfg(feature = "onion-service")))]
@@ -23,5 +25,6 @@ pub use incoming::{IncomingStream, IncomingStreamRequest};
pub use params::StreamParameters;
pub use raw::StreamReader;
pub use resolve::ResolveStream;
+pub(crate) use {data::DataCmdChecker, resolve::ResolveCmdChecker};
pub use tor_cell::relaycell::msg::IpVersionPreference;
diff --git a/crates/tor-proto/src/stream/cmdcheck.rs b/crates/tor-proto/src/stream/cmdcheck.rs
new file mode 100644
index 000000000..76808e0df
--- /dev/null
+++ b/crates/tor-proto/src/stream/cmdcheck.rs
@@ -0,0 +1,55 @@
+//! Declare a "command checker" trait that checks whether a given relay message
+//! is acceptable on a given stream.
+
+use tor_cell::relaycell::UnparsedRelayCell;
+
+use crate::Result;
+
+/// A value returned by CmdChecker on success.
+#[derive(Copy, Clone, Debug, Eq, PartialEq)]
+pub(crate) enum StreamStatus {
+ /// The stream is still open.
+ Open,
+ /// The stream has been closed successfully; any further messages received
+ /// on this stream would be a protocol violation, which should cause
+ /// us to close the circuit.
+ Closed,
+}
+
+/// An object that checks incoming commands before they are sent to a stream.
+///
+/// These checks are called from the circuit reactor code, which runs in its own
+/// task. The reactor code continues calling these checks we have sent our own
+/// END cell on the stream. See `crate::circuit::halfstream` for more
+/// information.
+///
+/// NOTE: The checking DOES NOT take SENDME messages into account; those are
+/// handled separately. Neither of the methods on this trait will ever be
+/// passed a SENDME message.
+///
+/// See [`circuit::reactor`](crate::circuit::reactor) for more information on
+/// how these checks relate to other checks performed on incoming messages.
+pub(crate) trait CmdChecker: std::fmt::Debug {
+ /// Look at a message `msg` and decide whether it can be handled on this
+ /// stream.
+ ///
+ /// If `msg` is invalid, return an error, indicating that the protocol has
+ /// been violated and the corresponding circuit should be closed.
+ ///
+ /// If `msg` is invalid, update the state of this checker, and return a
+ /// `StreamStatus` indicating whether the last message closed.
+ fn check_msg(&mut self, msg: &UnparsedRelayCell) -> Result<StreamStatus>;
+
+ /// Consume `msg` and make sure it can be parsed correctly.
+ ///
+ /// This is an additional check, beyond check_msg(), performed for half-open
+ /// streams. It should only be called if check_msg() succeeds. It shouldn't
+ /// be called on open streams: for those, the stream itself parses the message
+ /// and consumes it.
+ fn consume_checked_msg(&mut self, msg: UnparsedRelayCell) -> Result<()>;
+}
+
+/// Type alias for a CmdChecker of unspecified type.
+//
+// TODO: Someday we might turn this into an enum if we decide it's beneficial.
+pub(crate) type AnyCmdChecker = Box<dyn CmdChecker + Send + 'static>;
diff --git a/crates/tor-proto/src/stream/data.rs b/crates/tor-proto/src/stream/data.rs
index 1d460e801..24a097713 100644
--- a/crates/tor-proto/src/stream/data.rs
+++ b/crates/tor-proto/src/stream/data.rs
@@ -3,6 +3,7 @@
use crate::{Error, Result};
use tor_cell::relaycell::msg::EndReason;
+use tor_cell::relaycell::RelayCmd;
use futures::io::{AsyncRead, AsyncWrite};
use futures::task::{Context, Poll};
@@ -28,6 +29,8 @@ use tor_basic_utils::skip_fmt;
use tor_cell::relaycell::msg::Data;
use tor_error::internal;
+use super::AnyCmdChecker;
+
/// An anonymized stream over the Tor network.
///
/// For most purposes, you can think of this type as an anonymized
@@ -657,3 +660,60 @@ impl DataReaderImpl {
}
}
}
+
+/// A `CmdChecker` that enforces invariants for outbound data streams.
+#[derive(Debug, Default)]
+pub(crate) struct DataCmdChecker {
+ /// True if we have received a CONNECTED message on this stream.
+ connected_received: bool,
+}
+
+impl super::CmdChecker for DataCmdChecker {
+ fn check_msg(
+ &mut self,
+ msg: &tor_cell::relaycell::UnparsedRelayCell,
+ ) -> Result<super::StreamStatus> {
+ use super::StreamStatus::*;
+ match msg.cmd() {
+ RelayCmd::CONNECTED => {
+ if self.connected_received {
+ Err(Error::StreamProto(
+ "Received CONNECTED twice on a stream.".into(),
+ ))
+ } else {
+ self.connected_received = true;
+ Ok(Open)
+ }
+ }
+ RelayCmd::DATA => {
+ if self.connected_received {
+ Ok(Open)
+ } else {
+ Err(Error::StreamProto(
+ "Received DATA before CONNECTED on a stream".into(),
+ ))
+ }
+ }
+ RelayCmd::END => Ok(Closed),
+ _ => Err(Error::StreamProto(format!(
+ "Unexpected {} on a data stream!",
+ msg.cmd()
+ ))),
+ }
+ }
+
+ fn consume_checked_msg(&mut self, msg: tor_cell::relaycell::UnparsedRelayCell) -> Result<()> {
+ let _ = msg
+ .decode::<DataStreamMsg>()
+ .map_err(|err| Error::from_bytes_err(err, "cell on half-closed stream"))?;
+ Ok(())
+ }
+}
+
+impl DataCmdChecker {
+ /// Return a new boxed `DataCmdChecker` in a state suitable for a newly
+ /// constructed connection.
+ pub(crate) fn new_any() -> AnyCmdChecker {
+ Box::<Self>::default()
+ }
+}
diff --git a/crates/tor-proto/src/stream/resolve.rs b/crates/tor-proto/src/stream/resolve.rs
index e81823e3d..f5c84b155 100644
--- a/crates/tor-proto/src/stream/resolve.rs
+++ b/crates/tor-proto/src/stream/resolve.rs
@@ -3,8 +3,11 @@
use crate::stream::StreamReader;
use crate::{Error, Result};
use tor_cell::relaycell::msg::Resolved;
+use tor_cell::relaycell::RelayCmd;
use tor_cell::restricted_msg;
+use super::AnyCmdChecker;
+
/// A ResolveStream represents a pending DNS request made with a RESOLVE
/// cell.
pub struct ResolveStream {
@@ -46,3 +49,40 @@ impl ResolveStream {
}
}
}
+
+/// A `CmdChecker` that enforces correctness for incoming commands on an
+/// outbound resolve stream.
+#[derive(Debug, Default)]
+pub(crate) struct ResolveCmdChecker {}
+
+impl super::CmdChecker for ResolveCmdChecker {
+ fn check_msg(
+ &mut self,
+ msg: &tor_cell::relaycell::UnparsedRelayCell,
+ ) -> Result<super::StreamStatus> {
+ use super::StreamStatus::Closed;
+ match msg.cmd() {
+ RelayCmd::RESOLVED => Ok(Closed),
+ RelayCmd::END => Ok(Closed),
+ _ => Err(Error::StreamProto(format!(
+ "Unexpected {} on resolve stream",
+ msg.cmd()
+ ))),
+ }
+ }
+
+ fn consume_checked_msg(&mut self, msg: tor_cell::relaycell::UnparsedRelayCell) -> Result<()> {
+ let _ = msg
+ .decode::<ResolveResponseMsg>()
+ .map_err(|err| Error::from_bytes_err(err, "message on resolve stream."))?;
+ Ok(())
+ }
+}
+
+impl ResolveCmdChecker {
+ /// Return a new boxed `DataCmdChecker` in a state suitable for a newly
+ /// constructed connection.
+ pub(crate) fn new_any() -> AnyCmdChecker {
+ Box::<Self>::default()
+ }
+}
generated by cgit v1.2.3 (git 2.46.0) at 2026-08-29 03:16:57 +0000