mod auth;
mod blocking;
mod stream;
mod tls;
pub(crate) use auth::HELLO as AUTH_HELLO;
pub(crate) use blocking::{Conn, ConnWriter, Raw};
pub(crate) use stream::{TlsRead, TlsStream, TlsWrite};
use rustls::pki_types::{CertificateDer, PrivateKeyDer};
use std::io;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
pub(crate) const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);
#[derive(Clone, Debug, Default)]
pub struct Security {
pub tls: Option<Tls>,
pub token: Option<Token>,
}
impl Security {
pub fn is_open(&self) -> bool {
self.tls.is_none() && self.token.is_none()
}
pub fn from_options(options: &str) -> io::Result<Self> {
let mut ca = None;
let mut cert = None;
let mut key = None;
let mut name = None;
let mut token = None;
for pair in options.split('&').filter(|p| !p.is_empty()) {
let (k, v) = pair.split_once('=').unwrap_or((pair, ""));
if v.is_empty() {
return Err(invalid(format!("connection option '{k}' needs a value")));
}
let slot = match k {
"ca" => &mut ca,
"cert" => &mut cert,
"key" => &mut key,
"name" => &mut name,
"token-file" => &mut token,
_ => {
return Err(invalid(format!(
"unknown connection option '{k}' (expected ca, cert, key, name, token-file or max-shards)"
)));
}
};
if slot.replace(v.to_owned()).is_some() {
return Err(invalid(format!("connection option '{k}' is given twice")));
}
}
let tls = match (&ca, &cert, &key) {
(None, None, None) if name.is_some() => {
return Err(invalid("name= applies to TLS, which needs ca="));
}
(None, None, None) => None,
(None, _, _) => {
return Err(invalid(
"TLS needs ca=PATH to verify the server; cert= and key= alone are not enough",
));
}
_ => {
let tls = Tls::from_pem_files(
cert.as_deref().map(Path::new),
key.as_deref().map(Path::new),
ca.as_deref().map(Path::new),
)?;
Some(match name {
Some(n) => tls.with_server_name(n),
None => tls,
})
}
};
Ok(Self {
tls,
token: token.map(Token::from_file).transpose()?,
})
}
}
#[derive(Clone)]
pub struct Tls {
certs: Vec<CertificateDer<'static>>,
key: Option<Arc<PrivateKeyDer<'static>>>,
roots: Option<Arc<rustls::RootCertStore>>,
server_name: Option<String>,
}
impl std::fmt::Debug for Tls {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Tls")
.field("certificates", &self.certs.len())
.field("key", &self.key.is_some())
.field("ca", &self.roots.as_ref().map(|r| r.len()))
.field("server_name", &self.server_name)
.finish()
}
}
impl Tls {
pub fn from_pem_files(
cert: Option<&Path>,
key: Option<&Path>,
ca: Option<&Path>,
) -> io::Result<Self> {
let read = |path: Option<&Path>| -> io::Result<Option<(PathBuf, Vec<u8>)>> {
path.map(|p| {
std::fs::read(p)
.map(|bytes| (p.to_owned(), bytes))
.map_err(|e| io::Error::new(e.kind(), format!("{}: {e}", p.display())))
})
.transpose()
};
let (cert, key, ca) = (read(cert)?, read(key)?, read(ca)?);
tls::load(
cert.as_ref().map(|(p, b)| (p.as_path(), b.as_slice())),
key.as_ref().map(|(p, b)| (p.as_path(), b.as_slice())),
ca.as_ref().map(|(p, b)| (p.as_path(), b.as_slice())),
)
}
pub fn from_pem(
cert: Option<&[u8]>,
key: Option<&[u8]>,
ca: Option<&[u8]>,
) -> io::Result<Self> {
let unnamed = Path::new("PEM");
tls::load(
cert.map(|b| (unnamed, b)),
key.map(|b| (unnamed, b)),
ca.map(|b| (unnamed, b)),
)
}
pub fn with_server_name(mut self, name: impl Into<String>) -> Self {
self.server_name = Some(name.into());
self
}
fn has_identity(&self) -> bool {
self.key.is_some()
}
pub(crate) fn has_identity_pub(&self) -> bool {
self.has_identity()
}
pub(crate) fn has_roots(&self) -> bool {
self.roots.is_some()
}
pub(crate) fn with_roots_of(mut self, other: &Tls) -> Self {
self.roots = other.roots.clone();
self
}
}
pub(crate) fn https_client(tls: &Tls) -> io::Result<Arc<rustls::ClientConfig>> {
let mut c = (*tls::client_config(tls)?).clone();
c.alpn_protocols = vec![b"http/1.1".to_vec()];
Ok(Arc::new(c))
}
pub(crate) fn https_server(tls: &Tls) -> io::Result<Arc<rustls::ServerConfig>> {
let mut c = (*tls::server_config(tls)?).clone();
c.alpn_protocols = vec![b"http/1.1".to_vec()];
Ok(Arc::new(c))
}
#[derive(Clone)]
pub struct Token(Arc<ring::hmac::Key>);
impl std::fmt::Debug for Token {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("Token(redacted)")
}
}
impl Token {
pub const MIN_LEN: usize = 16;
pub fn new(secret: &[u8]) -> io::Result<Self> {
if secret.len() < Self::MIN_LEN {
return Err(invalid(format!(
"a token needs at least {} bytes; use 32 random bytes",
Self::MIN_LEN
)));
}
Ok(Self(Arc::new(ring::hmac::Key::new(
ring::hmac::HMAC_SHA256,
secret,
))))
}
pub fn from_file(path: impl AsRef<Path>) -> io::Result<Self> {
let path = path.as_ref();
let bytes = std::fs::read(path)
.map_err(|e| io::Error::new(e.kind(), format!("{}: {e}", path.display())))?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
if let Ok(meta) = std::fs::metadata(path)
&& meta.permissions().mode() & 0o077 != 0
{
eprintln!(
"fusee: warning: token file {} is readable by other users; chmod 600 it",
path.display()
);
}
}
Self::new(bytes.trim_ascii())
.map_err(|e| io::Error::new(e.kind(), format!("{}: {e}", path.display())))
}
fn key(&self) -> &ring::hmac::Key {
&self.0
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum Purpose {
Server = 1,
Replica = 2,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Scope {
Worker,
Client,
Read,
}
pub(crate) const READ_CALLS: &[&str] = &[
"list",
"describe",
"query",
"result",
"streamRead",
"types",
"metrics",
"info",
"limits",
"export",
"invocations",
];
pub(crate) const NO_SENDER: u8 = u8::MAX;
pub(crate) struct Acceptor {
tls: Option<Arc<rustls::ServerConfig>>,
token: Option<Token>,
scoped: Vec<(Token, Scope)>,
purpose: Purpose,
}
pub(crate) enum Secured<S> {
Plain(S),
Tls(TlsStream),
}
enum Established {
Plain(tokio::net::TcpStream),
Tls(tokio::net::TcpStream, Box<rustls::Connection>),
}
enum Detached {
Plain(std::net::TcpStream),
Tls(std::net::TcpStream, Box<rustls::Connection>),
}
pub(crate) struct Handshaken {
established: Detached,
deadline: tokio::time::Instant,
}
impl Acceptor {
pub(crate) fn new(security: &Security, purpose: Purpose) -> io::Result<Self> {
let tls = security
.tls
.as_ref()
.map(|t| {
if purpose == Purpose::Replica && t.roots.is_none() {
return Err(invalid(
"replica TLS needs a CA that signs every member's certificate",
));
}
tls::server_config(t)
})
.transpose()?;
Ok(Self {
tls,
token: security.token.clone(),
scoped: Vec::new(),
purpose,
})
}
pub(crate) fn with_scoped(mut self, scoped: Vec<(Token, Scope)>) -> io::Result<Self> {
if !scoped.is_empty() && self.token.is_none() {
return Err(invalid(
"client and read tokens need the worker token too (--token-file): \
without it, every connection is a worker's",
));
}
self.scoped = scoped;
Ok(self)
}
pub(crate) fn open(purpose: Purpose) -> Self {
Self {
tls: None,
token: None,
scoped: Vec::new(),
purpose,
}
}
pub(crate) fn is_open(&self) -> bool {
self.tls.is_none() && self.token.is_none()
}
pub(crate) fn has_tls(&self) -> bool {
self.tls.is_some()
}
pub(crate) fn has_token(&self) -> bool {
self.token.is_some()
}
pub(crate) async fn accept_tcp(
&self,
sock: tokio::net::TcpStream,
) -> io::Result<(Secured<tokio::net::TcpStream>, Option<u8>)> {
let work = async {
let established = self.establish(sock).await?;
self.authenticate(established).await
};
tokio::time::timeout(HANDSHAKE_TIMEOUT, work)
.await
.map_err(|_| timed_out())?
}
pub(crate) async fn handshake_tls(
&self,
sock: tokio::net::TcpStream,
) -> io::Result<Handshaken> {
let deadline = tokio::time::Instant::now() + HANDSHAKE_TIMEOUT;
let established = tokio::time::timeout_at(deadline, self.establish(sock))
.await
.map_err(|_| timed_out())??;
let established = match established {
Established::Plain(s) => Detached::Plain(s.into_std()?),
Established::Tls(s, conn) => Detached::Tls(s.into_std()?, conn),
};
Ok(Handshaken {
established,
deadline,
})
}
pub(crate) async fn finish(
&self,
handshaken: Handshaken,
) -> io::Result<(Secured<tokio::net::TcpStream>, Option<u8>)> {
let Handshaken {
established,
deadline,
} = handshaken;
let established = match established {
Detached::Plain(s) => Established::Plain(tokio::net::TcpStream::from_std(s)?),
Detached::Tls(s, conn) => Established::Tls(tokio::net::TcpStream::from_std(s)?, conn),
};
tokio::time::timeout_at(deadline, self.authenticate(established))
.await
.map_err(|_| timed_out())?
}
async fn establish(&self, sock: tokio::net::TcpStream) -> io::Result<Established> {
let Some(config) = &self.tls else {
return Ok(Established::Plain(sock));
};
let conn = rustls::ServerConnection::new(config.clone())
.map_err(|e| tls::error("TLS setup", e))?;
let (sock, conn) = TlsStream::handshake_only(sock, conn.into()).await?;
Ok(Established::Tls(sock, Box::new(conn)))
}
async fn authenticate(
&self,
established: Established,
) -> io::Result<(Secured<tokio::net::TcpStream>, Option<u8>)> {
match established {
Established::Plain(mut sock) => {
let sender = self.respond(&mut sock, &[]).await?;
Ok((Secured::Plain(sock), sender))
}
Established::Tls(sock, conn) => {
let mut stream = TlsStream::wrap(sock, *conn);
let binding = stream.binding()?;
let sender = self.respond(&mut stream, &binding).await?;
Ok((Secured::Tls(stream), sender))
}
}
}
async fn respond<S>(&self, stream: &mut S, binding: &[u8]) -> io::Result<Option<u8>>
where
S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
self.respond_scoped(stream, binding).await.map(|(s, _)| s)
}
async fn respond_scoped<S>(
&self,
stream: &mut S,
binding: &[u8],
) -> io::Result<(Option<u8>, Scope)>
where
S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
let Some(token) = &self.token else {
return Ok((None, Scope::Worker));
};
let mut tokens = vec![token];
tokens.extend(self.scoped.iter().map(|(t, _)| t));
let (sender, i) = auth::respond_any(stream, &tokens, self.purpose, binding).await?;
let scope = i.checked_sub(1).map_or(Scope::Worker, |i| self.scoped[i].1);
Ok(((self.purpose == Purpose::Replica).then_some(sender), scope))
}
pub(crate) async fn accept_tcp_scoped(
&self,
sock: tokio::net::TcpStream,
) -> io::Result<(Secured<tokio::net::TcpStream>, Scope)> {
let work = async {
match self.establish(sock).await? {
Established::Plain(mut sock) => {
let (_, scope) = self.respond_scoped(&mut sock, &[]).await?;
Ok((Secured::Plain(sock), scope))
}
Established::Tls(sock, conn) => {
let mut stream = TlsStream::wrap(sock, *conn);
let binding = stream.binding()?;
let (_, scope) = self.respond_scoped(&mut stream, &binding).await?;
Ok((Secured::Tls(stream), scope))
}
}
};
tokio::time::timeout(HANDSHAKE_TIMEOUT, work)
.await
.map_err(|_| timed_out())?
}
pub(crate) async fn accept_unix_scoped(
&self,
mut sock: tokio::net::UnixStream,
) -> io::Result<(tokio::net::UnixStream, Scope)> {
let (_, scope) =
tokio::time::timeout(HANDSHAKE_TIMEOUT, self.respond_scoped(&mut sock, &[]))
.await
.map_err(|_| timed_out())??;
Ok((sock, scope))
}
}
#[derive(Clone)]
pub(crate) struct Connector {
tls: Option<Arc<rustls::ClientConfig>>,
server_name: Option<String>,
token: Option<Token>,
purpose: Purpose,
sender: u8,
}
impl std::fmt::Debug for Connector {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Connector")
.field("tls", &self.tls.is_some())
.field("token", &self.token.is_some())
.finish()
}
}
impl Connector {
pub(crate) fn new(security: &Security, purpose: Purpose, sender: u8) -> io::Result<Self> {
let tls = security
.tls
.as_ref()
.map(|t| {
if purpose == Purpose::Replica && !t.has_identity() {
return Err(invalid(
"replica TLS needs this member's certificate and key",
));
}
tls::client_config(t)
})
.transpose()?;
Ok(Self {
tls,
server_name: security.tls.as_ref().and_then(|t| t.server_name.clone()),
token: security.token.clone(),
purpose,
sender,
})
}
pub(crate) fn is_open(&self) -> bool {
self.tls.is_none() && self.token.is_none()
}
fn name(&self, address: &str) -> io::Result<rustls::pki_types::ServerName<'static>> {
let host = match &self.server_name {
Some(name) => name.as_str(),
None => host_of(address),
};
rustls::pki_types::ServerName::try_from(host.to_owned())
.map_err(|_| invalid(format!("'{host}' is not a valid TLS server name")))
}
pub(crate) fn secure(&self, mut raw: Raw, address: &str) -> io::Result<Conn> {
let mut conn = match (&self.tls, &raw) {
(None, _) => Conn::plain(raw)?,
(Some(_), Raw::Unix(_)) => {
return Err(invalid(
"TLS is not used over Unix sockets; drop ca=, cert= and key=",
));
}
(Some(config), Raw::Tcp(_)) => {
let conn =
blocking::client_handshake(&mut raw, config.clone(), self.name(address)?)
.map_err(|e| context(e, &format!("TLS with {address}")))?;
Conn::tls(raw, conn)?
}
};
if let Some(token) = &self.token {
let binding = conn.binding()?;
auth::initiate_blocking(&mut conn, token, self.purpose, self.sender, &binding)?;
}
Ok(conn)
}
pub(crate) async fn secure_tcp(
&self,
sock: tokio::net::TcpStream,
address: &str,
) -> io::Result<Secured<tokio::net::TcpStream>> {
let work = async {
let mut secured = match &self.tls {
None => Secured::Plain(sock),
Some(config) => {
let conn = rustls::ClientConnection::new(config.clone(), self.name(address)?)
.map_err(|e| tls::error("TLS setup", e))?;
Secured::Tls(TlsStream::handshake(sock, conn.into()).await?)
}
};
if let Some(token) = &self.token {
match &mut secured {
Secured::Plain(s) => {
auth::initiate(s, token, self.purpose, self.sender, &[]).await?
}
Secured::Tls(s) => {
let binding = s.binding()?;
auth::initiate(s, token, self.purpose, self.sender, &binding).await?
}
}
}
Ok(secured)
};
tokio::time::timeout(HANDSHAKE_TIMEOUT, work)
.await
.map_err(|_| timed_out())?
}
}
pub(crate) fn members_named(cert: &CertificateDer<'_>, peers: &[String]) -> u32 {
let Ok(parsed) = rustls::server::ParsedCertificate::try_from(cert) else {
return 0;
};
let mut mask = 0;
for (i, peer) in peers.iter().enumerate().take(32) {
let named = rustls::pki_types::ServerName::try_from(host_of(peer))
.is_ok_and(|name| rustls::client::verify_server_name(&parsed, &name).is_ok());
if named {
mask |= 1 << i;
}
}
mask
}
pub(crate) fn host_of(address: &str) -> &str {
let host = address.rsplit_once(':').map_or(address, |(h, _)| h);
host.strip_prefix('[')
.and_then(|h| h.strip_suffix(']'))
.unwrap_or(host)
}
pub(crate) fn warn_if_exposed(what: &str, addr: SocketAddr, tls: bool, token: bool) {
if addr.ip().is_loopback() || tls {
return;
}
if token {
eprintln!(
"fusee {what}: warning: {addr} authenticates connections with a token but has no TLS; traffic is readable and can be altered on the network path"
);
} else {
eprintln!(
"fusee {what}: warning: {addr} has neither TLS nor a token; anyone who can reach it can read and change workflow state"
);
}
}
pub(crate) fn refused(what: std::fmt::Arguments<'_>, e: &io::Error) {
static LAST: AtomicU64 = AtomicU64::new(0);
let now = fusee::local::monotonic_us() / 1000;
let last = LAST.load(Ordering::Relaxed);
if (last == 0 || now.saturating_sub(last) >= 10_000)
&& LAST
.compare_exchange(last, now.max(1), Ordering::Relaxed, Ordering::Relaxed)
.is_ok()
{
eprintln!("fusee {what}: {e}");
}
}
fn invalid(message: impl Into<String>) -> io::Error {
io::Error::new(io::ErrorKind::InvalidInput, message.into())
}
fn timed_out() -> io::Error {
io::Error::new(
io::ErrorKind::TimedOut,
"the connection did not finish TLS and authentication in time",
)
}
fn context(e: io::Error, what: &str) -> io::Error {
io::Error::new(e.kind(), format!("{what}: {e}"))
}
#[cfg(test)]
mod tests;