sdk/src/guest_node.rs
Rama main · réplica de lectura del repositorio original.
//! This device's node in one share, and the guest API it speaks.
use crate::Observer;
use crate::storage;
use crate::transports::{
ConnectionCheck, InviteRefusedError, Stream, TerminalState, Transport, TransportError,
};
use anyhow::{Context, Result};
use http_body_util::{BodyExt, Full};
use hyper::StatusCode;
use hyper::client::conn::http1 as http1_client;
use hyper_util::rt::TokioIo;
use std::sync::{Arc, Mutex};
use tokio::time::Instant;
use tracing::{debug, warn};
use wire::{ConfigHash, ShareInfo};
use wispers_access_wire as wire;
/// This device's guest node in one share: its transport to the host node,
/// its row in the store, and the guest API calls it makes. Restored on the
/// share's first use, by `Client::restore_guest_node`.
pub struct GuestNode {
/// The `<share>` label in `<app>.<share>.wa.localhost`.
label: String,
row: storage::Row,
transport: Box<dyn Transport>,
observer: Arc<dyn Observer>,
/// Set once the transport reports a terminal failure mid-session. From
/// then on the share answers without dialing.
dead: Mutex<Option<TerminalState>>,
/// Held while `check_connection` probes the cached connection, so a
/// stream asked for meanwhile waits for the verdict rather than going
/// out on a connection that may be dead.
probe_gate: tokio::sync::Mutex<()>,
/// Time when the last QUIC stream was opened, used to determine whether or
/// not to reestablish the connection if it drops.
last_used: Mutex<Option<Instant>>,
}
impl GuestNode {
pub fn new(
label: String,
row: storage::Row,
transport: Box<dyn Transport>,
observer: Arc<dyn Observer>,
) -> Self {
Self {
label,
row,
transport,
observer,
dead: Mutex::new(None),
probe_gate: tokio::sync::Mutex::new(()),
last_used: Mutex::new(None),
}
}
/// Check connection health and drop/redial the connection if necessary.
/// Dead connections get redialed if they had been used recently (within
/// `RECENT_USE`). Otherwise, we just drop them. The probe delay is capped
/// by `PROBE_DEADLINE`.
pub async fn check_connection(&self) {
let _gate = self.probe_gate.lock().await;
match self.transport.check_connection(PROBE_DEADLINE).await {
Ok(ConnectionCheck::Alive) => {
debug!(share = self.label, "cached connection answered");
return;
}
Ok(ConnectionCheck::NoConnection) => {}
Ok(ConnectionCheck::Dead) => {
warn!(share = self.label, "cached connection is dead; dropped");
}
Err(TransportError::Terminal(state)) => {
self.record_terminal_error(state);
return;
}
Err(TransportError::Transient(e)) => {
warn!(
share = self.label,
error = format!("{e:#}"),
"could not check the connection"
);
}
}
if self.recently_used() {
self.reconnect().await;
}
}
fn recently_used(&self) -> bool {
self.last_used
.lock()
.expect("unpoisoned")
.is_some_and(|at| at.elapsed() < RECENT_USE)
}
/// Dials the host node now, in anticipation of future requests, with the
/// aim to cut down on user-visible latency. This ignores errors — followup
/// requests will deal with those.
async fn reconnect(&self) {
match self.open_stream_ungated().await {
Ok(_) => debug!(share = self.label, "reconnected ahead of the next request"),
Err(TransportError::Terminal(_)) => {}
Err(TransportError::Transient(e)) => warn!(
share = self.label,
error = format!("{e:#}"),
"could not reconnect ahead of the next request"
),
}
}
/// Forget the cached connection.
pub fn drop_connection(&self) {
self.transport.drop_connection();
}
/// Shut the transport down. Used when leaving a share.
pub async fn close(&self) {
self.transport.close().await;
}
pub fn label(&self) -> &str {
&self.label
}
/// Opens a REVERSE_PROXY stream for one app, ready for the HTTP request.
pub async fn open_reverse_proxy_stream(&self, app_id: &str) -> Result<Stream, TransportError> {
let mut stream = self.open_stream().await?;
wire::open_reverse_proxy_stream(&mut stream, app_id)
.await
.map_err(|e| TransportError::Transient(e.into()))?;
Ok(stream)
}
/// Opens a FORWARD_PROXY stream to `host:port`.
pub async fn open_forward_proxy_stream(
&self,
host: &str,
port: u16,
) -> Result<Stream, TransportError> {
let mut stream = self.open_stream().await?;
wire::open_forward_proxy_stream(&mut stream, host, port)
.await
.map_err(|e| TransportError::Transient(e.into()))?;
Ok(stream)
}
/// Asks the host node for the share if it changed since the stored copy,
/// and stores the answer. `None` = unchanged. This may reply with an error
/// indicating that the guest node's status has changed, e.g. it got revoked
/// or its SDK version is too old.
pub async fn refresh(&self) -> Result<Option<ShareInfo>, RefreshError> {
let stream = self.open_stream().await.map_err(|e| match e {
TransportError::Terminal(_) => RefreshError::Terminal,
TransportError::Transient(e) => RefreshError::Transient(e),
})?;
let known = self.row.read_share_config_hash()?;
let info = match fetch_info(stream, known).await {
Ok(info) => info,
Err(e) if e.is::<ClientTooOldError>() => {
if self.row.write_outdated(true)? {
warn!(
share = self.label,
"the host node needs a newer version of this app"
);
self.observer.on_share_changed(self.row.read_share()?);
}
return Err(RefreshError::Outdated);
}
Err(e) => return Err(RefreshError::Transient(e)),
};
let previously_outdated = self.row.write_outdated(false)?;
if let Some(info) = &info {
self.row.write_share_info(info)?;
}
if previously_outdated || info.is_some() {
self.observer.on_share_changed(self.row.read_share()?);
}
Ok(info)
}
/// Open a QUIC stream to the host node on the share's transport, opening a
/// new QUIC connection if necessary. This will wait for in-flight probes to
/// finish.
async fn open_stream(&self) -> Result<Stream, TransportError> {
drop(self.probe_gate.lock().await);
self.open_stream_ungated().await
}
/// Open a QUIC stream to the host node like `open_stream`, except without
/// blocking on in-flight probes. Any terminal errors (e.g. if this guest
/// node has been revoked) get recorded, to prevent futile retries in the
/// future.
async fn open_stream_ungated(&self) -> Result<Stream, TransportError> {
if let Some(state) = *self.dead.lock().expect("unpoisoned") {
return Err(TransportError::Terminal(state));
}
match self.transport.open_stream().await {
Err(TransportError::Terminal(state)) => {
self.record_terminal_error(state);
Err(TransportError::Terminal(state))
}
Ok(stream) => {
*self.last_used.lock().expect("unpoisoned") = Some(Instant::now());
Ok(stream)
}
other => other,
}
}
/// Record that we've received a terminal error from the host node,
/// preventing retries in the future.
fn record_terminal_error(&self, state: TerminalState) {
warn!(
share = self.label,
"share is no longer available: {}",
state.describe()
);
*self.dead.lock().expect("unpoisoned") = Some(state);
match self
.row
.write_terminal_state(state.as_str())
.and_then(|()| self.row.read_share())
{
Ok(share) => self.observer.on_share_changed(share),
Err(e) => warn!(
share = self.label,
error = format!("{e:#}"),
"could not record the share's state"
),
}
}
}
/// How recently a share must have been used for `check_connection` to
/// redial it unasked: long enough for a trip to another app and back, short
/// enough that this morning's share is not dialled at noon.
const RECENT_USE: std::time::Duration = std::time::Duration::from_secs(10 * 60);
/// How long a checked connection gets to answer. A live one answers in one
/// round trip, well under a second even on cellular; this only ever waits
/// on a dead one whose death the transport has not noticed yet.
const PROBE_DEADLINE: std::time::Duration = std::time::Duration::from_secs(2);
pub enum RefreshError {
/// The host node has turned this device away for good. The store has
/// the state.
Terminal,
/// This guest node's SDK is too old to talk to the host node.
Outdated,
Transient(anyhow::Error),
}
impl From<anyhow::Error> for RefreshError {
fn from(e: anyhow::Error) -> Self {
RefreshError::Transient(e)
}
}
/// The host node answered 426 Upgrade Required: it serves no SDK this old.
/// Any guest API call can get it, so it is one error, found by downcast.
#[derive(Debug, thiserror::Error)]
#[error("this app is too old for the share's host; update Wispers Access and try again")]
pub struct ClientTooOldError;
/// `GET /v1/share` over a fresh stream. `None` when the host node says the
/// share is unchanged since `known`. Also what `join` uses, before there is a
/// share to hang it on.
pub async fn fetch_info(stream: Stream, known: Option<ConfigHash>) -> Result<Option<ShareInfo>> {
let mut req = hyper::Request::builder()
.method(hyper::Method::GET)
.uri(wire::SHARE_PATH);
if let Some(known) = known {
req = req.header(hyper::header::IF_NONE_MATCH, known.etag());
}
let req = req
.body(Full::new(bytes::Bytes::new()))
.expect("static request is valid");
let resp = guest_api_request(stream, req)
.await
.context("GET /v1/share (is waserver up to date?)")?;
match resp.status() {
StatusCode::NOT_MODIFIED => Ok(None),
StatusCode::OK => Ok(Some(read_share(resp).await?)),
StatusCode::UPGRADE_REQUIRED => Err(ClientTooOldError.into()),
status => anyhow::bail!("the host node answered {} to GET /v1/share", status),
}
}
/// `POST /v1/activation` over a fresh stream on a connection whose key the host
/// node does not know yet: binds that key to the invite and returns the share.
/// A refusal carries the contract's reason.
pub async fn activate(stream: Stream, secret: &wire::InviteSecret) -> Result<ShareInfo> {
let body = serde_json::to_vec(&wire::Activation { secret: *secret })?;
let req = hyper::Request::builder()
.method(hyper::Method::POST)
.uri(wire::ACTIVATION_PATH)
.header(hyper::header::CONTENT_TYPE, "application/json")
.body(Full::new(bytes::Bytes::from(body)))
.expect("static request is valid");
let resp = guest_api_request(stream, req)
.await
.context("POST /v1/activation")?;
if resp.status() == StatusCode::OK {
return read_share(resp).await;
}
if resp.status() == StatusCode::UPGRADE_REQUIRED {
return Err(ClientTooOldError.into());
}
let status = resp.status();
let body = resp.into_body().collect().await?.to_bytes();
match serde_json::from_slice::<wire::ApiError>(&body) {
Ok(err) => match wire::ActivationError::parse(&err.error).and_then(refusal) {
Some(refused) => Err(refused.into()),
None => anyhow::bail!("activation refused: {}", err.error),
},
Err(_) => anyhow::bail!("the host node answered {} to the activation", status),
}
}
/// What to tell the user about an invite the host node refused. `None` for
/// refusals about this device's key rather than the invite, which a join's
/// fresh key never gets.
fn refusal(why: wire::ActivationError) -> Option<InviteRefusedError> {
use wire::ActivationError::*;
match why {
InviteConsumed => Some(InviteRefusedError(
"This invite has been used already. Ask for a new one.",
)),
InviteExpired => Some(InviteRefusedError(
"This invite has expired. Ask for a new one.",
)),
InviteUnknown => Some(InviteRefusedError(
"This invite is no longer valid. Ask for a new one.",
)),
Malformed | Revoked | AlreadyGuest => None,
}
}
/// `DELETE /v1/guest` over a fresh stream: this device leaves the share.
pub async fn leave(stream: Stream) -> Result<()> {
let req = hyper::Request::builder()
.method(hyper::Method::DELETE)
.uri(wire::GUEST_PATH)
.body(Full::new(bytes::Bytes::new()))
.expect("static request is valid");
let resp = guest_api_request(stream, req)
.await
.context("DELETE /v1/guest")?;
match resp.status() {
StatusCode::NO_CONTENT => Ok(()),
StatusCode::UPGRADE_REQUIRED => Err(ClientTooOldError.into()),
status => anyhow::bail!("the host node answered {} to DELETE /v1/guest", status),
}
}
/// One request to the guest API, on a CTRL stream that carries nothing else.
async fn guest_api_request(
mut stream: Stream,
mut req: hyper::Request<Full<bytes::Bytes>>,
) -> Result<hyper::Response<hyper::body::Incoming>> {
wire::open_ctrl_stream(&mut stream).await?;
let (mut sender, conn) = http1_client::handshake(TokioIo::new(stream))
.await
.context("guest API handshake (is waserver up to date?)")?;
tokio::spawn(async move {
let _ = conn.await;
});
let headers = req.headers_mut();
headers.insert(hyper::header::HOST, "waserver".parse().expect("valid"));
headers.insert(hyper::header::CONNECTION, "close".parse().expect("valid"));
headers.insert(
hyper::header::USER_AGENT,
wire::client_user_agent(env!("CARGO_PKG_VERSION"))
.parse()
.expect("a semver is a valid header value"),
);
Ok(sender.send_request(req).await?)
}
async fn read_share(resp: hyper::Response<hyper::body::Incoming>) -> Result<ShareInfo> {
let body = resp
.into_body()
.collect()
.await
.context("reading the share")?
.to_bytes();
serde_json::from_slice(&body).context("parsing the share")
}
#[cfg(test)]
mod tests {
use super::*;
use crate::NoObserver;
use crate::transports::BoxFuture;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
/// A transport whose host node never answers: streams open, then hang,
/// and a check finds the connection dead.
struct Silent {
connected: Arc<AtomicBool>,
dials: Arc<AtomicUsize>,
/// The far ends, kept so the streams stay open rather than see EOF.
far_ends: Mutex<Vec<tokio::io::DuplexStream>>,
}
impl Transport for Silent {
fn open_stream(&self) -> BoxFuture<'_, Result<Stream, TransportError>> {
Box::pin(async {
if !self.connected.swap(true, Ordering::SeqCst) {
self.dials.fetch_add(1, Ordering::SeqCst);
}
let (near, far) = tokio::io::duplex(4096);
self.far_ends.lock().expect("unpoisoned").push(far);
Ok(Box::new(near) as Stream)
})
}
fn check_connection(
&self,
_deadline: std::time::Duration,
) -> BoxFuture<'_, Result<ConnectionCheck, TransportError>> {
Box::pin(async {
// Nobody ever answers: a cached connection is found dead.
if self.connected.swap(false, Ordering::SeqCst) {
Ok(ConnectionCheck::Dead)
} else {
Ok(ConnectionCheck::NoConnection)
}
})
}
fn drop_connection(&self) {
self.connected.store(false, Ordering::SeqCst);
}
}
/// A transport whose host node answers every guest API request with
/// `status` and the JSON `body`, after noting the version the request
/// named.
struct Answering {
status: &'static str,
body: &'static str,
user_agents: Arc<Mutex<Vec<String>>>,
}
impl Transport for Answering {
fn open_stream(&self) -> BoxFuture<'_, Result<Stream, TransportError>> {
Box::pin(async {
let (near, far) = tokio::io::duplex(4096);
let (status, body) = (self.status, self.body);
let user_agents = self.user_agents.clone();
tokio::spawn(async move {
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
let mut far = BufReader::new(far);
// The CTRL stream's type byte, then HTTP/1.1 headers.
let mut first = [0u8; 1];
tokio::io::AsyncReadExt::read_exact(&mut far, &mut first)
.await
.unwrap();
let mut line = String::new();
loop {
line.clear();
far.read_line(&mut line).await.unwrap();
if let Some(ua) = line.strip_prefix("user-agent: ") {
user_agents.lock().unwrap().push(ua.trim().to_owned());
}
if line == "\r\n" {
break;
}
}
let response = format!(
"HTTP/1.1 {status}\r\ncontent-type: application/json\r\n\
content-length: {}\r\nconnection: close\r\n\r\n{body}",
body.len()
);
far.get_mut().write_all(response.as_bytes()).await.unwrap();
far.get_mut().shutdown().await.unwrap();
});
Ok(Box::new(near) as Stream)
})
}
fn check_connection(
&self,
_deadline: std::time::Duration,
) -> BoxFuture<'_, Result<ConnectionCheck, TransportError>> {
Box::pin(async { Ok(ConnectionCheck::NoConnection) })
}
fn drop_connection(&self) {}
}
#[tokio::test]
async fn a_426_marks_the_share_outdated_until_the_host_node_serves_again() {
let db = storage::DB::in_memory();
let row = db.new_row().unwrap();
row.write_deduped_hostname("rt").unwrap();
row.mark_complete().unwrap();
let user_agents = Arc::new(Mutex::new(Vec::new()));
// A node per answer; the row outlives them in the store.
let stored_row = || db.find_row("rt").unwrap().expect("the row");
let node = |status| {
GuestNode::new(
"rt".into(),
stored_row(),
Box::new(Answering {
status,
body: r#"{"error":"client-too-old"}"#,
user_agents: user_agents.clone(),
}),
Arc::new(NoObserver),
)
};
assert!(matches!(
node("426 Upgrade Required").refresh().await,
Err(RefreshError::Outdated)
));
assert!(stored_row().read_outdated().unwrap());
assert_eq!(
user_agents.lock().unwrap().as_slice(),
[wire::client_user_agent(env!("CARGO_PKG_VERSION"))]
);
// Updated and served again: back to live, without a share body.
assert!(matches!(node("304 Not Modified").refresh().await, Ok(None)));
assert!(!stored_row().read_outdated().unwrap());
}
#[tokio::test]
async fn an_invite_the_host_node_refuses_says_why() {
let activate = |status, body| async move {
let host = Answering {
status,
body,
user_agents: Default::default(),
};
let stream = host.open_stream().await.ok().expect("a stream");
activate(stream, &wire::InviteSecret([7; 16])).await
};
let used = activate("409 Conflict", r#"{"error":"invite-consumed"}"#)
.await
.expect_err("a used invite fails");
assert_eq!(
used.downcast_ref::<InviteRefusedError>()
.map(ToString::to_string),
Some("This invite has been used already. Ask for a new one.".into())
);
// About the key, not the invite, so no sentence for the user.
let revoked = activate("403 Forbidden", r#"{"error":"revoked"}"#)
.await
.expect_err("a revoked key fails");
assert!(!revoked.is::<InviteRefusedError>());
}
#[tokio::test(start_paused = true)]
async fn a_connection_that_stays_silent_is_dropped_by_the_check() {
let db = storage::DB::in_memory();
let row = db.new_row().unwrap();
row.write_deduped_hostname("rt").unwrap();
row.mark_complete().unwrap();
let connected = Arc::new(AtomicBool::new(false));
let dials = Arc::new(AtomicUsize::new(0));
let node = GuestNode::new(
"rt".into(),
row,
Box::new(Silent {
connected: connected.clone(),
dials: dials.clone(),
far_ends: Mutex::new(Vec::new()),
}),
Arc::new(NoObserver),
);
// Never in use: nothing to probe, no dial.
node.check_connection().await;
assert_eq!(dials.load(Ordering::SeqCst), 0);
// A cached connection that never answers is dropped by the check
// and, since the share was in use, redialled at once.
drop(node.open_stream().await.ok().expect("a stream"));
assert_eq!(dials.load(Ordering::SeqCst), 1);
node.check_connection().await;
assert_eq!(dials.load(Ordering::SeqCst), 2);
assert!(connected.load(Ordering::SeqCst));
// Dropped by the app (a network change) and checked on resume:
// redialled without a probe.
node.drop_connection();
node.check_connection().await;
assert_eq!(dials.load(Ordering::SeqCst), 3);
// Long unused: left for the next request to dial.
node.drop_connection();
tokio::time::advance(RECENT_USE + std::time::Duration::from_secs(1)).await;
node.check_connection().await;
assert_eq!(dials.load(Ordering::SeqCst), 3);
}
}