Start9/start-os
2
0
Fork 0
mirror of https://github.com/Start9Labs/start-os.git synced 2026-03-26 02:11:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor/networking (#2189)

Browse Source 
* refactor networking and account

* add interfaces from manifest automatically

* use nistp256 to satisfy firefox

* use ed25519 if available

* fix ip signing

* fix SQL error

* update prettytable to fix segfault

* fix migration

* fix migration

* bump welcome-ack

* add redirect if connecting to https over http

* misc rebase fixes

* fix compression

* bump rustc version
This commit is contained in:
Aiden McClelland
2023-03-08 19:30:46 -07:00
committed by GitHub
parent da55d6f7cd
commit bbb9980941
79 changed files with 3577 additions and 3587 deletions
Download Patch File Download Diff File Expand all files Collapse all files

215
backend/src/net/cert_resolver.rs
View File

@@ -1,215 +0,0 @@
use std::collections::BTreeMap;
use std::io;
use std::pin::Pin;
use std::str::FromStr;
use std::sync::{Arc, RwLock};
use std::task::{Context, Poll};
use color_eyre::eyre::eyre;
use futures::{ready, Future};
use hyper::server::accept::Accept;
use hyper::server::conn::{AddrIncoming, AddrStream};
use openssl::pkey::{PKey, Private};
use openssl::x509::X509;
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use tokio_rustls::rustls::server::ResolvesServerCert;
use tokio_rustls::rustls::sign::{any_supported_type, CertifiedKey};
use tokio_rustls::rustls::{Certificate, PrivateKey, ServerConfig};
use crate::net::net_utils::ResourceFqdn;
use crate::Error;
enum State {
Handshaking(tokio_rustls::Accept<AddrStream>),
Streaming(tokio_rustls::server::TlsStream<AddrStream>),
}
// tokio_rustls::server::TlsStream doesn't expose constructor methods,
// so we have to TlsAcceptor::accept and handshake to have access to it
// TlsStream implements AsyncRead/AsyncWrite handshaking tokio_rustls::Accept first
pub struct TlsStream {
state: State,
}
impl TlsStream {
fn new(stream: AddrStream, config: Arc<ServerConfig>) -> TlsStream {
let accept = tokio_rustls::TlsAcceptor::from(config).accept(stream);
TlsStream {
state: State::Handshaking(accept),
}
}
}
impl AsyncRead for TlsStream {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context,
buf: &mut ReadBuf,
) -> Poll<io::Result<()>> {
let pin = self.get_mut();
match pin.state {
State::Handshaking(ref mut accept) => match ready!(Pin::new(accept).poll(cx)) {
Ok(mut stream) => {
let result = Pin::new(&mut stream).poll_read(cx, buf);
pin.state = State::Streaming(stream);
result
}
Err(err) => Poll::Ready(Err(err)),
},
State::Streaming(ref mut stream) => Pin::new(stream).poll_read(cx, buf),
}
}
}
impl AsyncWrite for TlsStream {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<io::Result<usize>> {
let pin = self.get_mut();
match pin.state {
State::Handshaking(ref mut accept) => match ready!(Pin::new(accept).poll(cx)) {
Ok(mut stream) => {
let result = Pin::new(&mut stream).poll_write(cx, buf);
pin.state = State::Streaming(stream);
result
}
Err(err) => Poll::Ready(Err(err)),
},
State::Streaming(ref mut stream) => Pin::new(stream).poll_write(cx, buf),
}
}
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
match self.state {
State::Handshaking(_) => Poll::Ready(Ok(())),
State::Streaming(ref mut stream) => Pin::new(stream).poll_flush(cx),
}
}
fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
match self.state {
State::Handshaking(_) => Poll::Ready(Ok(())),
State::Streaming(ref mut stream) => Pin::new(stream).poll_shutdown(cx),
}
}
}
impl ResolvesServerCert for EmbassyCertResolver {
fn resolve(
&self,
client_hello: tokio_rustls::rustls::server::ClientHello,
) -> Option<Arc<tokio_rustls::rustls::sign::CertifiedKey>> {
let hostname_raw = client_hello.server_name();
match hostname_raw {
Some(hostname_str) => {
let full_fqdn = match ResourceFqdn::from_str(hostname_str) {
Ok(fqdn) => fqdn,
Err(_) => {
tracing::error!("Error converting {} to fqdn struct", hostname_str);
return None;
}
};
let lock = self.cert_mapping.read();
match lock {
Ok(lock) => lock
.get(&full_fqdn)
.map(|cert_key| Arc::new(cert_key.to_owned())),
Err(err) => {
tracing::error!("resolve fn Error: {}", err);
None
}
}
}
None => None,
}
}
}
#[derive(Clone, Default)]
pub struct EmbassyCertResolver {
cert_mapping: Arc<RwLock<BTreeMap<ResourceFqdn, CertifiedKey>>>,
}
impl EmbassyCertResolver {
pub fn new() -> Self {
Self::default()
}
pub async fn add_certificate_to_resolver(
&mut self,
service_resource_fqdn: ResourceFqdn,
package_cert_data: (PKey<Private>, Vec<X509>),
) -> Result<(), Error> {
let x509_cert_chain = package_cert_data.1;
let private_keys = package_cert_data
.0
.private_key_to_der()
.map_err(|err| Error::new(eyre!("{}", err), crate::ErrorKind::OpenSsl))?;
let mut full_rustls_certs = Vec::new();
for cert in x509_cert_chain.iter() {
let cert = Certificate(
cert.to_der()
.map_err(|err| Error::new(eyre!("{}", err), crate::ErrorKind::OpenSsl))?,
);
full_rustls_certs.push(cert);
}
let pre_sign_key = PrivateKey(private_keys);
let actual_sign_key = any_supported_type(&pre_sign_key)
.map_err(|err| Error::new(eyre!("{}", err), crate::ErrorKind::OpenSsl))?;
let cert_key = CertifiedKey::new(full_rustls_certs, actual_sign_key);
let mut lock = self
.cert_mapping
.write()
.map_err(|err| Error::new(eyre!("{}", err), crate::ErrorKind::Network))?;
lock.insert(service_resource_fqdn, cert_key);
Ok(())
}
pub async fn remove_cert(&mut self, hostname: ResourceFqdn) -> Result<(), Error> {
let mut lock = self
.cert_mapping
.write()
.map_err(|err| Error::new(eyre!("{}", err), crate::ErrorKind::Network))?;
lock.remove(&hostname);
Ok(())
}
}
pub struct TlsAcceptor {
config: Arc<ServerConfig>,
incoming: AddrIncoming,
}
impl TlsAcceptor {
pub fn new(config: Arc<ServerConfig>, incoming: AddrIncoming) -> TlsAcceptor {
TlsAcceptor { config, incoming }
}
}
impl Accept for TlsAcceptor {
type Conn = TlsStream;
type Error = io::Error;
fn poll_accept(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<Result<Self::Conn, Self::Error>>> {
let pin = self.get_mut();
match ready!(Pin::new(&mut pin.incoming).poll_accept(cx)) {
Some(Ok(sock)) => Poll::Ready(Some(Ok(TlsStream::new(sock, pin.config.clone())))),
Some(Err(e)) => Poll::Ready(Some(Err(e))),
None => Poll::Ready(None),
}
}
}

46
backend/src/net/dhcp.rs
View File

@@ -1,7 +1,9 @@
use std::collections::BTreeMap;
use std::collections::{BTreeMap, BTreeSet};
use std::net::IpAddr;
use futures::TryStreamExt;
use rpc_toolkit::command;
use tokio::sync::RwLock;
use crate::context::RpcContext;
use crate::db::model::IpInfo;
@@ -9,6 +11,32 @@ use crate::net::net_utils::{iface_is_physical, list_interfaces};
use crate::util::display_none;
use crate::Error;
lazy_static::lazy_static! {
static ref CACHED_IPS: RwLock<BTreeSet<IpAddr>> = RwLock::new(BTreeSet::new());
}
async fn _ips() -> Result<BTreeSet<IpAddr>, Error> {
Ok(init_ips()
.await?
.values()
.flat_map(|i| {
std::iter::empty()
.chain(i.ipv4.map(IpAddr::from))
.chain(i.ipv6.map(IpAddr::from))
})
.collect())
}
pub async fn ips() -> Result<BTreeSet<IpAddr>, Error> {
let ips = CACHED_IPS.read().await.clone();
if !ips.is_empty() {
return Ok(ips);
}
let ips = _ips().await?;
*CACHED_IPS.write().await = ips.clone();
Ok(ips)
}
pub async fn init_ips() -> Result<BTreeMap<String, IpInfo>, Error> {
let mut res = BTreeMap::new();
let mut ifaces = list_interfaces();
@@ -29,15 +57,23 @@ pub async fn dhcp() -> Result<(), Error> {
#[command(display(display_none))]
pub async fn update(#[context] ctx: RpcContext, #[arg] interface: String) -> Result<(), Error> {
if iface_is_physical(&interface).await {
let ip_info = IpInfo::for_interface(&interface).await?;
crate::db::DatabaseModel::new()
.server_info()
.ip_info()
.idx_model(&interface)
.put(
&mut ctx.db.handle(),
&IpInfo::for_interface(&interface).await?,
)
.put(&mut ctx.db.handle(), &ip_info)
.await?;
let mut cached = CACHED_IPS.write().await;
if cached.is_empty() {
*cached = _ips().await?;
} else {
cached.extend(
std::iter::empty()
.chain(ip_info.ipv4.map(IpAddr::from))
.chain(ip_info.ipv6.map(IpAddr::from)),
);
}
}
Ok(())
}

89
backend/src/net/dns.rs
View File

@@ -1,9 +1,10 @@
use std::borrow::Borrow;
use std::collections::BTreeMap;
use std::net::{Ipv4Addr, SocketAddr};
use std::sync::Arc;
use std::sync::{Arc, Weak};
use std::time::Duration;
use color_eyre::eyre::eyre;
use futures::TryFutureExt;
use helpers::NonDetachingJoinHandle;
use models::PackageId;
@@ -13,39 +14,52 @@ use tokio::sync::RwLock;
use trust_dns_server::authority::MessageResponseBuilder;
use trust_dns_server::client::op::{Header, ResponseCode};
use trust_dns_server::client::rr::{Name, Record, RecordType};
use trust_dns_server::proto::rr::rdata::a;
use trust_dns_server::server::{Request, RequestHandler, ResponseHandler, ResponseInfo};
use trust_dns_server::ServerFuture;
use crate::util::Invoke;
use crate::{Error, ErrorKind, ResultExt, HOST_IP};
use crate::{Error, ErrorKind, ResultExt};
pub struct DnsController {
services: Arc<RwLock<BTreeMap<PackageId, Vec<Ipv4Addr>>>>,
services: Weak<RwLock<BTreeMap<Option<PackageId>, BTreeMap<Ipv4Addr, Weak<()>>>>>,
#[allow(dead_code)]
dns_server: NonDetachingJoinHandle<Result<(), Error>>,
}
struct Resolver {
services: Arc<RwLock<BTreeMap<PackageId, Vec<Ipv4Addr>>>>,
services: Arc<RwLock<BTreeMap<Option<PackageId>, BTreeMap<Ipv4Addr, Weak<()>>>>>,
}
impl Resolver {
async fn resolve(&self, name: &Name) -> Option<Vec<Ipv4Addr>> {
match name.iter().next_back() {
Some(b"embassy") => {
if let Some(pkg) = name.iter().rev().skip(1).next() {
if let Some(ip) = self
.services
.read()
.await
.get(std::str::from_utf8(pkg).unwrap_or_default())
{
Some(ip.iter().copied().collect())
if let Some(ip) = self.services.read().await.get(&Some(
std::str::from_utf8(pkg)
.unwrap_or_default()
.parse()
.unwrap_or_default(),
)) {
Some(
ip.iter()
.filter(|(_, rc)| rc.strong_count() > 0)
.map(|(ip, _)| *ip)
.collect(),
)
} else {
None
}
} else {
Some(vec![HOST_IP.into()])
if let Some(ip) = self.services.read().await.get(&None) {
Some(
ip.iter()
.filter(|(_, rc)| rc.strong_count() > 0)
.map(|(ip, _)| *ip)
.collect(),
)
} else {
None
}
}
}
_ => None,
@@ -162,26 +176,47 @@ impl DnsController {
.into();
Ok(Self {
services,
services: Arc::downgrade(&services),
dns_server,
})
}
pub async fn add(&self, pkg_id: &PackageId, ip: Ipv4Addr) {
let mut writable = self.services.write().await;
let mut ips = writable.remove(pkg_id).unwrap_or_default();
ips.push(ip);
writable.insert(pkg_id.clone(), ips);
pub async fn add(&self, pkg_id: Option<PackageId>, ip: Ipv4Addr) -> Result<Arc<()>, Error> {
if let Some(services) = Weak::upgrade(&self.services) {
let mut writable = services.write().await;
let mut ips = writable.remove(&pkg_id).unwrap_or_default();
let rc = if let Some(rc) = Weak::upgrade(&ips.remove(&ip).unwrap_or_default()) {
rc
} else {
Arc::new(())
};
ips.insert(ip, Arc::downgrade(&rc));
writable.insert(pkg_id, ips);
Ok(rc)
} else {
Err(Error::new(
eyre!("DNS Server Thread has exited"),
crate::ErrorKind::Network,
))
}
}
pub async fn remove(&self, pkg_id: &PackageId, ip: Ipv4Addr) {
let mut writable = self.services.write().await;
let mut ips = writable.remove(pkg_id).unwrap_or_default();
if let Some((idx, _)) = ips.iter().copied().enumerate().find(|(_, x)| *x == ip) {
ips.swap_remove(idx);
}
if !ips.is_empty() {
writable.insert(pkg_id.clone(), ips);
pub async fn gc(&self, pkg_id: Option<PackageId>, ip: Ipv4Addr) -> Result<(), Error> {
if let Some(services) = Weak::upgrade(&self.services) {
let mut writable = services.write().await;
let mut ips = writable.remove(&pkg_id).unwrap_or_default();
if let Some(rc) = Weak::upgrade(&ips.remove(&ip).unwrap_or_default()) {
ips.insert(ip, Arc::downgrade(&rc));
}
if !ips.is_empty() {
writable.insert(pkg_id, ips);
}
Ok(())
} else {
Err(Error::new(
eyre!("DNS Server Thread has exited"),
crate::ErrorKind::Network,
))
}
}
}

173
backend/src/net/embassy_service_http_server.rs
View File

@@ -1,173 +0,0 @@
use std::collections::BTreeMap;
use std::net::{IpAddr, SocketAddr};
use std::sync::Arc;
use helpers::NonDetachingJoinHandle;
use http::StatusCode;
use hyper::server::conn::AddrIncoming;
use hyper::service::{make_service_fn, service_fn};
use hyper::{Body, Error as HyperError, Response, Server};
use tokio::sync::oneshot;
use tokio_rustls::rustls::ServerConfig;
use tracing::error;
use crate::net::cert_resolver::TlsAcceptor;
use crate::net::net_utils::{host_addr_fqdn, ResourceFqdn};
use crate::net::HttpHandler;
use crate::Error;
static RES_NOT_FOUND: &[u8] = b"503 Service Unavailable";
static NO_HOST: &[u8] = b"No host header found";
pub struct EmbassyServiceHTTPServer {
pub svc_mapping: Arc<tokio::sync::RwLock<BTreeMap<ResourceFqdn, HttpHandler>>>,
pub shutdown: oneshot::Sender<()>,
pub handle: NonDetachingJoinHandle<()>,
pub ssl_cfg: Option<Arc<ServerConfig>>,
}
impl EmbassyServiceHTTPServer {
pub async fn new(
listener_addr: IpAddr,
port: u16,
ssl_cfg: Option<Arc<ServerConfig>>,
) -> Result<Self, Error> {
let (tx, rx) = tokio::sync::oneshot::channel::<()>();
let listener_socket_addr = SocketAddr::from((listener_addr, port));
let server_service_mapping = Arc::new(tokio::sync::RwLock::new(BTreeMap::<
ResourceFqdn,
HttpHandler,
>::new()));
let server_service_mapping1 = server_service_mapping.clone();
let bare_make_service_fn = move || {
let server_service_mapping = server_service_mapping.clone();
async move {
Ok::<_, HyperError>(service_fn(move |req| {
let mut server_service_mapping = server_service_mapping.clone();
async move {
server_service_mapping = server_service_mapping.clone();
let host = host_addr_fqdn(&req);
match host {
Ok(host_uri) => {
let res = {
let mapping = server_service_mapping.read().await;
let opt_handler = mapping.get(&host_uri).cloned();
opt_handler
};
match res {
Some(opt_handler) => {
let response = opt_handler(req).await;
match response {
Ok(resp) => Ok::<Response<Body>, hyper::Error>(resp),
Err(err) => Ok(respond_hyper_error(err)),
}
}
None => Ok(res_not_found()),
}
}
Err(e) => Ok(no_host_found(e)),
}
}
}))
}
};
let inner_ssl_cfg = ssl_cfg.clone();
let handle = tokio::spawn(async move {
match inner_ssl_cfg {
Some(cfg) => {
let incoming = AddrIncoming::bind(&listener_socket_addr).unwrap();
let server = Server::builder(TlsAcceptor::new(cfg, incoming))
.http1_preserve_header_case(true)
.http1_title_case_headers(true)
.serve(make_service_fn(|_| bare_make_service_fn()))
.with_graceful_shutdown({
async {
rx.await.ok();
}
});
if let Err(e) = server.await {
error!("Spawning hyper server errorr: {}", e);
}
}
None => {
let server = Server::bind(&listener_socket_addr)
.http1_preserve_header_case(true)
.http1_title_case_headers(true)
.serve(make_service_fn(|_| bare_make_service_fn()))
.with_graceful_shutdown({
async {
rx.await.ok();
}
});
if let Err(e) = server.await {
error!("Spawning hyper server errorr: {}", e);
}
}
};
});
Ok(Self {
svc_mapping: server_service_mapping1,
handle: handle.into(),
shutdown: tx,
ssl_cfg,
})
}
pub async fn add_svc_handler_mapping(
&mut self,
fqdn: ResourceFqdn,
svc_handle: HttpHandler,
) -> Result<(), Error> {
let mut mapping = self.svc_mapping.write().await;
mapping.insert(fqdn.clone(), svc_handle);
Ok(())
}
pub async fn remove_svc_handler_mapping(&mut self, fqdn: ResourceFqdn) -> Result<(), Error> {
let mut mapping = self.svc_mapping.write().await;
mapping.remove(&fqdn);
Ok(())
}
}
/// HTTP status code 503
fn res_not_found() -> Response<Body> {
Response::builder()
.status(StatusCode::SERVICE_UNAVAILABLE)
.body(RES_NOT_FOUND.into())
.unwrap()
}
fn no_host_found(err: Error) -> Response<Body> {
let err_txt = format!("{}: Error {}", String::from_utf8_lossy(NO_HOST), err);
Response::builder()
.status(StatusCode::BAD_REQUEST)
.body(err_txt.into())
.unwrap()
}
fn respond_hyper_error(err: hyper::Error) -> Response<Body> {
let err_txt = format!("{}: Error {}", String::from_utf8_lossy(NO_HOST), err);
Response::builder()
.status(StatusCode::BAD_REQUEST)
.body(err_txt.into())
.unwrap()
}

34
backend/src/net/interface.rs
View File

@@ -1,9 +1,6 @@
use std::collections::BTreeMap;
use color_eyre::eyre::eyre;
use futures::TryStreamExt;
use indexmap::IndexSet;
use itertools::Either;
pub use models::InterfaceId;
use serde::{Deserialize, Deserializer, Serialize};
use sqlx::{Executor, Postgres};
@@ -11,7 +8,6 @@ use torut::onion::TorSecretKeyV3;
use tracing::instrument;
use crate::db::model::{InterfaceAddressMap, InterfaceAddresses};
use crate::id::Id;
use crate::s9pk::manifest::PackageId;
use crate::util::serde::Port;
use crate::{Error, ResultExt};
@@ -81,36 +77,6 @@ impl Interfaces {
}
Ok(interface_addresses)
}
#[instrument(skip(secrets))]
pub async fn tor_keys<Ex>(
&self,
secrets: &mut Ex,
package_id: &PackageId,
) -> Result<BTreeMap<InterfaceId, TorSecretKeyV3>, Error>
where
for<'a> &'a mut Ex: Executor<'a, Database = Postgres>,
{
Ok(sqlx::query!(
"SELECT interface, key FROM tor WHERE package = $1",
**package_id
)
.fetch_many(secrets)
.map_err(Error::from)
.try_filter_map(|qr| async move {
Ok(if let Either::Right(r) = qr {
let mut buf = [0; 64];
buf.clone_from_slice(r.key.get(0..64).ok_or_else(|| {
Error::new(eyre!("Invalid Tor Key Length"), crate::ErrorKind::Database)
})?);
Some((InterfaceId::from(Id::try_from(r.interface)?), buf.into()))
} else {
None
})
})
.try_collect()
.await?)
}
}
#[derive(Clone, Debug, Deserialize, Serialize)]

272
backend/src/net/keys.rs Normal file
View File

@@ -0,0 +1,272 @@
use color_eyre::eyre::eyre;
use ed25519_dalek::{ExpandedSecretKey, SecretKey};
use models::{Id, InterfaceId, PackageId};
use openssl::pkey::{PKey, Private};
use openssl::sha::Sha256;
use openssl::x509::X509;
use p256::elliptic_curve::pkcs8::EncodePrivateKey;
use sqlx::PgExecutor;
use ssh_key::private::Ed25519PrivateKey;
use torut::onion::{OnionAddressV3, TorSecretKeyV3};
use zeroize::Zeroize;
use crate::net::ssl::CertPair;
use crate::Error;
// TODO: delete once we may change tor addresses
async fn compat(
secrets: impl PgExecutor<'_>,
interface: &Option<(PackageId, InterfaceId)>,
) -> Result<Option<ExpandedSecretKey>, Error> {
if let Some((package, interface)) = interface {
if let Some(r) = sqlx::query!(
"SELECT key FROM tor WHERE package = $1 AND interface = $2",
**package,
**interface
)
.fetch_optional(secrets)
.await?
{
Ok(Some(ExpandedSecretKey::from_bytes(&r.key)?))
} else {
Ok(None)
}
} else {
if let Some(key) = sqlx::query!("SELECT tor_key FROM account WHERE id = 0")
.fetch_one(secrets)
.await?
.tor_key
{
Ok(Some(ExpandedSecretKey::from_bytes(&key)?))
} else {
Ok(None)
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct Key {
interface: Option<(PackageId, InterfaceId)>,
base: [u8; 32],
tor_key: [u8; 64], // Does NOT necessarily match base
}
impl Key {
pub fn interface(&self) -> Option<(PackageId, InterfaceId)> {
self.interface.clone()
}
pub fn as_bytes(&self) -> [u8; 32] {
self.base
}
pub fn internal_address(&self) -> String {
self.interface
.as_ref()
.map(|(pkg_id, _)| format!("{}.embassy", pkg_id))
.unwrap_or_else(|| "embassy".to_owned())
}
pub fn tor_key(&self) -> TorSecretKeyV3 {
ed25519_dalek::ExpandedSecretKey::from_bytes(&self.tor_key)
.unwrap()
.to_bytes()
.into()
}
pub fn tor_address(&self) -> OnionAddressV3 {
self.tor_key().public().get_onion_address()
}
pub fn base_address(&self) -> String {
self.tor_key()
.public()
.get_onion_address()
.get_address_without_dot_onion()
}
pub fn local_address(&self) -> String {
self.base_address() + ".local"
}
pub fn openssl_key_ed25519(&self) -> PKey<Private> {
PKey::private_key_from_raw_bytes(&self.base, openssl::pkey::Id::ED25519).unwrap()
}
pub fn openssl_key_nistp256(&self) -> PKey<Private> {
let mut buf = self.base;
loop {
if let Ok(k) = p256::SecretKey::from_be_bytes(&buf) {
return PKey::private_key_from_pkcs8(&*k.to_pkcs8_der().unwrap().as_bytes())
.unwrap();
}
let mut sha = Sha256::new();
sha.update(&buf);
buf = sha.finish();
}
}
pub fn ssh_key(&self) -> Ed25519PrivateKey {
Ed25519PrivateKey::from_bytes(&self.base)
}
pub(crate) fn from_pair(
interface: Option<(PackageId, InterfaceId)>,
bytes: [u8; 32],
tor_key: [u8; 64],
) -> Self {
Self {
interface,
tor_key,
base: bytes,
}
}
pub fn from_bytes(interface: Option<(PackageId, InterfaceId)>, bytes: [u8; 32]) -> Self {
Self::from_pair(
interface,
bytes,
ExpandedSecretKey::from(&SecretKey::from_bytes(&bytes).unwrap()).to_bytes(),
)
}
pub fn new(interface: Option<(PackageId, InterfaceId)>) -> Self {
Self::from_bytes(interface, rand::random())
}
pub(super) fn with_certs(self, certs: CertPair, int: X509, root: X509) -> KeyInfo {
KeyInfo {
key: self,
certs,
int,
root,
}
}
pub async fn for_package(
secrets: impl PgExecutor<'_>,
package: &PackageId,
) -> Result<Vec<Self>, Error> {
sqlx::query!(
r#"
SELECT
network_keys.package,
network_keys.interface,
network_keys.key,
tor.key AS "tor_key?"
FROM
network_keys
LEFT JOIN
tor
ON
network_keys.package = tor.package
AND
network_keys.interface = tor.interface
WHERE
network_keys.package = $1
"#,
**package
)
.fetch_all(secrets)
.await?
.into_iter()
.map(|row| {
let interface = Some((
package.clone(),
InterfaceId::from(Id::try_from(row.interface)?),
));
let bytes = row.key.try_into().map_err(|e: Vec<u8>| {
Error::new(
eyre!("Invalid length for network key {} expected 32", e.len()),
crate::ErrorKind::Database,
)
})?;
Ok(match row.tor_key {
Some(tor_key) => Key::from_pair(
interface,
bytes,
tor_key.try_into().map_err(|e: Vec<u8>| {
Error::new(
eyre!("Invalid length for tor key {} expected 64", e.len()),
crate::ErrorKind::Database,
)
})?,
),
None => Key::from_bytes(interface, bytes),
})
})
.collect()
}
pub async fn for_interface<Ex>(
secrets: &mut Ex,
interface: Option<(PackageId, InterfaceId)>,
) -> Result<Self, Error>
where
for<'a> &'a mut Ex: PgExecutor<'a>,
{
let tentative = rand::random::<[u8; 32]>();
let actual = if let Some((pkg, iface)) = &interface {
let k = tentative.as_slice();
let actual = sqlx::query!(
"INSERT INTO network_keys (package, interface, key) VALUES ($1, $2, $3) ON CONFLICT (package, interface) DO UPDATE SET package = EXCLUDED.package RETURNING key",
**pkg,
**iface,
k,
)
.fetch_one(&mut *secrets)
.await?.key;
let mut bytes = tentative;
bytes.clone_from_slice(actual.get(0..32).ok_or_else(|| {
Error::new(
eyre!("Invalid key size returned from DB"),
crate::ErrorKind::Database,
)
})?);
bytes
} else {
let actual = sqlx::query!("SELECT network_key FROM account WHERE id = 0")
.fetch_one(&mut *secrets)
.await?
.network_key;
let mut bytes = tentative;
bytes.clone_from_slice(actual.get(0..32).ok_or_else(|| {
Error::new(
eyre!("Invalid key size returned from DB"),
crate::ErrorKind::Database,
)
})?);
bytes
};
let mut res = Self::from_bytes(interface, actual);
if let Some(tor_key) = compat(secrets, &res.interface).await? {
res.tor_key = tor_key.to_bytes();
}
Ok(res)
}
}
impl Drop for Key {
fn drop(&mut self) {
self.base.zeroize();
self.tor_key.zeroize();
}
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub struct KeyInfo {
key: Key,
certs: CertPair,
int: X509,
root: X509,
}
impl KeyInfo {
pub fn key(&self) -> &Key {
&self.key
}
pub fn certs(&self) -> &CertPair {
&self.certs
}
pub fn int_ca(&self) -> &X509 {
&self.int
}
pub fn root_ca(&self) -> &X509 {
&self.root
}
pub fn fullchain_ed25519(&self) -> Vec<&X509> {
vec![&self.certs.ed25519, &self.int, &self.root]
}
pub fn fullchain_nistp256(&self) -> Vec<&X509> {
vec![&self.certs.nistp256, &self.int, &self.root]
}
}
#[test]
pub fn test_keygen() {
let key = Key::new(None);
key.tor_key();
key.openssl_key_nistp256();
}

63
backend/src/net/mdns.rs
View File

@@ -1,13 +1,11 @@
use std::collections::BTreeMap;
use std::net::Ipv4Addr;
use std::sync::{Arc, Weak};
use color_eyre::eyre::eyre;
use tokio::process::{Child, Command};
use tokio::sync::Mutex;
use torut::onion::TorSecretKeyV3;
use super::interface::InterfaceId;
use crate::s9pk::manifest::PackageId;
use crate::util::Invoke;
use crate::{Error, ResultExt};
@@ -39,25 +37,17 @@ impl MdnsController {
MdnsControllerInner::init().await?,
)))
}
pub async fn add<'a, I: IntoIterator<Item = (InterfaceId, TorSecretKeyV3)>>(
&self,
pkg_id: &PackageId,
interfaces: I,
) -> Result<(), Error> {
self.0.lock().await.add(pkg_id, interfaces).await
pub async fn add(&self, alias: String) -> Result<Arc<()>, Error> {
self.0.lock().await.add(alias).await
}
pub async fn remove<I: IntoIterator<Item = InterfaceId>>(
&self,
pkg_id: &PackageId,
interfaces: I,
) -> Result<(), Error> {
self.0.lock().await.remove(pkg_id, interfaces).await
pub async fn gc(&self, alias: String) -> Result<(), Error> {
self.0.lock().await.gc(alias).await
}
}
pub struct MdnsControllerInner {
alias_cmd: Option<Child>,
services: BTreeMap<(PackageId, InterfaceId), TorSecretKeyV3>,
services: BTreeMap<String, Weak<()>>,
}
impl MdnsControllerInner {
@@ -76,35 +66,30 @@ impl MdnsControllerInner {
self.alias_cmd = Some(
Command::new("avahi-alias")
.kill_on_drop(true)
.args(self.services.iter().map(|(_, key)| {
key.public()
.get_onion_address()
.get_address_without_dot_onion()
}))
.args(
self.services
.iter()
.filter(|(_, rc)| rc.strong_count() > 0)
.map(|(s, _)| s),
)
.spawn()?,
);
Ok(())
}
async fn add<'a, I: IntoIterator<Item = (InterfaceId, TorSecretKeyV3)>>(
&mut self,
pkg_id: &PackageId,
interfaces: I,
) -> Result<(), Error> {
self.services.extend(
interfaces
.into_iter()
.map(|(interface_id, key)| ((pkg_id.clone(), interface_id), key)),
);
async fn add(&mut self, alias: String) -> Result<Arc<()>, Error> {
let rc = if let Some(rc) = Weak::upgrade(&self.services.remove(&alias).unwrap_or_default())
{
rc
} else {
Arc::new(())
};
self.services.insert(alias, Arc::downgrade(&rc));
self.sync().await?;
Ok(())
Ok(rc)
}
async fn remove<I: IntoIterator<Item = InterfaceId>>(
&mut self,
pkg_id: &PackageId,
interfaces: I,
) -> Result<(), Error> {
for interface_id in interfaces {
self.services.remove(&(pkg_id.clone(), interface_id));
async fn gc(&mut self, alias: String) -> Result<(), Error> {
if let Some(rc) = Weak::upgrade(&self.services.remove(&alias).unwrap_or_default()) {
self.services.insert(alias, Arc::downgrade(&rc));
}
self.sync().await?;
Ok(())

27
backend/src/net/mod.rs
View File

@@ -1,54 +1,33 @@
use std::collections::BTreeMap;
use std::sync::Arc;
use futures::future::BoxFuture;
use hyper::{Body, Error as HyperError, Request, Response};
use indexmap::IndexSet;
use rpc_toolkit::command;
use self::interface::InterfaceId;
use crate::net::interface::LanPortConfig;
use crate::util::serde::Port;
use crate::Error;
pub mod cert_resolver;
pub mod dhcp;
pub mod dns;
pub mod embassy_service_http_server;
pub mod interface;
pub mod keys;
#[cfg(feature = "avahi")]
pub mod mdns;
pub mod net_controller;
pub mod net_utils;
pub mod proxy_controller;
pub mod ssl;
pub mod static_server;
pub mod tor;
pub mod vhost_controller;
pub mod web_server;
pub mod wifi;
const PACKAGE_CERT_PATH: &str = "/var/lib/embassy/ssl";
pub const PACKAGE_CERT_PATH: &str = "/var/lib/embassy/ssl";
#[command(subcommands(tor::tor, dhcp::dhcp))]
pub fn net() -> Result<(), Error> {
Ok(())
}
#[derive(Default)]
struct PackageNetInfo {
interfaces: BTreeMap<InterfaceId, InterfaceMetadata>,
}
pub struct InterfaceMetadata {
pub fqdn: String,
pub lan_config: BTreeMap<Port, LanPortConfig>,
pub protocols: IndexSet<String>,
}
/// Indicates that the net controller has created the
/// SSL keys
#[derive(Clone, Copy)]
pub struct GeneratedCertificateMountPoint(());
pub type HttpHandler = Arc<
dyn Fn(Request<Body>) -> BoxFuture<'static, Result<Response<Body>, HyperError>> + Send + Sync,
>;

562
backend/src/net/net_controller.rs
View File

@@ -1,278 +1,346 @@
use std::net::{Ipv4Addr, SocketAddr};
use std::path::PathBuf;
use std::str::FromStr;
use std::collections::BTreeMap;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::{Arc, Weak};
use color_eyre::eyre::eyre;
use models::InterfaceId;
use openssl::pkey::{PKey, Private};
use openssl::x509::X509;
use patch_db::DbHandle;
use sqlx::PgPool;
use torut::onion::{OnionAddressV3, TorSecretKeyV3};
use sqlx::PgExecutor;
use tracing::instrument;
use crate::context::RpcContext;
use crate::hostname::{get_embassyd_tor_addr, get_hostname, HostNameReceipt};
use crate::error::ErrorCollection;
use crate::hostname::Hostname;
use crate::net::dns::DnsController;
use crate::net::interface::{Interface, TorConfig};
use crate::net::keys::Key;
#[cfg(feature = "avahi")]
use crate::net::mdns::MdnsController;
use crate::net::net_utils::ResourceFqdn;
use crate::net::proxy_controller::ProxyController;
use crate::net::ssl::SslManager;
use crate::net::ssl::{export_cert, SslManager};
use crate::net::tor::TorController;
use crate::net::{
GeneratedCertificateMountPoint, HttpHandler, InterfaceMetadata, PACKAGE_CERT_PATH,
};
use crate::net::vhost_controller::VHostController;
use crate::s9pk::manifest::PackageId;
use crate::Error;
use crate::volume::cert_dir;
use crate::{Error, HOST_IP};
pub struct NetController {
pub tor: TorController,
pub(super) tor: TorController,
#[cfg(feature = "avahi")]
pub mdns: MdnsController,
pub proxy: ProxyController,
pub ssl: SslManager,
pub dns: DnsController,
pub(super) mdns: MdnsController,
pub(super) vhost: VHostController,
pub(super) dns: DnsController,
pub(super) ssl: Arc<SslManager>,
pub(super) os_bindings: Vec<Arc<()>>,
}
impl NetController {
#[instrument(skip(db, db_handle))]
pub async fn init<Db: DbHandle>(
embassyd_addr: SocketAddr,
embassyd_tor_key: TorSecretKeyV3,
#[instrument]
pub async fn init(
tor_control: SocketAddr,
dns_bind: &[SocketAddr],
db: PgPool,
db_handle: &mut Db,
import_root_ca: Option<(PKey<Private>, X509)>,
ssl: SslManager,
hostname: &Hostname,
os_key: &Key,
) -> Result<Self, Error> {
let receipts = HostNameReceipt::new(db_handle).await?;
let embassy_host_name = get_hostname(db_handle, &receipts).await?;
let embassy_name = embassy_host_name.local_domain_name();
let fqdn_name = ResourceFqdn::from_str(&embassy_name)?;
let ssl = match import_root_ca {
None => SslManager::init(db.clone(), db_handle).await,
Some(a) => SslManager::import_root_ca(db.clone(), a.0, a.1).await,
}?;
Ok(Self {
tor: TorController::init(embassyd_addr, embassyd_tor_key, tor_control).await?,
let ssl = Arc::new(ssl);
let mut res = Self {
tor: TorController::init(tor_control).await?,
#[cfg(feature = "avahi")]
mdns: MdnsController::init().await?,
proxy: ProxyController::init(embassyd_addr, fqdn_name, ssl.clone()).await?,
ssl,
vhost: VHostController::new(ssl.clone()),
dns: DnsController::init(dns_bind).await?,
ssl,
os_bindings: Vec::new(),
};
res.add_os_bindings(hostname, os_key).await?;
Ok(res)
}
async fn add_os_bindings(&mut self, hostname: &Hostname, key: &Key) -> Result<(), Error> {
// Internal DNS
self.vhost
.add(
key.clone(),
Some("embassy".into()),
443,
([127, 0, 0, 1], 80).into(),
false,
)
.await?;
self.os_bindings
.push(self.dns.add(None, HOST_IP.into()).await?);
// LAN IP
self.os_bindings.push(
self.vhost
.add(key.clone(), None, 443, ([127, 0, 0, 1], 80).into(), false)
.await?,
);
// localhost
self.os_bindings.push(
self.vhost
.add(
key.clone(),
Some("localhost".into()),
443,
([127, 0, 0, 1], 80).into(),
false,
)
.await?,
);
self.os_bindings.push(
self.vhost
.add(
key.clone(),
Some(hostname.no_dot_host_name()),
443,
([127, 0, 0, 1], 80).into(),
false,
)
.await?,
);
// LAN mDNS
self.os_bindings.push(
self.vhost
.add(
key.clone(),
Some(hostname.local_domain_name()),
443,
([127, 0, 0, 1], 80).into(),
false,
)
.await?,
);
// Tor (http)
self.os_bindings.push(
self.tor
.add(&key.tor_key(), 80, ([127, 0, 0, 1], 80).into())
.await?,
);
// Tor (https)
self.os_bindings.push(
self.vhost
.add(
key.clone(),
Some(key.tor_address().to_string()),
443,
([127, 0, 0, 1], 80).into(),
false,
)
.await?,
);
self.os_bindings.push(
self.tor
.add(&key.tor_key(), 443, ([127, 0, 0, 1], 443).into())
.await?,
);
Ok(())
}
#[instrument(skip(self))]
pub async fn create_service(
self: &Arc<Self>,
package: PackageId,
ip: Ipv4Addr,
) -> Result<NetService, Error> {
let dns = self.dns.add(Some(package.clone()), ip).await?;
Ok(NetService {
id: package,
ip,
dns,
controller: Arc::downgrade(self),
tor: BTreeMap::new(),
lan: BTreeMap::new(),
})
}
pub async fn setup_embassy_ui(rpc_ctx: RpcContext) -> Result<(), Error> {
NetController::setup_embassy_http_ui_handle(rpc_ctx.clone()).await?;
NetController::setup_embassy_https_ui_handle(rpc_ctx.clone()).await?;
Ok(())
}
async fn setup_embassy_https_ui_handle(rpc_ctx: RpcContext) -> Result<(), Error> {
let host_name = rpc_ctx.net_controller.proxy.get_hostname().await;
let host_name_fqdn: ResourceFqdn = host_name.parse()?;
let handler: HttpHandler =
crate::net::static_server::main_ui_server_router(rpc_ctx.clone()).await?;
let eos_pkg_id: PackageId = "embassy".parse().unwrap();
if let ResourceFqdn::Uri {
full_uri: _,
root,
tld: _,
} = host_name_fqdn.clone()
{
let root_cert = rpc_ctx
.net_controller
.ssl
.certificate_for(&root, &eos_pkg_id)
.await?;
rpc_ctx
.net_controller
.proxy
.add_certificate_to_resolver(host_name_fqdn.clone(), root_cert.clone())
.await?;
rpc_ctx
.net_controller
.proxy
.add_handle(443, host_name_fqdn.clone(), handler.clone(), true)
.await?;
};
// serving ip https is not yet supported
Ok(())
}
async fn setup_embassy_http_ui_handle(rpc_ctx: RpcContext) -> Result<(), Error> {
let host_name = rpc_ctx.net_controller.proxy.get_hostname().await;
let embassy_tor_addr = get_embassyd_tor_addr(rpc_ctx.clone()).await?;
let embassy_tor_fqdn: ResourceFqdn = embassy_tor_addr.parse()?;
let host_name_fqdn: ResourceFqdn = host_name.parse()?;
let ip_fqdn: ResourceFqdn = ResourceFqdn::IpAddr;
let localhost_fqdn = ResourceFqdn::LocalHost;
let handler: HttpHandler =
crate::net::static_server::main_ui_server_router(rpc_ctx.clone()).await?;
rpc_ctx
.net_controller
.proxy
.add_handle(80, embassy_tor_fqdn.clone(), handler.clone(), false)
.await?;
rpc_ctx
.net_controller
.proxy
.add_handle(80, host_name_fqdn.clone(), handler.clone(), false)
.await?;
rpc_ctx
.net_controller
.proxy
.add_handle(80, ip_fqdn.clone(), handler.clone(), false)
.await?;
rpc_ctx
.net_controller
.proxy
.add_handle(80, localhost_fqdn.clone(), handler.clone(), false)
.await?;
Ok(())
}
pub fn ssl_directory_for(pkg_id: &PackageId) -> PathBuf {
PathBuf::from(PACKAGE_CERT_PATH).join(pkg_id)
}
#[instrument(skip(self, interfaces, _generated_certificate))]
pub async fn add<'a, I>(
async fn add_tor(
&self,
pkg_id: &PackageId,
ip: Ipv4Addr,
interfaces: I,
_generated_certificate: GeneratedCertificateMountPoint,
) -> Result<(), Error>
where
I: IntoIterator<Item = (InterfaceId, &'a Interface, TorSecretKeyV3)> + Clone,
for<'b> &'b I: IntoIterator<Item = &'b (InterfaceId, &'a Interface, TorSecretKeyV3)>,
{
let interfaces_tor = interfaces
.clone()
.into_iter()
.filter_map(|i| match i.1.tor_config.clone() {
None => None,
Some(cfg) => Some((i.0, cfg, i.2)),
})
.collect::<Vec<(InterfaceId, TorConfig, TorSecretKeyV3)>>();
let (tor_res, _, proxy_res, _) = tokio::join!(
self.tor.add(pkg_id, ip, interfaces_tor),
{
#[cfg(feature = "avahi")]
let mdns_fut = self.mdns.add(
pkg_id,
interfaces
.clone()
.into_iter()
.map(|(interface_id, _, key)| (interface_id, key)),
);
key: &Key,
external: u16,
target: SocketAddr,
) -> Result<Vec<Arc<()>>, Error> {
let mut rcs = Vec::with_capacity(1);
rcs.push(self.tor.add(&key.tor_key(), external, target).await?);
Ok(rcs)
}
#[cfg(not(feature = "avahi"))]
let mdns_fut = futures::future::ready(());
mdns_fut
},
{
let interfaces =
interfaces
.clone()
.into_iter()
.filter_map(|(id, interface, tor_key)| {
interface.lan_config.as_ref().map(|cfg| {
(
id,
InterfaceMetadata {
fqdn: OnionAddressV3::from(&tor_key.public())
.get_address_without_dot_onion()
+ ".local",
lan_config: cfg.clone(),
protocols: interface.protocols.clone(),
},
)
})
});
self.proxy
.add_docker_service(pkg_id.clone(), ip, interfaces)
},
self.dns.add(pkg_id, ip),
async fn remove_tor(&self, key: &Key, external: u16, rcs: Vec<Arc<()>>) -> Result<(), Error> {
drop(rcs);
self.tor.gc(&key.tor_key(), external).await
}
async fn add_lan(
&self,
key: Key,
external: u16,
target: SocketAddr,
connect_ssl: bool,
) -> Result<Vec<Arc<()>>, Error> {
let mut rcs = Vec::with_capacity(2);
rcs.push(
self.vhost
.add(
key.clone(),
Some(key.local_address()),
external,
target.into(),
connect_ssl,
)
.await?,
);
tor_res?;
proxy_res?;
Ok(())
#[cfg(feature = "avahi")]
rcs.push(self.mdns.add(key.base_address()).await?);
Ok(rcs)
}
#[instrument(skip(self, interfaces))]
pub async fn remove<I: IntoIterator<Item = InterfaceId> + Clone>(
&self,
pkg_id: &PackageId,
ip: Ipv4Addr,
interfaces: I,
) -> Result<(), Error> {
let (tor_res, _, proxy_res, _) = tokio::join!(
self.tor.remove(pkg_id, interfaces.clone()),
{
#[cfg(feature = "avahi")]
let mdns_fut = self.mdns.remove(pkg_id, interfaces);
#[cfg(not(feature = "avahi"))]
let mdns_fut = futures::future::ready(());
mdns_fut
},
self.proxy.remove_docker_service(pkg_id),
self.dns.remove(pkg_id, ip),
);
tor_res?;
proxy_res?;
Ok(())
}
pub async fn generate_certificate_mountpoint<'a, I>(
&self,
pkg_id: &PackageId,
interfaces: &I,
) -> Result<GeneratedCertificateMountPoint, Error>
where
I: IntoIterator<Item = (InterfaceId, &'a Interface, TorSecretKeyV3)> + Clone,
for<'b> &'b I: IntoIterator<Item = &'b (InterfaceId, &'a Interface, TorSecretKeyV3)>,
{
tracing::info!("Generating SSL Certificate mountpoints for {}", pkg_id);
let package_path = PathBuf::from(PACKAGE_CERT_PATH).join(pkg_id);
tokio::fs::create_dir_all(&package_path).await?;
for (id, _, key) in interfaces {
let dns_base = OnionAddressV3::from(&key.public()).get_address_without_dot_onion();
let ssl_path_key = package_path.join(format!("{}.key.pem", id));
let ssl_path_cert = package_path.join(format!("{}.cert.pem", id));
let (key, chain) = self.ssl.certificate_for(&dns_base, pkg_id).await?;
tokio::try_join!(
crate::net::ssl::export_key(&key, &ssl_path_key),
crate::net::ssl::export_cert(&chain, &ssl_path_cert)
)?;
}
Ok(GeneratedCertificateMountPoint(()))
}
pub async fn export_root_ca(&self) -> Result<(PKey<Private>, X509), Error> {
self.ssl.export_root_ca().await
async fn remove_lan(&self, key: &Key, external: u16, rcs: Vec<Arc<()>>) -> Result<(), Error> {
drop(rcs);
#[cfg(feature = "avahi")]
self.mdns.gc(key.base_address()).await?;
self.vhost.gc(Some(key.local_address()), external).await
}
}
pub struct NetService {
id: PackageId,
ip: Ipv4Addr,
dns: Arc<()>,
controller: Weak<NetController>,
tor: BTreeMap<(InterfaceId, u16), (Key, Vec<Arc<()>>)>,
lan: BTreeMap<(InterfaceId, u16), (Key, Vec<Arc<()>>)>,
}
impl NetService {
fn net_controller(&self) -> Result<Arc<NetController>, Error> {
Weak::upgrade(&self.controller).ok_or_else(|| {
Error::new(
eyre!("NetController is shutdown"),
crate::ErrorKind::Network,
)
})
}
pub async fn add_tor<Ex>(
&mut self,
secrets: &mut Ex,
id: InterfaceId,
external: u16,
internal: u16,
) -> Result<(), Error>
where
for<'a> &'a mut Ex: PgExecutor<'a>,
{
let key = Key::for_interface(secrets, Some((self.id.clone(), id.clone()))).await?;
let ctrl = self.net_controller()?;
let tor_idx = (id, external);
let mut tor = self
.tor
.remove(&tor_idx)
.unwrap_or_else(|| (key.clone(), Vec::new()));
tor.1.append(
&mut ctrl
.add_tor(&key, external, SocketAddr::new(self.ip.into(), internal))
.await?,
);
self.tor.insert(tor_idx, tor);
Ok(())
}
pub async fn remove_tor(&mut self, id: InterfaceId, external: u16) -> Result<(), Error> {
let ctrl = self.net_controller()?;
if let Some((key, rcs)) = self.tor.remove(&(id, external)) {
ctrl.remove_tor(&key, external, rcs).await?;
}
Ok(())
}
pub async fn add_lan<Ex>(
&mut self,
secrets: &mut Ex,
id: InterfaceId,
external: u16,
internal: u16,
connect_ssl: bool,
) -> Result<(), Error>
where
for<'a> &'a mut Ex: PgExecutor<'a>,
{
let key = Key::for_interface(secrets, Some((self.id.clone(), id.clone()))).await?;
let ctrl = self.net_controller()?;
let lan_idx = (id, external);
let mut lan = self
.lan
.remove(&lan_idx)
.unwrap_or_else(|| (key.clone(), Vec::new()));
lan.1.append(
&mut ctrl
.add_lan(
key,
external,
SocketAddr::new(self.ip.into(), internal),
connect_ssl,
)
.await?,
);
self.lan.insert(lan_idx, lan);
Ok(())
}
pub async fn remove_lan(&mut self, id: InterfaceId, external: u16) -> Result<(), Error> {
let ctrl = self.net_controller()?;
if let Some((key, rcs)) = self.lan.remove(&(id, external)) {
ctrl.remove_lan(&key, external, rcs).await?;
}
Ok(())
}
pub async fn export_cert<Ex>(
&self,
secrets: &mut Ex,
id: &InterfaceId,
ip: IpAddr,
) -> Result<(), Error>
where
for<'a> &'a mut Ex: PgExecutor<'a>,
{
let key = Key::for_interface(secrets, Some((self.id.clone(), id.clone()))).await?;
let ctrl = self.net_controller()?;
let cert = ctrl.ssl.with_certs(key, ip).await?;
export_cert(&cert.fullchain_nistp256(), &cert_dir(&self.id, id)).await?; // TODO: can upgrade to ed25519?
Ok(())
}
pub async fn remove_all(mut self) -> Result<(), Error> {
let mut errors = ErrorCollection::new();
if let Some(ctrl) = Weak::upgrade(&self.controller) {
for ((_, external), (key, rcs)) in std::mem::take(&mut self.lan) {
errors.handle(ctrl.remove_lan(&key, external, rcs).await);
}
for ((_, external), (key, rcs)) in std::mem::take(&mut self.tor) {
errors.handle(ctrl.remove_tor(&key, external, rcs).await);
}
std::mem::take(&mut self.dns);
errors.handle(ctrl.dns.gc(Some(self.id.clone()), self.ip).await);
errors.into_result()
} else {
Err(Error::new(
eyre!("NetController is shutdown"),
crate::ErrorKind::Network,
))
}
}
}
impl Drop for NetService {
fn drop(&mut self) {
let svc = std::mem::replace(
self,
NetService {
id: Default::default(),
ip: [0, 0, 0, 0].into(),
dns: Default::default(),
controller: Default::default(),
tor: Default::default(),
lan: Default::default(),
},
);
tokio::spawn(async move { svc.remove_all().await.unwrap() });
}
}

160
backend/src/net/net_utils.rs
View File

@@ -1,14 +1,12 @@
use std::fmt;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
use std::convert::Infallible;
use std::net::{Ipv4Addr, Ipv6Addr};
use std::path::Path;
use std::str::FromStr;
use async_stream::try_stream;
use color_eyre::eyre::eyre;
use futures::stream::BoxStream;
use futures::{StreamExt, TryStreamExt};
use http::{Request, Uri};
use hyper::Body;
use ipnet::{Ipv4Net, Ipv6Net};
use tokio::process::Command;
use crate::util::Invoke;
@@ -19,11 +17,7 @@ fn parse_iface_ip(output: &str) -> Result<Option<&str>, Error> {
if output.is_empty() {
return Ok(None);
}
if let Some(ip) = output
.split_ascii_whitespace()
.nth(3)
.and_then(|range| range.split("/").next())
{
if let Some(ip) = output.split_ascii_whitespace().nth(3) {
Ok(Some(ip))
} else {
Err(Error::new(
@@ -33,7 +27,7 @@ fn parse_iface_ip(output: &str) -> Result<Option<&str>, Error> {
}
}
pub async fn get_iface_ipv4_addr(iface: &str) -> Result<Option<Ipv4Addr>, Error> {
pub async fn get_iface_ipv4_addr(iface: &str) -> Result<Option<(Ipv4Addr, Ipv4Net)>, Error> {
Ok(parse_iface_ip(&String::from_utf8(
Command::new("ip")
.arg("-4")
@@ -44,11 +38,11 @@ pub async fn get_iface_ipv4_addr(iface: &str) -> Result<Option<Ipv4Addr>, Error>
.invoke(crate::ErrorKind::Network)
.await?,
)?)?
.map(|s| s.parse())
.map(|s| Ok::<_, Error>((s.split("/").next().unwrap().parse()?, s.parse()?)))
.transpose()?)
}
pub async fn get_iface_ipv6_addr(iface: &str) -> Result<Option<Ipv6Addr>, Error> {
pub async fn get_iface_ipv6_addr(iface: &str) -> Result<Option<(Ipv6Addr, Ipv6Net)>, Error> {
Ok(parse_iface_ip(&String::from_utf8(
Command::new("ip")
.arg("-6")
@@ -59,7 +53,7 @@ pub async fn get_iface_ipv6_addr(iface: &str) -> Result<Option<Ipv6Addr>, Error>
.invoke(crate::ErrorKind::Network)
.await?,
)?)?
.map(|s| s.parse())
.map(|s| Ok::<_, Error>((s.split("/").next().unwrap().parse()?, s.parse()?)))
.transpose()?)
}
@@ -110,132 +104,20 @@ pub async fn find_eth_iface() -> Result<String, Error> {
))
}
pub fn host_addr_fqdn(req: &Request<Body>) -> Result<ResourceFqdn, Error> {
let host = req.headers().get(http::header::HOST);
match host {
Some(host) => {
let host_str = host
.to_str()
.map_err(|e| Error::new(eyre!("{}", e), crate::ErrorKind::Ascii))?
.to_string();
let host_uri: ResourceFqdn = host_str.split(':').next().unwrap().parse()?;
Ok(host_uri)
}
None => Err(Error::new(
eyre!("No Host header"),
crate::ErrorKind::MissingHeader,
)),
#[pin_project::pin_project]
pub struct SingleAccept<T>(Option<T>);
impl<T> SingleAccept<T> {
pub fn new(conn: T) -> Self {
Self(Some(conn))
}
}
#[derive(Eq, PartialEq, PartialOrd, Ord, Debug, Clone)]
pub enum ResourceFqdn {
IpAddr,
Uri {
full_uri: String,
root: String,
tld: Tld,
},
LocalHost,
}
impl fmt::Display for ResourceFqdn {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
ResourceFqdn::Uri {
full_uri,
root: _,
tld: _,
} => {
write!(f, "{}", full_uri)
}
ResourceFqdn::LocalHost => write!(f, "localhost"),
ResourceFqdn::IpAddr => write!(f, "ip-address"),
}
impl<T> hyper::server::accept::Accept for SingleAccept<T> {
type Conn = T;
type Error = Infallible;
fn poll_accept(
self: std::pin::Pin<&mut Self>,
_cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Option<Result<Self::Conn, Self::Error>>> {
std::task::Poll::Ready(self.project().0.take().map(Ok))
}
}
#[derive(PartialEq, Eq, PartialOrd, Ord, Debug, Clone)]
pub enum Tld {
Local,
Onion,
Embassy,
}
impl fmt::Display for Tld {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
match self {
Tld::Local => write!(f, ".local"),
Tld::Onion => write!(f, ".onion"),
Tld::Embassy => write!(f, ".embassy"),
}
}
}
impl FromStr for ResourceFqdn {
type Err = Error;
fn from_str(input: &str) -> Result<ResourceFqdn, Self::Err> {
if input.parse::<IpAddr>().is_ok() {
return Ok(ResourceFqdn::IpAddr);
}
if input == "localhost" {
return Ok(ResourceFqdn::LocalHost);
}
let hostname_split: Vec<&str> = input.split('.').collect();
if hostname_split.len() != 2 {
return Err(Error::new(
eyre!("invalid url tld number: add support for tldextract to parse complex urls like blah.domain.co.uk and etc?"),
crate::ErrorKind::ParseUrl,
));
}
match hostname_split[1] {
"local" => Ok(ResourceFqdn::Uri {
full_uri: input.to_owned(),
root: hostname_split[0].to_owned(),
tld: Tld::Local,
}),
"embassy" => Ok(ResourceFqdn::Uri {
full_uri: input.to_owned(),
root: hostname_split[0].to_owned(),
tld: Tld::Embassy,
}),
"onion" => Ok(ResourceFqdn::Uri {
full_uri: input.to_owned(),
root: hostname_split[0].to_owned(),
tld: Tld::Onion,
}),
_ => Err(Error::new(
eyre!("Unknown TLD for enum"),
crate::ErrorKind::ParseUrl,
)),
}
}
}
impl TryFrom<Uri> for ResourceFqdn {
type Error = Error;
fn try_from(value: Uri) -> Result<Self, Self::Error> {
Self::from_str(&value.to_string())
}
}
pub fn is_upgrade_req(req: &Request<Body>) -> bool {
req.headers()
.get("connection")
.and_then(|c| c.to_str().ok())
.map(|c| {
c.split(",")
.any(|c| c.trim().eq_ignore_ascii_case("upgrade"))
})
.unwrap_or(false)
}

334
backend/src/net/proxy_controller.rs
View File

@@ -1,334 +0,0 @@
use std::collections::BTreeMap;
use std::net::{Ipv4Addr, SocketAddr};
use std::str::FromStr;
use std::sync::Arc;
use color_eyre::eyre::eyre;
use futures::FutureExt;
use http::uri::{Authority, Scheme};
use http::{Request, Response, Uri};
use hyper::{Body, Error as HyperError};
use models::{InterfaceId, PackageId};
use openssl::pkey::{PKey, Private};
use openssl::x509::X509;
use tokio::sync::Mutex;
use tracing::{error, instrument};
use crate::net::net_utils::{is_upgrade_req, ResourceFqdn};
use crate::net::ssl::SslManager;
use crate::net::vhost_controller::VHOSTController;
use crate::net::{HttpHandler, InterfaceMetadata, PackageNetInfo};
use crate::{Error, ResultExt};
pub struct ProxyController {
inner: Mutex<ProxyControllerInner>,
}
impl ProxyController {
pub async fn init(
embassyd_socket_addr: SocketAddr,
embassy_fqdn: ResourceFqdn,
ssl_manager: SslManager,
) -> Result<Self, Error> {
Ok(ProxyController {
inner: Mutex::new(
ProxyControllerInner::init(embassyd_socket_addr, embassy_fqdn, ssl_manager).await?,
),
})
}
pub async fn add_docker_service<I: IntoIterator<Item = (InterfaceId, InterfaceMetadata)>>(
&self,
package: PackageId,
ipv4: Ipv4Addr,
interfaces: I,
) -> Result<(), Error> {
self.inner
.lock()
.await
.add_docker_service(package, ipv4, interfaces)
.await
}
pub async fn remove_docker_service(&self, package: &PackageId) -> Result<(), Error> {
self.inner.lock().await.remove_docker_service(package).await
}
pub async fn add_certificate_to_resolver(
&self,
fqdn: ResourceFqdn,
cert_data: (PKey<Private>, Vec<X509>),
) -> Result<(), Error> {
self.inner
.lock()
.await
.add_certificate_to_resolver(fqdn, cert_data)
.await
}
pub async fn add_handle(
&self,
ext_port: u16,
fqdn: ResourceFqdn,
handler: HttpHandler,
is_ssl: bool,
) -> Result<(), Error> {
self.inner
.lock()
.await
.add_handle(ext_port, fqdn, handler, is_ssl)
.await
}
pub async fn get_hostname(&self) -> String {
self.inner.lock().await.get_embassy_hostname()
}
async fn proxy(
client: &hyper::Client<hyper::client::HttpConnector>,
mut req: Request<Body>,
addr: SocketAddr,
) -> Result<Response<Body>, HyperError> {
let mut uri = std::mem::take(req.uri_mut()).into_parts();
uri.scheme = Some(Scheme::HTTP);
uri.authority = Authority::from_str(&addr.to_string()).ok();
match Uri::from_parts(uri) {
Ok(uri) => *req.uri_mut() = uri,
Err(e) => error!("Error rewriting uri: {}", e),
}
let addr = req.uri().to_string();
if is_upgrade_req(&req) {
let upgraded_req = hyper::upgrade::on(&mut req);
let mut res = client.request(req).await?;
let upgraded_res = hyper::upgrade::on(&mut res);
tokio::spawn(async move {
if let Err(e) = async {
let mut req = upgraded_req.await?;
let mut res = upgraded_res.await?;
tokio::io::copy_bidirectional(&mut req, &mut res).await?;
Ok::<_, color_eyre::eyre::Report>(())
}
.await
{
error!("error binding together tcp streams for {}: {}", addr, e);
}
});
Ok(res)
} else {
client.request(req).await
}
}
}
struct ProxyControllerInner {
ssl_manager: SslManager,
vhosts: VHOSTController,
embassyd_fqdn: ResourceFqdn,
docker_interfaces: BTreeMap<PackageId, PackageNetInfo>,
docker_iface_lookups: BTreeMap<(PackageId, InterfaceId), ResourceFqdn>,
}
impl ProxyControllerInner {
#[instrument]
async fn init(
embassyd_socket_addr: SocketAddr,
embassyd_fqdn: ResourceFqdn,
ssl_manager: SslManager,
) -> Result<Self, Error> {
let inner = ProxyControllerInner {
vhosts: VHOSTController::init(embassyd_socket_addr),
ssl_manager,
embassyd_fqdn,
docker_interfaces: BTreeMap::new(),
docker_iface_lookups: BTreeMap::new(),
};
Ok(inner)
}
async fn add_certificate_to_resolver(
&mut self,
hostname: ResourceFqdn,
cert_data: (PKey<Private>, Vec<X509>),
) -> Result<(), Error> {
self.vhosts
.cert_resolver
.add_certificate_to_resolver(hostname, cert_data)
.await
.map_err(|err| {
Error::new(
eyre!("Unable to add ssl cert to the resolver: {}", err),
crate::ErrorKind::Network,
)
})?;
Ok(())
}
async fn add_package_certificate_to_resolver(
&mut self,
resource_fqdn: ResourceFqdn,
pkg_id: PackageId,
) -> Result<(), Error> {
let package_cert = match resource_fqdn.clone() {
ResourceFqdn::IpAddr => {
return Err(Error::new(
eyre!("ssl not supported for ip addresses"),
crate::ErrorKind::Network,
))
}
ResourceFqdn::Uri {
full_uri: _,
root,
tld: _,
} => self.ssl_manager.certificate_for(&root, &pkg_id).await?,
ResourceFqdn::LocalHost => {
return Err(Error::new(
eyre!("ssl not supported for localhost"),
crate::ErrorKind::Network,
))
}
};
self.vhosts
.cert_resolver
.add_certificate_to_resolver(resource_fqdn, package_cert)
.await
.map_err(|err| {
Error::new(
eyre!("Unable to add ssl cert to the resolver: {}", err),
crate::ErrorKind::Network,
)
})?;
Ok(())
}
pub async fn add_handle(
&mut self,
external_svc_port: u16,
fqdn: ResourceFqdn,
svc_handler: HttpHandler,
is_ssl: bool,
) -> Result<(), Error> {
self.vhosts
.add_server_or_handle(external_svc_port, fqdn, svc_handler, is_ssl)
.await
}
#[instrument(skip(self, interfaces))]
pub async fn add_docker_service<I: IntoIterator<Item = (InterfaceId, InterfaceMetadata)>>(
&mut self,
package: PackageId,
docker_ipv4: Ipv4Addr,
interfaces: I,
) -> Result<(), Error> {
let mut interface_map = interfaces
.into_iter()
.filter(|(_, meta)| {
// don't add stuff for anything we can't connect to over some flavor of http
(meta.protocols.contains("http") || meta.protocols.contains("https"))
// also don't add anything unless it has at least one exposed port
&& !meta.lan_config.is_empty()
})
.collect::<BTreeMap<InterfaceId, InterfaceMetadata>>();
for (id, meta) in interface_map.iter() {
for (external_svc_port, lan_port_config) in meta.lan_config.iter() {
let full_fqdn = ResourceFqdn::from_str(&meta.fqdn).unwrap();
self.docker_iface_lookups
.insert((package.clone(), id.clone()), full_fqdn.clone());
self.add_package_certificate_to_resolver(full_fqdn.clone(), package.clone())
.await?;
let svc_handler =
Self::create_docker_handle((docker_ipv4, lan_port_config.internal).into())
.await;
self.add_handle(
external_svc_port.0,
full_fqdn.clone(),
svc_handler,
lan_port_config.ssl,
)
.await?;
}
}
let docker_interface = self.docker_interfaces.entry(package.clone()).or_default();
docker_interface.interfaces.append(&mut interface_map);
Ok(())
}
async fn create_docker_handle(internal_addr: SocketAddr) -> HttpHandler {
let svc_handler: HttpHandler = Arc::new(move |req| {
let client = hyper::client::Client::builder()
.set_host(false)
.build_http();
async move { ProxyController::proxy(&client, req, internal_addr).await }.boxed()
});
svc_handler
}
#[instrument(skip(self))]
pub async fn remove_docker_service(&mut self, package: &PackageId) -> Result<(), Error> {
let mut server_removals: Vec<(u16, InterfaceId)> = Default::default();
let net_info = match self.docker_interfaces.get(package) {
Some(a) => a,
None => return Ok(()),
};
for (id, meta) in &net_info.interfaces {
for (service_ext_port, _lan_port_config) in meta.lan_config.iter() {
if let Some(server) = self.vhosts.service_servers.get_mut(&service_ext_port.0) {
if let Some(fqdn) = self
.docker_iface_lookups
.get(&(package.clone(), id.clone()))
{
server.remove_svc_handler_mapping(fqdn.to_owned()).await?;
self.vhosts
.cert_resolver
.remove_cert(fqdn.to_owned())
.await?;
let mapping = server.svc_mapping.read().await;
if mapping.is_empty() {
server_removals.push((service_ext_port.0, id.to_owned()));
}
}
}
}
}
for (port, interface_id) in server_removals {
if let Some(removed_server) = self.vhosts.service_servers.remove(&port) {
removed_server.shutdown.send(()).map_err(|_| {
Error::new(
eyre!("Hyper server did not quit properly"),
crate::ErrorKind::Unknown,
)
})?;
removed_server
.handle
.await
.with_kind(crate::ErrorKind::Unknown)?;
self.docker_interfaces.remove(&package.clone());
self.docker_iface_lookups
.remove(&(package.clone(), interface_id));
}
}
Ok(())
}
pub fn get_embassy_hostname(&self) -> String {
self.embassyd_fqdn.to_string()
}
}

441
backend/src/net/ssl.rs
View File

@@ -1,7 +1,8 @@
use std::cmp::Ordering;
use std::collections::{BTreeMap, BTreeSet};
use std::net::IpAddr;
use std::path::Path;
use color_eyre::eyre::eyre;
use futures::FutureExt;
use openssl::asn1::{Asn1Integer, Asn1Time};
use openssl::bn::{BigNum, MsbOption};
@@ -11,147 +12,109 @@ use openssl::nid::Nid;
use openssl::pkey::{PKey, Private};
use openssl::x509::{X509Builder, X509Extension, X509NameBuilder, X509};
use openssl::*;
use patch_db::DbHandle;
use sqlx::PgPool;
use tokio::process::Command;
use tokio::sync::Mutex;
use tokio::sync::{Mutex, RwLock};
use tracing::instrument;
use crate::s9pk::manifest::PackageId;
use crate::util::Invoke;
use crate::account::AccountInfo;
use crate::hostname::Hostname;
use crate::net::dhcp::ips;
use crate::net::keys::{Key, KeyInfo};
use crate::{Error, ErrorKind, ResultExt};
static CERTIFICATE_VERSION: i32 = 2; // X509 version 3 is actually encoded as '2' in the cert because fuck you.
pub const ROOT_CA_STATIC_PATH: &str = "/var/lib/embassy/ssl/root-ca.crt";
#[derive(Debug, Clone)]
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub struct CertPair {
pub ed25519: X509,
pub nistp256: X509,
}
impl CertPair {
fn updated(
pair: Option<&Self>,
hostname: &Hostname,
signer: (&PKey<Private>, &X509),
applicant: &Key,
ip: BTreeSet<IpAddr>,
) -> Result<(Self, bool), Error> {
let mut updated = false;
let mut updated_cert = |cert: Option<&X509>, osk: PKey<Private>| -> Result<X509, Error> {
let mut ips = BTreeSet::new();
if let Some(cert) = cert {
ips.extend(
cert.subject_alt_names()
.iter()
.flatten()
.filter_map(|a| a.ipaddress())
.filter_map(|a| match a.len() {
4 => Some::<IpAddr>(<[u8; 4]>::try_from(a).unwrap().into()),
16 => Some::<IpAddr>(<[u8; 16]>::try_from(a).unwrap().into()),
_ => None,
}),
);
if cert
.not_after()
.compare(Asn1Time::days_from_now(30)?.as_ref())?
== Ordering::Greater
&& ips.is_superset(&ip)
{
return Ok(cert.clone());
}
}
ips.extend(ip.iter().copied());
updated = true;
make_leaf_cert(signer, (&osk, &SANInfo::new(&applicant, hostname, ips)))
};
Ok((
Self {
ed25519: updated_cert(pair.map(|c| &c.ed25519), applicant.openssl_key_ed25519())?,
nistp256: updated_cert(
pair.map(|c| &c.nistp256),
applicant.openssl_key_nistp256(),
)?,
},
updated,
))
}
}
#[derive(Debug)]
pub struct SslManager {
store: SslStore,
hostname: Hostname,
root_cert: X509,
int_key: PKey<Private>,
int_cert: X509,
cert_cache: RwLock<BTreeMap<Key, CertPair>>,
}
impl SslManager {
pub fn new(account: &AccountInfo) -> Result<Self, Error> {
let int_key = generate_key()?;
let int_cert = make_int_cert((&account.root_ca_key, &account.root_ca_cert), &int_key)?;
Ok(Self {
hostname: account.hostname.clone(),
root_cert: account.root_ca_cert.clone(),
int_key,
int_cert,
cert_cache: RwLock::new(BTreeMap::new()),
})
}
pub async fn with_certs(&self, key: Key, ip: IpAddr) -> Result<KeyInfo, Error> {
let mut ips = ips().await?;
ips.insert(ip);
let (pair, updated) = CertPair::updated(
self.cert_cache.read().await.get(&key),
&self.hostname,
(&self.int_key, &self.int_cert),
&key,
ips,
)?;
if updated {
self.cert_cache
.write()
.await
.insert(key.clone(), pair.clone());
}
#[derive(Debug, Clone)]
struct SslStore {
secret_store: PgPool,
}
impl SslStore {
fn new(db: PgPool) -> Result<Self, Error> {
Ok(SslStore { secret_store: db })
}
#[instrument(skip(self))]
async fn save_root_certificate(&self, key: &PKey<Private>, cert: &X509) -> Result<(), Error> {
let key_str = String::from_utf8(key.private_key_to_pem_pkcs8()?)?;
let cert_str = String::from_utf8(cert.to_pem()?)?;
let _n = sqlx::query!("INSERT INTO certificates (id, priv_key_pem, certificate_pem, lookup_string, created_at, updated_at) VALUES (0, $1, $2, NULL, now(), now())", key_str, cert_str).execute(&self.secret_store).await?;
Ok(())
}
#[instrument(skip(self))]
async fn load_root_certificate(&self) -> Result<Option<(PKey<Private>, X509)>, Error> {
let m_row =
sqlx::query!("SELECT priv_key_pem, certificate_pem FROM certificates WHERE id = 0;")
.fetch_optional(&self.secret_store)
.await?;
match m_row {
None => Ok(None),
Some(row) => {
let priv_key = PKey::private_key_from_pem(&row.priv_key_pem.into_bytes())?;
let certificate = X509::from_pem(&row.certificate_pem.into_bytes())?;
Ok(Some((priv_key, certificate)))
}
}
}
#[instrument(skip(self))]
async fn save_intermediate_certificate(
&self,
key: &PKey<Private>,
cert: &X509,
) -> Result<(), Error> {
let key_str = String::from_utf8(key.private_key_to_pem_pkcs8()?)?;
let cert_str = String::from_utf8(cert.to_pem()?)?;
let _n = sqlx::query!("INSERT INTO certificates (id, priv_key_pem, certificate_pem, lookup_string, created_at, updated_at) VALUES (1, $1, $2, NULL, now(), now())", key_str, cert_str).execute(&self.secret_store).await?;
Ok(())
}
async fn load_intermediate_certificate(&self) -> Result<Option<(PKey<Private>, X509)>, Error> {
let m_row =
sqlx::query!("SELECT priv_key_pem, certificate_pem FROM certificates WHERE id = 1;")
.fetch_optional(&self.secret_store)
.await?;
match m_row {
None => Ok(None),
Some(row) => {
let priv_key = PKey::private_key_from_pem(&row.priv_key_pem.into_bytes())?;
let certificate = X509::from_pem(&row.certificate_pem.into_bytes())?;
Ok(Some((priv_key, certificate)))
}
}
}
#[instrument(skip(self))]
async fn import_root_certificate(
&self,
root_key: &PKey<Private>,
root_cert: &X509,
) -> Result<(), Error> {
// remove records for both root and intermediate CA
sqlx::query!("DELETE FROM certificates WHERE id = 0 OR id = 1;")
.execute(&self.secret_store)
.await?;
self.save_root_certificate(root_key, root_cert).await?;
Ok(())
}
#[instrument(skip(self))]
async fn save_certificate(
&self,
key: &PKey<Private>,
cert: &X509,
lookup_string: &str,
) -> Result<(), Error> {
let key_str = String::from_utf8(key.private_key_to_pem_pkcs8()?)?;
let cert_str = String::from_utf8(cert.to_pem()?)?;
let _n = sqlx::query!("INSERT INTO certificates (priv_key_pem, certificate_pem, lookup_string, created_at, updated_at) VALUES ($1, $2, $3, now(), now())", key_str, cert_str, lookup_string).execute(&self.secret_store).await?;
Ok(())
}
async fn load_certificate(
&self,
lookup_string: &str,
) -> Result<Option<(PKey<Private>, X509)>, Error> {
let m_row = sqlx::query!(
"SELECT priv_key_pem, certificate_pem FROM certificates WHERE lookup_string = $1",
lookup_string
)
.fetch_optional(&self.secret_store)
.await?;
match m_row {
None => Ok(None),
Some(row) => {
let priv_key = PKey::private_key_from_pem(&row.priv_key_pem.into_bytes())?;
let certificate = X509::from_pem(&row.certificate_pem.into_bytes())?;
Ok(Some((priv_key, certificate)))
}
}
}
#[instrument(skip(self))]
async fn update_certificate(
&self,
key: &PKey<Private>,
cert: &X509,
lookup_string: &str,
) -> Result<(), Error> {
let key_str = String::from_utf8(key.private_key_to_pem_pkcs8()?)?;
let cert_str = String::from_utf8(cert.to_pem()?)?;
let n = sqlx::query!("UPDATE certificates SET priv_key_pem = $1, certificate_pem = $2, updated_at = now() WHERE lookup_string = $3", key_str, cert_str, lookup_string)
.execute(&self.secret_store).await?;
if n.rows_affected() == 0 {
return Err(Error::new(
eyre!(
"Attempted to update non-existent certificate: {}",
lookup_string
),
ErrorKind::OpenSsl,
));
}
Ok(())
Ok(key.with_certs(pair, self.int_cert.clone(), self.root_cert.clone()))
}
}
@@ -161,150 +124,13 @@ lazy_static::lazy_static! {
static ref SSL_MUTEX: Mutex<()> = Mutex::new(()); // TODO: make thread safe
}
impl SslManager {
#[instrument(skip(db, handle))]
pub async fn init<Db: DbHandle>(db: PgPool, handle: &mut Db) -> Result<Self, Error> {
let store = SslStore::new(db)?;
let receipts = crate::hostname::HostNameReceipt::new(handle).await?;
let id = crate::hostname::get_id(handle, &receipts).await?;
let (root_key, root_cert) = match store.load_root_certificate().await? {
None => {
let root_key = generate_key()?;
let server_id = id;
let root_cert = make_root_cert(&root_key, &server_id)?;
store.save_root_certificate(&root_key, &root_cert).await?;
Ok::<_, Error>((root_key, root_cert))
}
Some((key, cert)) => Ok((key, cert)),
}?;
// generate static file for download, this will gte blown up on embassy restart so it's good to write it on
// every ssl manager init
tokio::fs::create_dir_all(
Path::new(ROOT_CA_STATIC_PATH)
.parent()
.unwrap_or(Path::new("/")),
)
.await?;
tokio::fs::write(ROOT_CA_STATIC_PATH, root_cert.to_pem()?).await?;
// write to ca cert store
tokio::fs::write(
"/usr/local/share/ca-certificates/embassy-root-ca.crt",
root_cert.to_pem()?,
)
.await?;
Command::new("update-ca-certificates")
.invoke(crate::ErrorKind::OpenSsl)
.await?;
let (int_key, int_cert) = match store.load_intermediate_certificate().await? {
None => {
let int_key = generate_key()?;
let int_cert = make_int_cert((&root_key, &root_cert), &int_key)?;
store
.save_intermediate_certificate(&int_key, &int_cert)
.await?;
Ok::<_, Error>((int_key, int_cert))
}
Some((key, cert)) => Ok((key, cert)),
}?;
sqlx::query!("SELECT setval('certificates_id_seq', GREATEST(MAX(id) + 1, nextval('certificates_id_seq') - 1)) FROM certificates")
.fetch_one(&store.secret_store).await?;
Ok(SslManager {
store,
root_cert,
int_key,
int_cert,
})
}
// TODO: currently the burden of proof is on the caller to ensure that all of the arguments to this function are
// consistent. The following properties are assumed and not verified:
// 1. `root_cert` is self-signed and contains the public key that matches the private key `root_key`
// 2. certificate is not past its expiration date
// Warning: If this function ever fails, you must either call it again or regenerate your certificates from scratch
// since it is possible for it to fail after successfully saving the root certificate but before successfully saving
// the intermediate certificate
#[instrument(skip(db))]
pub async fn import_root_ca(
db: PgPool,
root_key: PKey<Private>,
root_cert: X509,
) -> Result<Self, Error> {
let store = SslStore::new(db)?;
store.import_root_certificate(&root_key, &root_cert).await?;
let int_key = generate_key()?;
let int_cert = make_int_cert((&root_key, &root_cert), &int_key)?;
store
.save_intermediate_certificate(&int_key, &int_cert)
.await?;
Ok(SslManager {
store,
root_cert,
int_key,
int_cert,
})
}
#[instrument(skip(self))]
pub async fn export_root_ca(&self) -> Result<(PKey<Private>, X509), Error> {
match self.store.load_root_certificate().await? {
None => Err(Error::new(
eyre!("Failed to export root certificate: root certificate has not been generated"),
ErrorKind::OpenSsl,
)),
Some(a) => Ok(a),
}
}
#[instrument(skip(self))]
pub async fn certificate_for(
&self,
dns_base: &str,
package_id: &PackageId,
) -> Result<(PKey<Private>, Vec<X509>), Error> {
let (key, cert) = match self.store.load_certificate(dns_base).await? {
None => {
let key = generate_key()?;
let cert = make_leaf_cert(
(&self.int_key, &self.int_cert),
(&key, dns_base, package_id),
)?;
self.store.save_certificate(&key, &cert, dns_base).await?;
Ok::<_, Error>((key, cert))
}
Some((key, cert)) => {
let window_end = Asn1Time::days_from_now(30)?;
let expiration = cert.not_after();
if expiration.compare(&window_end)? == Ordering::Less {
let key = generate_key()?;
let cert = make_leaf_cert(
(&self.int_key, &self.int_cert),
(&key, dns_base, package_id),
)?;
self.store.update_certificate(&key, &cert, dns_base).await?;
Ok((key, cert))
} else {
Ok((key, cert))
}
}
}?;
Ok((
key,
vec![cert, self.int_cert.clone(), self.root_cert.clone()],
))
}
}
pub async fn export_key(key: &PKey<Private>, target: &Path) -> Result<(), Error> {
tokio::fs::write(target, key.private_key_to_pem_pkcs8()?)
.map(|res| res.with_ctx(|_| (ErrorKind::Filesystem, target.display().to_string())))
.await?;
Ok(())
}
pub async fn export_cert(chain: &Vec<X509>, target: &Path) -> Result<(), Error> {
pub async fn export_cert(chain: &[&X509], target: &Path) -> Result<(), Error> {
tokio::fs::write(
target,
chain
@@ -315,6 +141,7 @@ pub async fn export_cert(chain: &Vec<X509>, target: &Path) -> Result<(), Error>
.await?;
Ok(())
}
#[instrument]
fn rand_serial() -> Result<Asn1Integer, Error> {
let mut bn = BigNum::new()?;
@@ -323,13 +150,14 @@ fn rand_serial() -> Result<Asn1Integer, Error> {
Ok(asn1)
}
#[instrument]
fn generate_key() -> Result<PKey<Private>, Error> {
pub fn generate_key() -> Result<PKey<Private>, Error> {
let new_key = EcKey::generate(EC_GROUP.as_ref())?;
let key = PKey::from_ec_key(new_key)?;
Ok(key)
}
#[instrument]
fn make_root_cert(root_key: &PKey<Private>, server_id: &str) -> Result<X509, Error> {
pub fn make_root_cert(root_key: &PKey<Private>, hostname: &Hostname) -> Result<X509, Error> {
let mut builder = X509Builder::new()?;
builder.set_version(CERTIFICATE_VERSION)?;
@@ -342,8 +170,7 @@ fn make_root_cert(root_key: &PKey<Private>, server_id: &str) -> Result<X509, Err
builder.set_serial_number(&*rand_serial()?)?;
let mut subject_name_builder = X509NameBuilder::new()?;
subject_name_builder
.append_entry_by_text("CN", &format!("Embassy Local Root CA ({})", server_id))?;
subject_name_builder.append_entry_by_text("CN", &format!("{} Local Root CA", &*hostname.0))?;
subject_name_builder.append_entry_by_text("O", "Start9")?;
subject_name_builder.append_entry_by_text("OU", "Embassy")?;
let subject_name = subject_name_builder.build();
@@ -381,7 +208,7 @@ fn make_root_cert(root_key: &PKey<Private>, server_id: &str) -> Result<X509, Err
Ok(cert)
}
#[instrument]
fn make_int_cert(
pub fn make_int_cert(
signer: (&PKey<Private>, &X509),
applicant: &PKey<Private>,
) -> Result<X509, Error> {
@@ -442,10 +269,52 @@ fn make_int_cert(
Ok(cert)
}
#[derive(Debug)]
pub struct SANInfo {
pub dns: BTreeSet<String>,
pub ips: BTreeSet<IpAddr>,
}
impl SANInfo {
pub fn new(key: &Key, hostname: &Hostname, ips: BTreeSet<IpAddr>) -> Self {
let mut dns = BTreeSet::new();
if let Some((id, _)) = key.interface() {
dns.insert(format!("{id}.embassy"));
dns.insert(key.local_address().to_string());
} else {
dns.insert("embassy".to_owned());
dns.insert(hostname.local_domain_name());
dns.insert(hostname.no_dot_host_name());
dns.insert("localhost".to_owned());
}
dns.insert(key.tor_address().to_string());
Self { dns, ips }
}
}
impl std::fmt::Display for SANInfo {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let mut written = false;
for dns in &self.dns {
if written {
write!(f, ",")?;
}
written = true;
write!(f, "DNS:{dns},DNS:*.{dns}")?;
}
for ip in &self.ips {
if written {
write!(f, ",")?;
}
written = true;
write!(f, "IP:{ip}")?;
}
Ok(())
}
}
#[instrument]
fn make_leaf_cert(
pub fn make_leaf_cert(
signer: (&PKey<Private>, &X509),
applicant: (&PKey<Private>, &str, &PackageId),
applicant: (&PKey<Private>, &SANInfo),
) -> Result<X509, Error> {
let mut builder = X509Builder::new()?;
builder.set_version(CERTIFICATE_VERSION)?;
@@ -461,7 +330,15 @@ fn make_leaf_cert(
builder.set_serial_number(&*rand_serial()?)?;
let mut subject_name_builder = X509NameBuilder::new()?;
subject_name_builder.append_entry_by_text("CN", &format!("{}.local", &applicant.1))?;
subject_name_builder.append_entry_by_text(
"CN",
applicant
.1
.dns
.first()
.map(String::as_str)
.unwrap_or("localhost"),
)?;
subject_name_builder.append_entry_by_text("O", "Start9")?;
subject_name_builder.append_entry_by_text("OU", "Embassy")?;
let subject_name = subject_name_builder.build();
@@ -493,15 +370,9 @@ fn make_leaf_cert(
"critical,digitalSignature,keyEncipherment",
)?;
let subject_alt_name = X509Extension::new_nid(
Some(&cfg),
Some(&ctx),
Nid::SUBJECT_ALT_NAME,
&format!(
"DNS:{}.local,DNS:*.{}.local,DNS:{}.onion,DNS:*.{}.onion,DNS:{}.embassy,DNS:*.{}.embassy",
&applicant.1, &applicant.1, &applicant.1, &applicant.1, &applicant.2, &applicant.2,
),
)?;
let san_string = applicant.1.to_string();
let subject_alt_name =
X509Extension::new_nid(Some(&cfg), Some(&ctx), Nid::SUBJECT_ALT_NAME, &san_string)?;
builder.append_extension(subject_key_identifier)?;
builder.append_extension(authority_key_identifier)?;
builder.append_extension(subject_alt_name)?;

62
backend/src/net/static_server.rs
View File

@@ -1,5 +1,5 @@
use std::fs::Metadata;
use std::path::{Path, PathBuf};
use std::path::Path;
use std::sync::Arc;
use std::time::UNIX_EPOCH;
@@ -9,8 +9,11 @@ use color_eyre::eyre::eyre;
use digest::Digest;
use futures::FutureExt;
use http::header::ACCEPT_ENCODING;
use http::header::CONTENT_ENCODING;
use http::response::Builder;
use hyper::{Body, Method, Request, Response, StatusCode};
use openssl::hash::MessageDigest;
use openssl::x509::X509;
use rpc_toolkit::rpc_handler;
use tokio::fs::File;
use tokio::io::BufReader;
@@ -249,7 +252,12 @@ async fn alt_ui(req: Request<Body>, ui_mode: UiMode) -> Result<Response<Body>, E
let full_path = Path::new(selected_root_dir).join(uri_path);
file_send(
if tokio::fs::metadata(&full_path).await.is_ok() {
if tokio::fs::metadata(&full_path)
.await
.ok()
.map(|f| f.is_file())
.unwrap_or(false)
{
full_path
} else {
Path::new(selected_root_dir).join("index.html")
@@ -296,10 +304,7 @@ async fn main_embassy_ui(req: Request<Body>, ctx: RpcContext) -> Result<Response
.await
} else if let Ok(rest) = sub_path.strip_prefix("eos") {
match rest.to_str() {
Some("local.crt") => {
file_send(crate::net::ssl::ROOT_CA_STATIC_PATH, &accept_encoding)
.await
}
Some("local.crt") => cert_send(&ctx.account.read().await.root_ca_cert),
None => Ok(bad_request()),
_ => Ok(not_found()),
}
@@ -312,13 +317,7 @@ async fn main_embassy_ui(req: Request<Body>, ctx: RpcContext) -> Result<Response
}
(&Method::GET, Some(("eos", "local.crt"))) => {
match HasValidSession::from_request_parts(&request_parts, &ctx).await {
Ok(_) => {
file_send(
PathBuf::from(crate::net::ssl::ROOT_CA_STATIC_PATH),
&accept_encoding,
)
.await
}
Ok(_) => cert_send(&ctx.account.read().await.root_ca_cert),
Err(e) => un_authorized(e, "eos/local.crt"),
}
}
@@ -331,7 +330,12 @@ async fn main_embassy_ui(req: Request<Body>, ctx: RpcContext) -> Result<Response
let full_path = Path::new(selected_root_dir).join(uri_path);
file_send(
if tokio::fs::metadata(&full_path).await.is_ok() {
if tokio::fs::metadata(&full_path)
.await
.ok()
.map(|f| f.is_file())
.unwrap_or(false)
{
full_path
} else {
Path::new(selected_root_dir).join("index.html")
@@ -383,6 +387,24 @@ fn bad_request() -> Response<Body> {
.unwrap()
}
fn cert_send(cert: &X509) -> Result<Response<Body>, Error> {
let pem = cert.to_pem()?;
Response::builder()
.status(StatusCode::OK)
.header(
http::header::ETAG,
base32::encode(
base32::Alphabet::RFC4648 { padding: false },
&*cert.digest(MessageDigest::sha256())?,
)
.to_lowercase(),
)
.header(http::header::CONTENT_TYPE, "application/x-pem-file")
.header(http::header::CONTENT_LENGTH, pem.len())
.body(Body::from(pem))
.with_kind(ErrorKind::Network)
}
async fn file_send(
path: impl AsRef<Path>,
accept_encoding: &[&str],
@@ -407,12 +429,14 @@ async fn file_send(
let mut builder = Response::builder().status(StatusCode::OK);
builder = with_e_tag(path, &metadata, builder)?;
builder = with_content_type(path, builder);
builder = with_content_length(&metadata, builder);
let body = if accept_encoding.contains(&"br") {
let body = if accept_encoding.contains(&"br") && metadata.len() > u16::MAX as u64 {
builder = builder.header(CONTENT_ENCODING, "br");
Body::wrap_stream(ReaderStream::new(BrotliEncoder::new(BufReader::new(file))))
} else if accept_encoding.contains(&"gzip") {
} else if accept_encoding.contains(&"gzip") && metadata.len() > u16::MAX as u64 {
builder = builder.header(CONTENT_ENCODING, "gzip");
Body::wrap_stream(ReaderStream::new(GzipEncoder::new(BufReader::new(file))))
} else {
builder = with_content_length(&metadata, builder);
Body::wrap_stream(ReaderStream::new(file))
};
builder.body(body).with_kind(ErrorKind::Network)
@@ -446,7 +470,7 @@ fn with_e_tag(path: &Path, metadata: &Metadata, builder: Builder) -> Result<Buil
);
let res = hasher.finalize();
Ok(builder.header(
"ETag",
http::header::ETAG,
base32::encode(base32::Alphabet::RFC4648 { padding: false }, res.as_slice()).to_lowercase(),
))
}
@@ -476,7 +500,7 @@ fn with_content_type(path: &Path, builder: Builder) -> Builder {
},
None => "text/plain",
};
builder.header("Content-Type", content_type)
builder.header(http::header::CONTENT_TYPE, content_type)
}
fn with_content_length(metadata: &Metadata, builder: Builder) -> Builder {

258
backend/src/net/tor.rs
View File

@@ -1,27 +1,25 @@
use std::collections::BTreeMap;
use std::net::{Ipv4Addr, SocketAddr};
use std::net::SocketAddr;
use std::sync::{Arc, Weak};
use clap::ArgMatches;
use color_eyre::eyre::eyre;
use futures::future::BoxFuture;
use futures::FutureExt;
use rpc_toolkit::command;
use sqlx::{Executor, Postgres};
use tokio::net::TcpStream;
use tokio::sync::Mutex;
use torut::control::{AsyncEvent, AuthenticatedConn, ConnError};
use torut::onion::{OnionAddressV3, TorSecretKeyV3};
use tracing::instrument;
use super::interface::{InterfaceId, TorConfig};
use crate::context::RpcContext;
use crate::s9pk::manifest::PackageId;
use crate::util::serde::{display_serializable, IoFormat};
use crate::{Error, ErrorKind, ResultExt as _};
#[test]
fn random_key() {
println!("x'{}'", hex::encode(TorSecretKeyV3::generate().as_bytes()));
println!("x'{}'", hex::encode(rand::random::<[u8; 32]>()));
}
#[command(subcommands(list_services))]
@@ -54,64 +52,29 @@ pub async fn list_services(
ctx.net_controller.tor.list_services().await
}
#[instrument(skip(secrets))]
pub async fn os_key<Ex>(secrets: &mut Ex) -> Result<TorSecretKeyV3, Error>
where
for<'a> &'a mut Ex: Executor<'a, Database = Postgres>,
{
let key = sqlx::query!("SELECT tor_key FROM account")
.fetch_one(secrets)
.await?
.tor_key;
let mut buf = [0; 64];
buf.clone_from_slice(
key.get(0..64).ok_or_else(|| {
Error::new(eyre!("Invalid Tor Key Length"), crate::ErrorKind::Database)
})?,
);
Ok(buf.into())
}
fn event_handler(_event: AsyncEvent<'static>) -> BoxFuture<'static, Result<(), ConnError>> {
async move { Ok(()) }.boxed()
}
pub struct TorController(Mutex<TorControllerInner>);
impl TorController {
pub async fn init(
embassyd_addr: SocketAddr,
embassyd_tor_key: TorSecretKeyV3,
tor_control: SocketAddr,
) -> Result<Self, Error> {
pub async fn init(tor_control: SocketAddr) -> Result<Self, Error> {
Ok(TorController(Mutex::new(
TorControllerInner::init(embassyd_addr, embassyd_tor_key, tor_control).await?,
TorControllerInner::init(tor_control).await?,
)))
}
pub async fn add<I: IntoIterator<Item = (InterfaceId, TorConfig, TorSecretKeyV3)> + Clone>(
pub async fn add(
&self,
pkg_id: &PackageId,
ip: Ipv4Addr,
interfaces: I,
) -> Result<(), Error> {
self.0.lock().await.add(pkg_id, ip, interfaces).await
key: &TorSecretKeyV3,
external: u16,
target: SocketAddr,
) -> Result<Arc<()>, Error> {
self.0.lock().await.add(key, external, target).await
}
pub async fn remove<I: IntoIterator<Item = InterfaceId> + Clone>(
&self,
pkg_id: &PackageId,
interfaces: I,
) -> Result<(), Error> {
self.0.lock().await.remove(pkg_id, interfaces).await
}
pub async fn embassyd_tor_key(&self) -> TorSecretKeyV3 {
self.0.lock().await.embassyd_tor_key.clone()
}
pub async fn embassyd_onion(&self) -> OnionAddressV3 {
self.0.lock().await.embassyd_onion()
pub async fn gc(&self, key: &TorSecretKeyV3, external: u16) -> Result<(), Error> {
self.0.lock().await.gc(key, external).await
}
pub async fn list_services(&self) -> Result<Vec<OnionAddressV3>, Error> {
@@ -124,92 +87,95 @@ type AuthenticatedConnection = AuthenticatedConn<
fn(AsyncEvent<'static>) -> BoxFuture<'static, Result<(), ConnError>>,
>;
#[derive(Clone, Debug, PartialEq, Eq)]
struct HiddenServiceConfig {
ip: Ipv4Addr,
cfg: TorConfig,
}
pub struct TorControllerInner {
embassyd_addr: SocketAddr,
embassyd_tor_key: TorSecretKeyV3,
control_addr: SocketAddr,
connection: Option<AuthenticatedConnection>,
services: BTreeMap<(PackageId, InterfaceId), (TorSecretKeyV3, TorConfig, Ipv4Addr)>,
connection: AuthenticatedConnection,
services: BTreeMap<String, BTreeMap<u16, BTreeMap<SocketAddr, Weak<()>>>>,
}
impl TorControllerInner {
#[instrument(skip(self, interfaces))]
async fn add<'a, I: IntoIterator<Item = (InterfaceId, TorConfig, TorSecretKeyV3)>>(
#[instrument(skip(self))]
async fn add(
&mut self,
pkg_id: &PackageId,
ip: Ipv4Addr,
interfaces: I,
) -> Result<(), Error> {
for (interface_id, tor_cfg, key) in interfaces {
let id = (pkg_id.clone(), interface_id);
match self.services.get(&id) {
Some(k) if k.0 != key => {
self.remove(pkg_id, std::iter::once(id.1.clone())).await?;
}
Some(_) => continue,
None => (),
}
self.connection
.as_mut()
.ok_or_else(|| {
Error::new(eyre!("Missing Tor Control Connection"), ErrorKind::Unknown)
})?
.add_onion_v3(
&key,
false,
false,
false,
None,
&mut tor_cfg
.port_mapping
.iter()
.map(|(external, internal)| {
(external.0, SocketAddr::from((ip, internal.0)))
})
.collect::<Vec<_>>()
.iter(),
)
.await?;
self.services.insert(id, (key, tor_cfg, ip));
}
Ok(())
key: &TorSecretKeyV3,
external: u16,
target: SocketAddr,
) -> Result<Arc<()>, Error> {
let mut rm_res = Ok(());
let onion_base = key
.public()
.get_onion_address()
.get_address_without_dot_onion();
let mut service = if let Some(service) = self.services.remove(&onion_base) {
rm_res = self.connection.del_onion(&onion_base).await;
service
} else {
BTreeMap::new()
};
let mut binding = service.remove(&external).unwrap_or_default();
let rc = if let Some(rc) = Weak::upgrade(&binding.remove(&target).unwrap_or_default()) {
rc
} else {
Arc::new(())
};
binding.insert(target, Arc::downgrade(&rc));
service.insert(external, binding);
let bindings = service
.iter()
.flat_map(|(ext, int)| {
int.iter()
.find(|(_, rc)| rc.strong_count() > 0)
.map(|(addr, _)| (*ext, SocketAddr::from(*addr)))
})
.collect::<Vec<_>>();
self.services.insert(onion_base, service);
rm_res?;
self.connection
.add_onion_v3(key, false, false, false, None, &mut bindings.iter())
.await?;
Ok(rc)
}
#[instrument(skip(self, interfaces))]
async fn remove<I: IntoIterator<Item = InterfaceId>>(
&mut self,
pkg_id: &PackageId,
interfaces: I,
) -> Result<(), Error> {
for interface_id in interfaces {
if let Some((key, _cfg, _ip)) = self.services.remove(&(pkg_id.clone(), interface_id)) {
#[instrument(skip(self))]
async fn gc(&mut self, key: &TorSecretKeyV3, external: u16) -> Result<(), Error> {
let onion_base = key
.public()
.get_onion_address()
.get_address_without_dot_onion();
if let Some(mut service) = self.services.remove(&onion_base) {
if let Some(mut binding) = service.remove(&external) {
binding = binding
.into_iter()
.filter(|(_, rc)| rc.strong_count() > 0)
.collect();
if !binding.is_empty() {
service.insert(external, binding);
}
}
let rm_res = self.connection.del_onion(&onion_base).await;
if !service.is_empty() {
let bindings = service
.iter()
.flat_map(|(ext, int)| {
int.iter()
.find(|(_, rc)| rc.strong_count() > 0)
.map(|(addr, _)| (*ext, SocketAddr::from(*addr)))
})
.collect::<Vec<_>>();
self.services.insert(onion_base, service);
rm_res?;
self.connection
.as_mut()
.ok_or_else(|| {
Error::new(eyre!("Missing Tor Control Connection"), ErrorKind::Tor)
})?
.del_onion(
&key.public()
.get_onion_address()
.get_address_without_dot_onion(),
)
.add_onion_v3(&key, false, false, false, None, &mut bindings.iter())
.await?;
} else {
rm_res?;
}
}
Ok(())
}
#[instrument]
async fn init(
embassyd_addr: SocketAddr,
embassyd_tor_key: TorSecretKeyV3,
tor_control: SocketAddr,
) -> Result<Self, Error> {
async fn init(tor_control: SocketAddr) -> Result<Self, Error> {
let mut conn = torut::control::UnauthenticatedConn::new(
TcpStream::connect(tor_control).await?, // TODO
);
@@ -223,51 +189,16 @@ impl TorControllerInner {
let mut connection: AuthenticatedConnection = conn.into_authenticated().await;
connection.set_async_event_handler(Some(event_handler));
let mut controller = TorControllerInner {
embassyd_addr,
embassyd_tor_key,
Ok(Self {
control_addr: tor_control,
connection: Some(connection),
connection,
services: BTreeMap::new(),
};
controller.add_embassyd_onion().await?;
Ok(controller)
}
#[instrument(skip(self))]
async fn add_embassyd_onion(&mut self) -> Result<(), Error> {
tracing::info!(
"Registering Main Tor Service: {}",
self.embassyd_tor_key.public().get_onion_address()
);
self.connection
.as_mut()
.ok_or_else(|| Error::new(eyre!("Missing Tor Control Connection"), ErrorKind::Tor))?
.add_onion_v3(
&self.embassyd_tor_key,
false,
false,
false,
None,
&mut std::iter::once(&(self.embassyd_addr.port(), self.embassyd_addr)),
)
.await?;
tracing::info!(
"Registered Main Tor Service: {}",
self.embassyd_tor_key.public().get_onion_address()
);
Ok(())
}
fn embassyd_onion(&self) -> OnionAddressV3 {
self.embassyd_tor_key.public().get_onion_address()
})
}
#[instrument(skip(self))]
async fn list_services(&mut self) -> Result<Vec<OnionAddressV3>, Error> {
self.connection
.as_mut()
.ok_or_else(|| Error::new(eyre!("Missing Tor Control Connection"), ErrorKind::Tor))?
.get_info("onions/current")
.await?
.lines()
@@ -312,6 +243,15 @@ async fn test() {
)
.await
.unwrap();
connection
.del_onion(
&tor_key
.public()
.get_onion_address()
.get_address_without_dot_onion(),
)
.await
.unwrap();
connection
.add_onion_v3(
&tor_key,

375
backend/src/net/vhost_controller.rs
View File

@@ -1,81 +1,322 @@
use std::collections::BTreeMap;
use std::net::SocketAddr;
use std::sync::Arc;
use std::convert::Infallible;
use std::net::{IpAddr, SocketAddr};
use std::sync::{Arc, Weak};
use tokio_rustls::rustls::ServerConfig;
use color_eyre::eyre::eyre;
use helpers::NonDetachingJoinHandle;
use http::Response;
use hyper::service::{make_service_fn, service_fn};
use hyper::Body;
use models::ResultExt;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::{Mutex, RwLock};
use tokio_rustls::rustls::server::Acceptor;
use tokio_rustls::rustls::{RootCertStore, ServerConfig};
use tokio_rustls::{LazyConfigAcceptor, TlsConnector};
use crate::net::cert_resolver::EmbassyCertResolver;
use crate::net::embassy_service_http_server::EmbassyServiceHTTPServer;
use crate::net::net_utils::ResourceFqdn;
use crate::net::HttpHandler;
use crate::net::keys::Key;
use crate::net::net_utils::SingleAccept;
use crate::net::ssl::SslManager;
use crate::util::io::BackTrackingReader;
use crate::Error;
pub struct VHOSTController {
pub service_servers: BTreeMap<u16, EmbassyServiceHTTPServer>,
pub cert_resolver: EmbassyCertResolver,
embassyd_addr: SocketAddr,
}
// not allowed: <=1024, >=32768, 5355, 5432, 9050, 6010, 9051, 5353
impl VHOSTController {
pub fn init(embassyd_addr: SocketAddr) -> Self {
pub struct VHostController {
ssl: Arc<SslManager>,
servers: Mutex<BTreeMap<u16, VHostServer>>,
}
impl VHostController {
pub fn new(ssl: Arc<SslManager>) -> Self {
Self {
embassyd_addr,
service_servers: BTreeMap::new(),
cert_resolver: EmbassyCertResolver::new(),
ssl,
servers: Mutex::new(BTreeMap::new()),
}
}
pub fn build_ssl_svr_cfg(&self) -> Result<Arc<ServerConfig>, Error> {
let ssl_cfg = ServerConfig::builder()
.with_safe_default_cipher_suites()
.with_safe_default_kx_groups()
.with_safe_default_protocol_versions()
.unwrap()
.with_no_client_auth()
.with_cert_resolver(Arc::new(self.cert_resolver.clone()));
Ok(Arc::new(ssl_cfg))
}
pub async fn add_server_or_handle(
&mut self,
external_svc_port: u16,
fqdn: ResourceFqdn,
svc_handler: HttpHandler,
is_ssl: bool,
) -> Result<(), Error> {
if let Some(server) = self.service_servers.get_mut(&external_svc_port) {
server.add_svc_handler_mapping(fqdn, svc_handler).await?;
pub async fn add(
&self,
key: Key,
hostname: Option<String>,
external: u16,
target: SocketAddr,
connect_ssl: bool,
) -> Result<Arc<()>, Error> {
let mut writable = self.servers.lock().await;
let server = if let Some(server) = writable.remove(&external) {
server
} else {
self.add_server(is_ssl, external_svc_port, fqdn, svc_handler)
.await?;
}
Ok(())
}
async fn add_server(
&mut self,
is_ssl: bool,
external_svc_port: u16,
fqdn: ResourceFqdn,
svc_handler: HttpHandler,
) -> Result<(), Error> {
let ssl_cfg = if is_ssl {
Some(self.build_ssl_svr_cfg()?)
} else {
None
VHostServer::new(external, self.ssl.clone()).await?
};
let mut new_service_server =
EmbassyServiceHTTPServer::new(self.embassyd_addr.ip(), external_svc_port, ssl_cfg)
.await?;
new_service_server
.add_svc_handler_mapping(fqdn.clone(), svc_handler)
.await?;
self.service_servers
.insert(external_svc_port, new_service_server);
let rc = server
.add(
hostname,
TargetInfo {
addr: target,
connect_ssl,
key,
},
)
.await;
writable.insert(external, server);
Ok(rc?)
}
pub async fn gc(&self, hostname: Option<String>, external: u16) -> Result<(), Error> {
let mut writable = self.servers.lock().await;
if let Some(server) = writable.remove(&external) {
server.gc(hostname).await?;
if !server.is_empty().await? {
writable.insert(external, server);
}
}
Ok(())
}
}
#[derive(Clone, PartialEq, Eq, PartialOrd, Ord)]
struct TargetInfo {
addr: SocketAddr,
connect_ssl: bool,
key: Key,
}
struct VHostServer {
mapping: Weak<RwLock<BTreeMap<Option<String>, BTreeMap<TargetInfo, Weak<()>>>>>,
_thread: NonDetachingJoinHandle<()>,
}
impl VHostServer {
async fn new(port: u16, ssl: Arc<SslManager>) -> Result<Self, Error> {
// check if port allowed
let listener = TcpListener::bind(SocketAddr::new([0, 0, 0, 0].into(), port))
.await
.with_kind(crate::ErrorKind::Network)?;
let mapping = Arc::new(RwLock::new(BTreeMap::new()));
Ok(Self {
mapping: Arc::downgrade(&mapping),
_thread: tokio::spawn(async move {
loop {
match listener.accept().await {
Ok((stream, _)) => {
let mut stream = BackTrackingReader::new(stream);
stream.start_buffering();
let mapping = mapping.clone();
let ssl = ssl.clone();
tokio::spawn(async move {
if let Err(e) = async {
let mid = match LazyConfigAcceptor::new(
Acceptor::default(),
&mut stream,
)
.await
{
Ok(a) => a,
Err(e) => {
stream.rewind();
return hyper::server::Server::builder(
SingleAccept::new(stream),
)
.serve(make_service_fn(|_| async {
Ok::<_, Infallible>(service_fn(|req| async move {
Response::builder()
.status(
http::StatusCode::TEMPORARY_REDIRECT,
)
.header(
http::header::LOCATION,
req.headers()
.get(http::header::HOST)
.and_then(|host| host.to_str().ok())
.map(|host| {
format!("https://{host}")
})
.unwrap_or_default(),
)
.body(Body::default())
}))
}))
.await
.with_kind(crate::ErrorKind::Network);
}
};
let target_name =
mid.client_hello().server_name().map(|s| s.to_owned());
let target = {
let mapping = mapping.read().await;
mapping
.get(&target_name)
.into_iter()
.flatten()
.find(|(_, rc)| rc.strong_count() > 0)
.or_else(|| {
if target_name
.map(|s| s.parse::<IpAddr>().is_ok())
.unwrap_or(true)
{
mapping
.get(&None)
.into_iter()
.flatten()
.find(|(_, rc)| rc.strong_count() > 0)
} else {
None
}
})
.map(|(target, _)| target.clone())
};
if let Some(target) = target {
let mut tcp_stream =
TcpStream::connect(target.addr).await?;
let key =
ssl.with_certs(target.key, target.addr.ip()).await?;
let cfg = ServerConfig::builder()
.with_safe_defaults()
.with_no_client_auth();
let cfg =
if mid.client_hello().signature_schemes().contains(
&tokio_rustls::rustls::SignatureScheme::ED25519,
) {
cfg.with_single_cert(
key.fullchain_ed25519()
.into_iter()
.map(|c| {
Ok(tokio_rustls::rustls::Certificate(
c.to_der()?,
))
})
.collect::<Result<_, Error>>()?,
tokio_rustls::rustls::PrivateKey(
key.key()
.openssl_key_ed25519()
.private_key_to_der()?,
),
)
} else {
cfg.with_single_cert(
key.fullchain_nistp256()
.into_iter()
.map(|c| {
Ok(tokio_rustls::rustls::Certificate(
c.to_der()?,
))
})
.collect::<Result<_, Error>>()?,
tokio_rustls::rustls::PrivateKey(
key.key()
.openssl_key_nistp256()
.private_key_to_der()?,
),
)
};
let mut tls_stream = mid
.into_stream(Arc::new(
cfg.with_kind(crate::ErrorKind::OpenSsl)?,
))
.await?;
tls_stream.get_mut().0.stop_buffering();
if target.connect_ssl {
tokio::io::copy_bidirectional(
&mut tls_stream,
&mut TlsConnector::from(Arc::new(
tokio_rustls::rustls::ClientConfig::builder()
.with_safe_defaults()
.with_root_certificates({
let mut store = RootCertStore::empty();
store.add(
&tokio_rustls::rustls::Certificate(
key.root_ca().to_der()?,
),
).with_kind(crate::ErrorKind::OpenSsl)?;
store
})
.with_no_client_auth(),
))
.connect(
key.key()
.internal_address()
.as_str()
.try_into()
.with_kind(crate::ErrorKind::OpenSsl)?,
tcp_stream,
)
.await
.with_kind(crate::ErrorKind::OpenSsl)?,
)
.await?;
} else {
tokio::io::copy_bidirectional(
&mut tls_stream,
&mut tcp_stream,
)
.await?;
}
} else {
// 503
}
Ok::<_, Error>(())
}
.await
{
tracing::error!("Error in VHostController on port {port}: {e}");
tracing::debug!("{e:?}")
}
});
}
Err(e) => {
tracing::error!("Error in VHostController on port {port}: {e}");
tracing::debug!("{e:?}");
}
}
}
})
.into(),
})
}
async fn add(&self, hostname: Option<String>, target: TargetInfo) -> Result<Arc<()>, Error> {
if let Some(mapping) = Weak::upgrade(&self.mapping) {
let mut writable = mapping.write().await;
let mut targets = writable.remove(&hostname).unwrap_or_default();
let rc = if let Some(rc) = Weak::upgrade(&targets.remove(&target).unwrap_or_default()) {
rc
} else {
Arc::new(())
};
targets.insert(target, Arc::downgrade(&rc));
writable.insert(hostname, targets);
Ok(rc)
} else {
Err(Error::new(
eyre!("VHost Service Thread has exited"),
crate::ErrorKind::Network,
))
}
}
async fn gc(&self, hostname: Option<String>) -> Result<(), Error> {
if let Some(mapping) = Weak::upgrade(&self.mapping) {
let mut writable = mapping.write().await;
let mut targets = writable.remove(&hostname).unwrap_or_default();
targets = targets
.into_iter()
.filter(|(_, rc)| rc.strong_count() > 0)
.collect();
if !targets.is_empty() {
writable.insert(hostname, targets);
}
Ok(())
} else {
Err(Error::new(
eyre!("VHost Service Thread has exited"),
crate::ErrorKind::Network,
))
}
}
async fn is_empty(&self) -> Result<bool, Error> {
if let Some(mapping) = Weak::upgrade(&self.mapping) {
Ok(mapping.read().await.is_empty())
} else {
Err(Error::new(
eyre!("VHost Service Thread has exited"),
crate::ErrorKind::Network,
))
}
}
}

61
backend/src/net/web_server.rs Normal file
View File

@@ -0,0 +1,61 @@
use std::convert::Infallible;
use std::net::SocketAddr;
use futures::future::ready;
use futures::FutureExt;
use helpers::NonDetachingJoinHandle;
use hyper::service::{make_service_fn, service_fn};
use hyper::Server;
use tokio::sync::oneshot;
use crate::context::{DiagnosticContext, InstallContext, RpcContext, SetupContext};
use crate::net::static_server::{
diag_ui_file_router, install_ui_file_router, main_ui_server_router, setup_ui_file_router,
};
use crate::net::HttpHandler;
use crate::Error;
pub struct WebServer {
shutdown: oneshot::Sender<()>,
thread: NonDetachingJoinHandle<()>,
}
impl WebServer {
pub fn new(bind: SocketAddr, router: HttpHandler) -> Self {
let (shutdown, shutdown_recv) = oneshot::channel();
let thread = NonDetachingJoinHandle::from(tokio::spawn(async move {
let server = Server::bind(&bind)
.http1_preserve_header_case(true)
.http1_title_case_headers(true)
.serve(make_service_fn(move |_| {
let router = router.clone();
ready(Ok::<_, Infallible>(service_fn(move |req| router(req))))
}))
.with_graceful_shutdown(shutdown_recv.map(|_| ()));
if let Err(e) = server.await {
tracing::error!("Spawning hyper server error: {}", e);
}
}));
Self { shutdown, thread }
}
pub async fn shutdown(self) {
self.shutdown.send(()).unwrap_or_default();
self.thread.await.unwrap()
}
pub async fn main(bind: SocketAddr, ctx: RpcContext) -> Result<Self, Error> {
Ok(Self::new(bind, main_ui_server_router(ctx).await?))
}
pub async fn setup(bind: SocketAddr, ctx: SetupContext) -> Result<Self, Error> {
Ok(Self::new(bind, setup_ui_file_router(ctx).await?))
}
pub async fn diagnostic(bind: SocketAddr, ctx: DiagnosticContext) -> Result<Self, Error> {
Ok(Self::new(bind, diag_ui_file_router(ctx).await?))
}
pub async fn install(bind: SocketAddr, ctx: InstallContext) -> Result<Self, Error> {
Ok(Self::new(bind, install_ui_file_router(ctx).await?))
}
}