1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
|
//! This module contains a WIP relay tunnel reactor.
//!
//! The initial version will duplicate some of the logic from
//! the client tunnel reactor.
//!
//! TODO(relay): refactor the relay tunnel
//! to share the same base tunnel implementation
//! as the client tunnel (to reduce code duplication).
//!
//! See the design notes at doc/dev/notes/relay-reactor.md
pub(crate) mod channel;
#[allow(unreachable_pub)] // TODO(relay): use in tor-chanmgr(?)
pub mod channel_provider;
pub(crate) mod reactor;
pub use channel::MaybeVerifiableRelayResponderChannel;
pub use channel::create_handler::{
CircNetParameters, CircuitIncomingStreamReceiver, CongestionControlNetParams,
CreateRequestHandler, IncomingStreamRequestFilterFactory,
};
use derive_deftly::Deftly;
use oneshot_fused_workaround as oneshot;
use tor_cell::chancell::msg::{self as chanmsg};
use tor_cell::relaycell::StreamId;
use tor_cell::relaycell::flow_ctrl::XonKBpsEwma;
use tor_memquota::derive_deftly_template_HasMemoryCost;
use crate::Error;
use crate::circuit::celltypes::derive_deftly_template_RestrictedChanMsgSet;
use crate::circuit::reactor::CircReactorHandle;
use crate::circuit::reactor::CtrlCmd;
use crate::circuit::reactor::forward;
use crate::relay::reactor::backward::Backward;
use crate::relay::reactor::forward::Forward;
use crate::stream::incoming::IncomingStreamRequestFilter;
/// A subclass of ChanMsg that can correctly arrive on a live relay
/// circuit (one where a CREATE* has been received).
#[derive(Debug, Deftly)]
#[derive_deftly(HasMemoryCost)]
#[derive_deftly(RestrictedChanMsgSet)]
#[deftly(usage = "on an open relay circuit")]
#[cfg(feature = "relay")]
#[cfg_attr(not(test), allow(unused))] // TODO(relay)
pub(crate) enum RelayCircChanMsg {
/// A relay cell telling us some kind of remote command from some
/// party on the circuit.
Relay(chanmsg::Relay),
/// A relay early cell that is allowed to contain a CREATE message.
RelayEarly(chanmsg::RelayEarly),
/// A cell telling us to destroy the circuit.
Destroy(chanmsg::Destroy),
/// A cell telling us to enable/disable channel padding.
PaddingNegotiate(chanmsg::PaddingNegotiate),
}
/// A handle for interacting with a relay circuit.
#[allow(unused)] // TODO(relay)
#[derive(Debug)]
pub struct RelayCirc(pub(crate) CircReactorHandle<Forward, Backward>);
impl RelayCirc {
/// Shut down this circuit, along with all streams that are using it.
/// Happens asynchronously (i.e. the tunnel won't necessarily be done shutting down
/// immediately after this function returns!).
///
/// Note that other references to this tunnel may exist.
/// If they do, they will stop working after you call this function.
///
/// It's not necessary to call this method if you're just done with a circuit:
/// the circuit should close on its own once nothing is using it any more.
pub fn terminate(&self) {
let _ = self.0.command.unbounded_send(CtrlCmd::Shutdown);
}
/// Return true if this circuit is closed and therefore unusable.
pub fn is_closing(&self) -> bool {
self.0.control.is_closed()
}
/// Inform the circuit reactor that there has been a change in the drain rate for this stream.
///
/// Typically the circuit reactor would send this new rate in an XON message to the other end of
/// the stream.
/// But it may decide not to, and may discard this update.
/// For example the stream may have a large amount of buffered data, and the reactor may not
/// want to send an XON while the buffer is large.
///
/// This sends a message to inform the circuit reactor of the new drain rate,
/// but it does not block or wait for a response from the reactor.
/// An error is only returned if we are unable to send the update.
//
// TODO(relay): this duplicates the ClientTunnel API and docs. Do we care?
pub(crate) fn drain_rate_update(
&self,
_stream_id: StreamId,
_rate: XonKBpsEwma,
) -> crate::Result<()> {
todo!()
}
/// Request to send a SENDME cell for this stream.
///
/// This sends a request to the circuit reactor to send a stream-level SENDME, but it does not
/// block or wait for a response from the circuit reactor.
/// An error is only returned if we are unable to send the request.
/// This means that if the circuit reactor is unable to send the SENDME, we are not notified of
/// this here and an error will not be returned.
//
// TODO(relay): this duplicates the ClientTunnel API and docs. Do we care?
pub(crate) fn send_sendme(&self, _stream_id: StreamId) -> crate::Result<()> {
todo!()
}
/// Close the pending stream that owns this StreamTarget, delivering the specified
/// END message (if any)
///
/// The stream is closed by sending a control message (`ClosePendingStream`)
/// to the reactor.
///
/// Returns a [`oneshot::Receiver`] that can be used to await the reactor's response.
///
/// The StreamTarget will set the correct stream ID and pick the
/// right hop, but will not validate that the message is well-formed
/// or meaningful in context.
///
/// Note that in many cases, the actual contents of an END message can leak unwanted
/// information. Please consider carefully before sending anything but an
/// [`End::new_misc()`](tor_cell::relaycell::msg::End::new_misc) message over a `ClientTunnel`.
/// (For onion services, we send [`DONE`](tor_cell::relaycell::msg::EndReason::DONE) )
///
/// In addition to sending the END message, this function also ensures
/// the state of the stream map entry of this stream is updated
/// accordingly.
///
/// Normally, you shouldn't need to call this function, as streams are implicitly closed by the
/// reactor when their corresponding `StreamTarget` is dropped. The only valid use of this
/// function is for closing pending incoming streams (a stream is said to be pending if we have
/// received the message initiating the stream but have not responded to it yet).
///
/// **NOTE**: This function should be called at most once per request.
/// Calling it twice is an error.
//
// TODO(relay): this duplicates the ClientTunnel API and docs. Do we care?
pub(crate) fn close_pending(
&self,
stream_id: StreamId,
message: crate::stream::CloseStreamBehavior,
) -> crate::Result<oneshot::Receiver<crate::Result<()>>> {
let (tx, rx) = oneshot::channel();
let cmd = forward::CtrlCmd::ClosePendingStream {
stream_id,
hop: None,
message,
done: tx,
};
self.0
.command
.unbounded_send(CtrlCmd::Forward(cmd))
.map_err(|_| Error::CircuitClosed)?;
Ok(rx)
}
}
#[cfg(test)]
mod test {
// @@ begin test lint list maintained by maint/add_warning @@
#![allow(clippy::bool_assert_comparison)]
#![allow(clippy::clone_on_copy)]
#![allow(clippy::dbg_macro)]
#![allow(clippy::mixed_attributes_style)]
#![allow(clippy::print_stderr)]
#![allow(clippy::print_stdout)]
#![allow(clippy::single_char_pattern)]
#![allow(clippy::unwrap_used)]
#![allow(clippy::unchecked_time_subtraction)]
#![allow(clippy::useless_vec)]
#![allow(clippy::needless_pass_by_value)]
#![allow(clippy::string_slice)] // See arti#2571
//! <!-- @@ end test lint list maintained by maint/add_warning @@ -->
#[test]
fn relay_circ_chan_msg() {
use tor_cell::chancell::msg::{self, AnyChanMsg};
fn good(m: AnyChanMsg) {
use crate::relay::RelayCircChanMsg;
assert!(RelayCircChanMsg::try_from(m).is_ok());
}
fn bad(m: AnyChanMsg) {
use crate::relay::RelayCircChanMsg;
assert!(RelayCircChanMsg::try_from(m).is_err());
}
good(msg::Destroy::new(2.into()).into());
bad(msg::CreatedFast::new(&b"The great globular mass"[..]).into());
bad(msg::Created2::new(&b"of protoplasmic slush"[..]).into());
good(msg::Relay::new(&b"undulated slightly,"[..]).into());
good(msg::AnyChanMsg::RelayEarly(
msg::Relay::new(&b"as if aware of him"[..]).into(),
));
bad(msg::Versions::new([1, 2, 3]).unwrap().into());
good(msg::PaddingNegotiate::start_default().into());
good(msg::RelayEarly::from(msg::Relay::new(b"snail-like unipedular organism")).into());
}
}
|
