Lines
0 %
Functions
Branches
100 %
//! A simple reverse-proxy implementation for onion services.
use futures::io::BufReader;
use futures::{
AsyncRead, AsyncWrite, Future, FutureExt as _, Stream, StreamExt as _, select_biased,
};
use itertools::iproduct;
use oneshot_fused_workaround as oneshot;
use safelog::sensitive as sv;
use std::collections::HashMap;
use std::io::Error as IoError;
use std::sync::{Arc, Mutex};
use strum::IntoEnumIterator;
use tor_cell::relaycell::msg as relaymsg;
use tor_error::{ErrorKind, HasKind, debug_report};
use tor_hsservice::{HsNickname, RendRequest, StreamRequest};
use tor_log_ratelim::log_ratelim;
use tor_proto::client::stream::DataStream;
use tor_proto::stream::IncomingStreamRequest;
use tor_rtcompat::{Runtime, SpawnExt as _};
use crate::config::{
Encapsulation, ProxyAction, ProxyActionDiscriminants, ProxyConfig, TargetAddr,
/// A reverse proxy that handles connections from an `OnionService` by routing
/// them to local addresses.
#[derive(Debug)]
pub struct OnionServiceReverseProxy {
/// Mutable state held by this reverse proxy.
state: Mutex<State>,
}
/// Mutable part of an RProxy
struct State {
/// The current configuration for this reverse proxy.
config: ProxyConfig,
/// A sender that we'll drop when it's time to shut down this proxy.
shutdown_tx: Option<oneshot::Sender<void::Void>>,
/// A receiver that we'll use to monitor for shutdown signals.
shutdown_rx: futures::future::Shared<oneshot::Receiver<void::Void>>,
/// An error that prevents further progress while processing requests.
#[derive(Clone, Debug, thiserror::Error)]
#[non_exhaustive]
pub enum HandleRequestsError {
/// The runtime says it was unable to spawn a task.
#[error("Unable to spawn a task")]
Spawn(#[source] Arc<futures::task::SpawnError>),
impl HasKind for HandleRequestsError {
fn kind(&self) -> ErrorKind {
match self {
HandleRequestsError::Spawn(e) => e.kind(),
impl OnionServiceReverseProxy {
/// Create a new proxy with a given configuration.
pub fn new(config: ProxyConfig) -> Arc<Self> {
let (shutdown_tx, shutdown_rx) = oneshot::channel();
Arc::new(Self {
state: Mutex::new(State {
config,
shutdown_tx: Some(shutdown_tx),
shutdown_rx: shutdown_rx.shared(),
}),
})
/// Try to change the configuration of this proxy.
///
/// This change applies only to new connections through the proxy; existing
/// connections are not affected.
pub fn reconfigure(
&self,
how: tor_config::Reconfigure,
) -> Result<(), tor_config::ReconfigureError> {
if how == tor_config::Reconfigure::CheckAllOrNothing {
// Every possible reconfiguration is allowed.
return Ok(());
let mut state = self.state.lock().expect("poisoned lock");
state.config = config;
// Note: we don't need to use a postage::watch here, since we just want
// to lock this configuration whenever we get a request. We could use a
// Mutex<Arc<>> instead, but the performance shouldn't matter.
//
Ok(())
/// Shut down all request-handlers running using with this proxy.
pub fn shutdown(&self) {
let _ = state.shutdown_tx.take();
/// Use this proxy to handle a stream of [`RendRequest`]s.
/// The future returned by this function blocks indefinitely, so you may
/// want to spawn a separate task for it.
/// The provided nickname is used for logging.
pub async fn handle_requests<R, S>(
runtime: R,
nickname: HsNickname,
requests: S,
) -> Result<(), HandleRequestsError>
where
R: Runtime,
S: Stream<Item = RendRequest> + Unpin,
{
let mut stream_requests = tor_hsservice::handle_rend_requests(requests).fuse();
let mut shutdown_rx = self
.state
.lock()
.expect("poisoned lock")
.shutdown_rx
.clone()
.fuse();
let nickname = Arc::new(nickname);
/// Which of the three counters for each action
#[cfg(feature = "metrics")]
#[derive(Clone, Copy, Eq, PartialEq, Hash)]
enum CounterSelector {
/// Two counters, one for successes, one for failures
Ret(Result<(), ()>),
/// One counter for the total
Total,
let metrics_counters = {
use CounterSelector as CS;
let counters = iproduct!(
ProxyActionDiscriminants::iter(),
[
(CS::Total, "arti_hss_proxy_connections_total"),
(CS::Ret(Ok(())), "arti_hss_proxy_connections_ok_total"),
(CS::Ret(Err(())), "arti_hss_proxy_connections_failed_total"),
],
)
.map(|(action, (outcome, name))| {
let k = (action, outcome);
let nickname = nickname.to_string();
let action: &str = action.into();
let v = metrics::counter!(name, "nickname" => nickname, "action" => action);
(k, v)
.collect::<HashMap<(ProxyActionDiscriminants, CounterSelector), _>>();
Arc::new(counters)
loop {
let stream_request = select_biased! {
_ = shutdown_rx => return Ok(()),
stream_request = stream_requests.next() => match stream_request {
None => return Ok(()),
Some(s) => s,
runtime.spawn({
let action = self.choose_action(stream_request.request());
let runtime = runtime.clone();
let nickname = nickname.clone();
let req = stream_request.request().clone();
let metrics_counters = metrics_counters.clone();
async move {
let outcome =
run_action(runtime, nickname.as_ref(), action.clone(), stream_request).await;
let action = ProxyActionDiscriminants::from(&action);
let outcome = outcome.as_ref().map(|_|()).map_err(|_|());
for outcome in [CS::Total, CS::Ret(outcome)] {
if let Some(counter) = metrics_counters.get(&(action, outcome)) {
counter.increment(1);
} else {
// statically be impossible, but let's not panic
log_ratelim!(
"Performing action on {}", nickname;
outcome;
Err(_) => WARN, "Unable to take action {:?} for request {:?}", sv(action), sv(req)
);
.map_err(|e| HandleRequestsError::Spawn(Arc::new(e)))?;
/// Choose the configured action that we should take in response to a
/// [`StreamRequest`], based on our current configuration.
fn choose_action(&self, stream_request: &IncomingStreamRequest) -> ProxyAction {
let port: u16 = match stream_request {
IncomingStreamRequest::Begin(begin) => {
// The C tor implementation deliberately ignores the address and
// flags on the BEGIN message, so we do too.
begin.port()
other => {
tracing::warn!(
"Rejecting onion service request for invalid command {:?}. Internal error.",
other
return ProxyAction::DestroyCircuit;
self.state
.config
.resolve_port_for_begin(port)
.cloned()
// The default action is "destroy the circuit."
.unwrap_or(ProxyAction::DestroyCircuit)
/// Take the configured action from `action` on the incoming request `request`.
async fn run_action<R: Runtime>(
nickname: &HsNickname,
action: ProxyAction,
request: StreamRequest,
) -> Result<(), RequestFailed> {
match action {
ProxyAction::DestroyCircuit => {
request
.shutdown_circuit()
.map_err(RequestFailed::CantDestroy)?;
ProxyAction::Forward(encap, target) => match (encap, target) {
(Encapsulation::Simple, ref addr @ TargetAddr::Inet(a)) => {
let rt_clone = runtime.clone();
// We don't use any custom options on the socket.
let connect_options = Default::default();
let stream = runtime.connect(&a, &connect_options);
forward_connection(rt_clone, request, stream, nickname, addr).await?;
// TODO: we need more tests for Unix Socket support
// I will open a issue or just finish it in this MR
// (UnixTests)
#[cfg(unix)]
(Encapsulation::Simple, TargetAddr::Unix(unix_addr)) => {
// We don't use any custom options on the unix socket.
let unix_addr_clone = unix_addr.clone();
let stream = runtime.connect(&unix_addr_clone, &connect_options);
forward_connection(
rt_clone,
request,
stream,
nickname,
&TargetAddr::Unix(unix_addr),
.await?;
},
ProxyAction::RejectStream => {
// C tor sends DONE in this case, so we do too.
let end = relaymsg::End::new_with_reason(relaymsg::EndReason::DONE);
.reject(end)
.await
.map_err(RequestFailed::CantReject)?;
ProxyAction::IgnoreStream => drop(request),
/// An error from a single attempt to handle an onion service request.
#[derive(thiserror::Error, Debug, Clone)]
enum RequestFailed {
/// Encountered an error trying to destroy a circuit.
#[error("Unable to destroy onion service circuit")]
CantDestroy(#[source] tor_error::Bug),
/// Encountered an error trying to reject a single stream request.
#[error("Unable to reject onion service request")]
CantReject(#[source] tor_hsservice::ClientError),
/// Encountered an error trying to tell the remote onion service client that
/// we have accepted their connection.
#[error("Unable to accept onion service connection")]
AcceptRemote(#[source] tor_hsservice::ClientError),
/// The runtime refused to spawn a task for us.
#[error("Unable to spawn task")]
impl HasKind for RequestFailed {
RequestFailed::CantDestroy(e) => e.kind(),
RequestFailed::CantReject(e) => e.kind(),
RequestFailed::AcceptRemote(e) => e.kind(),
RequestFailed::Spawn(e) => e.kind(),
/// Size of buffer to use for communication between Arti and the
/// target service.
// This particular value is chosen more or less arbitrarily.
// Larger values let us do fewer reads from the application,
// but consume more memory.
// (The default value for BufReader is 8k as of this writing.)
const STREAM_BUF_LEN: usize = 4096;
/// Try to open a connection to an appropriate local target using
/// `target_stream_future`. If successful, try to report success on `request`
/// and transmit data between the two stream indefinitely. On failure, close
/// `request`.
/// Only return an error if we were unable to behave as intended due to a
/// problem we did not already report.
async fn forward_connection<R, FUT, TS>(
target_stream_future: FUT,
addr: &TargetAddr,
) -> Result<(), RequestFailed>
FUT: Future<Output = Result<TS, IoError>>,
TS: AsyncRead + AsyncWrite + Send + 'static,
let local_stream = target_stream_future.await.map_err(Arc::new);
// TODO: change this to "log_ratelim!(nickname=%nickname, ..." when log_ratelim can do that
// (we should search for HSS log messages and make them all be in the same form)
"Connecting to {} for onion service {}", sv(addr), nickname;
local_stream
let local_stream = match local_stream {
Ok(s) => s,
Err(_) => {
if let Err(e_rejecting) = request.reject(end).await {
debug_report!(
&e_rejecting,
"Unable to reject onion service request from client"
return Err(RequestFailed::CantReject(e_rejecting));
// We reported the (rate-limited) error from local_stream in
// DEBUG_REPORT above.
let onion_service_stream: DataStream = {
let connected = relaymsg::Connected::new_empty();
.accept(connected)
.map_err(RequestFailed::AcceptRemote)?
let onion_service_stream = BufReader::with_capacity(STREAM_BUF_LEN, onion_service_stream);
let local_stream = BufReader::with_capacity(STREAM_BUF_LEN, local_stream);
runtime
.spawn(
futures_copy::copy_buf_bidirectional(
onion_service_stream,
local_stream,
futures_copy::eof::Close,
.map(|_| ()),
.map_err(|e| RequestFailed::Spawn(Arc::new(e)))?;