use super::{
config::{Place, Seat},
handshake::Session,
tools::Identity,
};
use crate::door::assistant_network::{OPCODE_BINARY, OPCODE_CLOSE, accept, encode};
use serde_json::{Value, json};
use sha2::{Digest as _, Sha256};
use tokio::io::{
AsyncBufRead, AsyncBufReadExt as _, AsyncRead, AsyncReadExt as _, AsyncWrite,
AsyncWriteExt as _, BufReader,
};
use tokio::net::UnixStream;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
const HEAD_BYTES: usize = 16_384;
type Reader = Box<dyn AsyncRead + Unpin + Send>;
type Writer = Box<dyn AsyncWrite + Unpin + Send>;
enum Wire {
Lines,
Messages(Masks),
}
enum Pending {
Frame(Value),
Flush(tokio::sync::oneshot::Sender<()>),
}
#[derive(Clone)]
pub(super) struct Output {
sender: mpsc::Sender<Pending>,
bound: usize,
}
impl Output {
pub(super) async fn send_wait(&self, value: Value) -> anyhow::Result<()> {
anyhow::ensure!(
serde_json::to_vec(&value)?.len().saturating_add(1) <= self.bound,
"door_frame_bytes_exceeded"
);
self.sender
.send(Pending::Frame(value))
.await
.map_err(|_| anyhow::anyhow!("door_writer_closed"))
}
pub async fn flush(&self) -> anyhow::Result<()> {
let (sent, received) = tokio::sync::oneshot::channel();
self.sender
.send(Pending::Flush(sent))
.await
.map_err(|_| anyhow::anyhow!("door_writer_closed"))?;
received
.await
.map_err(|_| anyhow::anyhow!("door_flush_failed"))
}
pub fn send(&self, value: Value) -> anyhow::Result<()> {
anyhow::ensure!(
serde_json::to_vec(&value)?.len().saturating_add(1) <= self.bound,
"door_frame_bytes_exceeded"
);
match self.sender.try_send(Pending::Frame(value)) {
Ok(()) => Ok(()),
Err(mpsc::error::TrySendError::Full(frame)) => self
.sender
.blocking_send(frame)
.map_err(|_| anyhow::anyhow!("door_writer_closed")),
Err(mpsc::error::TrySendError::Closed(_)) => anyhow::bail!("door_writer_closed"),
}
}
}
pub(super) struct Link {
pub person: String,
pub input: mpsc::Receiver<anyhow::Result<Option<Value>>>,
pub output: Output,
pub reader: JoinHandle<()>,
pub writer: JoinHandle<anyhow::Result<()>>,
}
impl Link {
pub async fn connect(
seat: &Seat,
identity: &Identity,
session: &Session,
) -> anyhow::Result<Self> {
let trust = match &seat.place {
Place::Local { .. } => None,
Place::Network { .. } => {
Some(crate::client::network::Trust::system().map_err(anyhow::Error::msg)?)
}
};
Self::connect_with(seat, identity, session, trust.as_ref()).await
}
pub async fn connect_with(
seat: &Seat,
identity: &Identity,
session: &Session,
trust: Option<&crate::client::network::Trust>,
) -> anyhow::Result<Self> {
let (reader, mut writer, wire): (BufReader<Reader>, Writer, Wire) = match &seat.place {
Place::Local { door_socket, .. } => {
let (reader, writer) = local(door_socket).await?;
(BufReader::new(reader), writer, Wire::Lines)
}
Place::Network { door_url, .. } => {
let trust = trust.ok_or_else(|| anyhow::anyhow!("assistant_door_trust_absent"))?;
let timeout = std::time::Duration::from_millis(seat.harness.session_timeout_ms);
let (reader, writer, masks) = network(door_url, trust, timeout).await?;
(reader, writer, Wire::Messages(masks))
}
};
let bound = seat.harness.acp_frame_bytes;
let (outgoing, mut pending) = mpsc::channel::<Pending>(seat.harness.write_queue);
let messages = matches!(wire, Wire::Messages(_));
let writer = tokio::spawn(async move {
let mut wire = wire;
while let Some(value) = pending.recv().await {
let value = match value {
Pending::Frame(value) => value,
Pending::Flush(receipt) => {
writer.flush().await?;
receipt
.send(())
.map_err(|()| anyhow::anyhow!("door_flush_receiver_closed"))?;
continue;
}
};
let mut bytes = serde_json::to_vec(&value)?;
match &mut wire {
Wire::Lines => bytes.push(b'\n'),
Wire::Messages(masks) => {
bytes = encode(OPCODE_BINARY, &bytes, Some(masks.mask()))?;
}
}
writer.write_all(&bytes).await?;
writer.flush().await?;
}
Ok(())
});
let (events, input) = mpsc::channel(1);
let reader = tokio::spawn(async move {
let mut reader = reader;
loop {
let value = if messages {
read_message(&mut reader, bound).await
} else {
read(&mut reader, bound).await
};
let ended = !matches!(value, Ok(Some(_)));
if events.send(value).await.is_err() || ended {
return;
}
}
});
let link = Self {
person: String::new(),
input,
output: Output {
sender: outgoing,
bound,
},
reader,
writer,
};
link.output.send_wait(json!({"hello":{"agent":identity.agent,"pass":identity.pass,"store":seat.store,
"resumable":session.resumable,"harness":{"id":seat.harness.id,"protocol":session.version.number(),
"version":session.info["version"].as_str().unwrap_or("not-advertised")}}})).await?;
Ok(link)
}
#[cfg(test)]
pub async fn open(seat: &Seat, identity: &Identity, session: &Session) -> anyhow::Result<Self> {
let mut link = Self::connect(seat, identity, session).await?;
link.welcome(seat, identity).await?;
Ok(link)
}
pub async fn welcome(&mut self, seat: &Seat, identity: &Identity) -> anyhow::Result<()> {
let welcome = self
.input
.recv()
.await
.ok_or_else(|| anyhow::anyhow!("assistant_door_closed"))??
.ok_or_else(|| anyhow::anyhow!("assistant_door_closed"))?;
if let Some(refused) = welcome.get("refused") {
let code = refused["code"]
.as_str()
.filter(|code| {
!code.is_empty()
&& code.len() <= 96
&& !code.contains(&identity.pass)
&& code
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || byte == b'_')
})
.unwrap_or("unnamed_refusal");
anyhow::bail!("assistant_door_refused: {code}");
}
let person = welcome["welcome"]["person"]
.as_str()
.filter(|person| !person.is_empty() && person.len() <= seat.harness.content_bytes)
.ok_or_else(|| anyhow::anyhow!("assistant_door_welcome_invalid"))?;
person.clone_into(&mut self.person);
Ok(())
}
}
impl Drop for Link {
fn drop(&mut self) {
self.reader.abort();
self.writer.abort();
}
}
async fn local(door_socket: &std::path::Path) -> anyhow::Result<(Reader, Writer)> {
use std::os::unix::fs::{FileTypeExt as _, MetadataExt as _, PermissionsExt as _};
let metadata = std::fs::symlink_metadata(door_socket)?;
anyhow::ensure!(
metadata.file_type().is_socket()
&& metadata.permissions().mode() & 0o777 == 0o600
&& metadata.uid() == rustix::process::geteuid().as_raw(),
"assistant_socket_requires_owner_and_mode_0600"
);
let stream = UnixStream::connect(door_socket).await?;
anyhow::ensure!(
stream.peer_cred()?.uid() == rustix::process::geteuid().as_raw(),
"assistant_socket_owner_differs"
);
let (reader, writer) = stream.into_split();
Ok((Box::new(reader), Box::new(writer)))
}
struct Masks {
seed: [u8; 32],
next: u64,
}
impl Masks {
fn new() -> anyhow::Result<Self> {
use std::io::Read as _;
let mut seed = [0_u8; 32];
std::fs::File::open("/dev/urandom")
.and_then(|mut source| source.read_exact(&mut seed))
.map_err(|_| anyhow::anyhow!("assistant_random_unavailable"))?;
Ok(Self { seed, next: 0 })
}
fn take(&mut self) -> [u8; 32] {
let digest = Sha256::new()
.chain_update(self.seed)
.chain_update(self.next.to_be_bytes())
.finalize();
self.next = self.next.wrapping_add(1);
let mut bytes = [0_u8; 32];
bytes.copy_from_slice(&digest);
bytes
}
fn mask(&mut self) -> [u8; 4] {
let bytes = self.take();
[bytes[0], bytes[1], bytes[2], bytes[3]]
}
fn key(&mut self) -> String {
use base64::Engine as _;
base64::engine::general_purpose::STANDARD.encode(&self.take()[..16])
}
}
async fn network(
door_url: &url::Url,
trust: &crate::client::network::Trust,
timeout: std::time::Duration,
) -> anyhow::Result<(BufReader<Reader>, Writer, Masks)> {
let host = door_url
.host_str()
.ok_or_else(|| anyhow::anyhow!("seat_door_url_invalid"))?;
let port = door_url
.port_or_known_default()
.ok_or_else(|| anyhow::anyhow!("seat_door_url_invalid"))?;
let bare = host.trim_start_matches('[').trim_end_matches(']');
let name =
crate::client::network::name(bare).map_err(|_| anyhow::anyhow!("seat_door_url_invalid"))?;
let connector = tokio_rustls::TlsConnector::from(trust.client_config());
let reached = tokio::time::timeout(timeout, async {
let tcp = tokio::net::TcpStream::connect((bare, port)).await?;
connector.connect(name, tcp).await
})
.await
.map_err(|_| anyhow::anyhow!("assistant_door_unreached: timed out"))?
.map_err(|error| anyhow::anyhow!("assistant_door_unreached: {}", error.kind()))?;
let (reader, mut writer) = tokio::io::split(reached);
let mut masks = Masks::new()?;
let key = masks.key();
let origin = door_url.origin().ascii_serialization();
let authority = &door_url[url::Position::BeforeHost..url::Position::AfterPort];
writer
.write_all(
format!(
"GET /assistant HTTP/1.1\r\nHost: {authority}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\nOrigin: {origin}\r\n\r\n"
)
.as_bytes(),
)
.await?;
writer.flush().await?;
let mut reader: BufReader<Reader> = BufReader::new(Box::new(reader));
tokio::time::timeout(timeout, upgraded(&mut reader, &accept(&key)))
.await
.map_err(|_| anyhow::anyhow!("assistant_door_upgrade_unanswered"))??;
Ok((reader, Box::new(writer), masks))
}
async fn upgraded(reader: &mut BufReader<Reader>, expected: &str) -> anyhow::Result<()> {
let mut head = Vec::new();
while !head.ends_with(b"\r\n\r\n") {
anyhow::ensure!(
head.len() < HEAD_BYTES,
"assistant_door_upgrade_head_too_long"
);
head.push(reader.read_u8().await?);
}
let head = String::from_utf8(head)
.map_err(|_| anyhow::anyhow!("assistant_door_upgrade_head_invalid"))?;
let mut lines = head.split("\r\n");
let status = lines
.next()
.and_then(|line| line.split(' ').nth(1))
.and_then(|code| code.parse::<u16>().ok())
.ok_or_else(|| anyhow::anyhow!("assistant_door_upgrade_head_invalid"))?;
let headers: Vec<(String, &str)> = lines
.filter_map(|line| line.split_once(':'))
.map(|(name, value)| (name.trim().to_ascii_lowercase(), value.trim()))
.collect();
let header = |name: &str| {
headers
.iter()
.find(|(named, _)| named == name)
.map(|(_, value)| *value)
};
if status == 101 {
anyhow::ensure!(
header("upgrade").is_some_and(|value| value.eq_ignore_ascii_case("websocket"))
&& header("sec-websocket-accept") == Some(expected),
"assistant_door_upgrade_invalid"
);
return Ok(());
}
let length = header("content-length")
.and_then(|value| value.parse::<usize>().ok())
.filter(|length| *length <= HEAD_BYTES)
.unwrap_or(0);
let mut body = vec![0_u8; length];
reader.read_exact(&mut body).await?;
let code = serde_json::from_slice::<Value>(&body)
.ok()
.and_then(|value| value["code"].as_str().map(str::to_owned))
.filter(|code| {
!code.is_empty()
&& code.len() <= 96
&& code
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || byte == b'_')
});
match code {
Some(code) => anyhow::bail!("assistant_door_refused: {code}"),
None => anyhow::bail!("assistant_door_upgrade_refused: status {status}"),
}
}
async fn read_message(
reader: &mut BufReader<Reader>,
bound: usize,
) -> anyhow::Result<Option<Value>> {
let mut head = [0_u8; 2];
if reader.read(&mut head[..1]).await? == 0 {
return Ok(None);
}
reader.read_exact(&mut head[1..]).await?;
anyhow::ensure!(
head[0] & 0x70 == 0 && head[1] & 0x80 == 0,
"door_frame_invalid"
);
let length = match head[1] & 0x7F {
126 => u64::from(reader.read_u16().await?),
127 => reader.read_u64().await?,
short => u64::from(short),
};
let length = usize::try_from(length)
.ok()
.filter(|length| *length < bound)
.ok_or_else(|| anyhow::anyhow!("door_frame_bytes_exceeded"))?;
let mut payload = vec![0_u8; length];
reader.read_exact(&mut payload).await?;
match (head[0] & 0x80 != 0, head[0] & 0x0F) {
(_, OPCODE_CLOSE) => Ok(None),
(true, OPCODE_BINARY) => {
let value: Value = serde_json::from_slice(&payload)
.map_err(|_| anyhow::anyhow!("door_frame_invalid_json"))?;
anyhow::ensure!(value.is_object(), "door_frame_requires_object");
Ok(Some(value))
}
_ => anyhow::bail!("door_frame_invalid"),
}
}
async fn read<R: AsyncBufRead + Unpin>(
reader: &mut R,
bound: usize,
) -> anyhow::Result<Option<Value>> {
let mut line = Vec::new();
loop {
let buffer = reader.fill_buf().await?;
if buffer.is_empty() {
anyhow::ensure!(line.is_empty(), "door_frame_missing_newline");
return Ok(None);
}
let count = buffer
.iter()
.position(|byte| *byte == b'\n')
.map_or(buffer.len(), |index| index + 1);
anyhow::ensure!(
count <= bound.saturating_sub(line.len()),
"door_frame_bytes_exceeded"
);
let ended = buffer[count - 1] == b'\n';
line.extend_from_slice(&buffer[..count]);
reader.consume(count);
if ended {
break;
}
}
let value: Value =
serde_json::from_slice(&line).map_err(|_| anyhow::anyhow!("door_frame_invalid_json"))?;
anyhow::ensure!(value.is_object(), "door_frame_requires_object");
Ok(Some(value))
}
#[cfg(test)]
mod queue_tests {
use super::{Output, Pending};
use serde_json::json;
use std::{future::Future as _, task::Poll};
#[tokio::test]
async fn round4_door_queue_waits_for_capacity_and_names_a_closed_writer() -> anyhow::Result<()>
{
let (sender, mut received) = tokio::sync::mpsc::channel(1);
let output = Output {
sender,
bound: 4096,
};
output.send_wait(json!({"item":1})).await?;
let waiting = output.send_wait(json!({"item":2}));
tokio::pin!(waiting);
let pending =
std::future::poll_fn(|cx| Poll::Ready(waiting.as_mut().poll(cx).is_pending())).await;
assert!(pending, "full door queue did not wait");
assert!(matches!(received.recv().await,Some(Pending::Frame(value)) if value["item"]==1));
waiting.await?;
assert!(matches!(received.recv().await,Some(Pending::Frame(value)) if value["item"]==2));
drop(received);
let error = output
.send_wait(json!({"item":3}))
.await
.err()
.ok_or_else(|| anyhow::anyhow!("closed writer accepted frame"))?;
assert_eq!(error.to_string(), "door_writer_closed");
Ok(())
}
}