summaryrefslogtreecommitdiff
path: root/crates/tor-proto/src/circuit
diff options
context:
space:
mode:
Diffstat (limited to 'crates/tor-proto/src/circuit')
-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
3 files changed, 103 insertions, 99 deletions
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(())
generated by cgit v1.2.3 (git 2.46.0) at 2026-08-29 05:41:58 +0000