tor-proto: Add a helper for adding a stream entry with a specific stream ID.

This adds a new `add_ent_with_id` function for adding a new entry to the
`StreamMap`. The existing `add_ent` function auto-generates a new stream
ID, which is not good if we're a hidden service, as stream IDs are
supposed to be chosen by the OP (client). When accepting a new stream,
services, exit relays, and dir auths need to use the stream ID received
in the BEGIN cell (instead of generating a new stream ID).

crates/tor-proto/src/circuit/streammap.rs

@@ -138,6 +138,41 @@ impl StreamMap {
         Err(Error::IdRangeFull)
     }
 
+    /// Add an entry to this map using the specified StreamId.
+    //
+    // TODO HSS: refacor this function and add_ent to reduce code duplication.
+    #[cfg(feature = "hs-service")]
+    pub(super) fn add_ent_with_id(
+        &mut self,
+        sink: mpsc::Sender<UnparsedRelayCell>,
+        rx: mpsc::Receiver<AnyRelayMsg>,
+        send_window: sendme::StreamSendWindow,
+        id: StreamId,
+        cmd_checker: AnyCmdChecker,
+    ) -> Result<()> {
+        // TODO HSS: perhaps StreamId should be a NonZeroU16.
+        if id.is_zero() {
+            return Err(Error::StreamIdZero);
+        }
+
+        let stream_ent = StreamEnt::Open {
+            sink,
+            rx,
+            send_window,
+            dropped: 0,
+            cmd_checker,
+        };
+
+        let ent = self.m.entry(id);
+        if let Entry::Vacant(_) = ent {
+            ent.or_insert(stream_ent);
+
+            Ok(())
+        } else {
+            Err(Error::IdUnavailable(id))
+        }
+    }
+
     /// Return the entry for `id` in this map, if any.
     pub(super) fn get_mut(&mut self, id: StreamId) -> Option<&mut StreamEnt> {
         self.m.get_mut(&id)

crates/tor-proto/src/util/err.rs

@@ -1,7 +1,7 @@
 //! Define an error type for the tor-proto crate.
 use std::{sync::Arc, time::Duration};
 use thiserror::Error;
-use tor_cell::relaycell::msg::EndReason;
+use tor_cell::relaycell::{msg::EndReason, StreamId};
 use tor_error::{ErrorKind, HasKind};
 use tor_linkspec::RelayIdType;
 
@@ -109,6 +109,12 @@ pub enum Error {
     /// Can't allocate any more circuit or stream IDs on a channel.
     #[error("Too many entries in map: can't allocate ID")]
     IdRangeFull,
+    /// Received a stream request with a stream ID that is already in use for another stream.
+    #[error("Stream ID {0} is already in use")]
+    IdUnavailable(StreamId),
+    /// Received a cell with a stream ID of zero.
+    #[error("Received a cell with a stream ID of zero")]
+    StreamIdZero,
     /// Couldn't extend a circuit because the extending relay or the
     /// target relay refused our request.
     #[error("Circuit extension refused: {0}")]
@@ -228,7 +234,9 @@ impl From<Error> for std::io::Error {
             | EncodeErr { .. }
             | ChanMismatch(_)
             | StreamProto(_)
-            | MissingId(_) => ErrorKind::InvalidData,
+            | MissingId(_)
+            | IdUnavailable(_)
+            | StreamIdZero => ErrorKind::InvalidData,
 
             Bug(ref e) if e.kind() == tor_error::ErrorKind::BadApiUsage => ErrorKind::InvalidData,
 
@@ -276,6 +284,8 @@ impl HasKind for Error {
             E::ResolveError(ResolveError::Transient) => EK::RemoteHostResolutionFailed,
             E::ResolveError(ResolveError::Unrecognized) => EK::RemoteHostResolutionFailed,
             E::MissingId(_) => EK::BadApiUsage,
+            E::IdUnavailable(_) => EK::BadApiUsage,
+            E::StreamIdZero => EK::BadApiUsage,
             E::Bug(e) => e.kind(),
         }
     }