#[macro_use] extern crate log; use std::{ sync::Arc, time::Duration, }; #[cfg(feature = "gemini_srv")] use std::{ convert::TryFrom, path::PathBuf, }; #[cfg(feature = "scgi_srv")] use std::{ collections::HashMap, net::SocketAddr, str::FromStr, }; use tokio::{ io, io::BufReader, net::TcpListener, net::ToSocketAddrs, prelude::*, }; #[cfg(feature = "scgi_srv")] use tokio::net::UnixListener; #[cfg(feature = "gemini_srv")] use tokio::{ time::timeout, net::TcpStream, }; #[cfg(feature = "ratelimiting")] use tokio::time::interval; #[cfg(feature = "gemini_srv")] use rustls::ClientCertVerifier; #[cfg(feature = "gemini_srv")] use rustls::internal::msgs::handshake::DigitallySignedStruct; #[cfg(feature = "gemini_srv")] use tokio_rustls::{rustls, TlsAcceptor}; #[cfg(feature = "gemini_srv")] use rustls::*; use anyhow::*; use crate::util::opt_timeout; use routing::RoutingNode; #[cfg(feature = "ratelimiting")] use ratelimiting::RateLimiter; pub mod types; pub mod util; pub mod routing; pub mod handling; #[cfg(feature = "ratelimiting")] pub mod ratelimiting; #[cfg(feature = "user_management")] pub mod user_management; #[cfg(feature = "certgen")] pub mod gencert; #[cfg(feature="user_management")] use user_management::UserManager; #[cfg(feature = "certgen")] use gencert::CertGenMode; pub use uriparse as uri; pub use types::*; pub const REQUEST_URI_MAX_LEN: usize = 1024; pub const GEMINI_PORT: u16 = 1965; use handling::Handler; #[derive(Clone)] struct ServerInner { #[cfg(feature = "gemini_srv")] tls_acceptor: TlsAcceptor, routes: Arc>, timeout: Duration, complex_timeout: Option, #[cfg(feature="ratelimiting")] rate_limits: Arc>>, #[cfg(feature="user_management")] manager: UserManager, } impl ServerInner { async fn serve_ip(self, listener: TcpListener) -> Result<()> { #[cfg(feature = "ratelimiting")] tokio::spawn(prune_ratelimit_log(self.rate_limits.clone())); loop { let (stream, _addr) = listener.accept().await .context("Failed to accept client")?; let this = self.clone(); tokio::spawn(async move { if let Err(err) = this.serve_client(stream).await { error!("{:?}", err); } }); } } #[cfg(feature = "scgi_srv")] // Yeah it's code duplication, but I can't find a way around it, so this is what we're // getting for now async fn serve_unix(self, listener: UnixListener) -> Result<()> { #[cfg(feature = "ratelimiting")] tokio::spawn(prune_ratelimit_log(self.rate_limits.clone())); loop { let (stream, _addr) = listener.accept().await .context("Failed to accept client")?; let this = self.clone(); tokio::spawn(async move { if let Err(err) = this.serve_client(stream).await { error!("{:?}", err); } }); } } async fn serve_client( &self, #[cfg(feature = "gemini_srv")] stream: TcpStream, #[cfg(feature = "scgi_srv")] stream: impl AsyncWrite + AsyncRead + Unpin, ) -> Result<()> { let fut_accept_request = async { #[cfg(feature = "gemini_srv")] let stream = self.tls_acceptor.accept(stream).await .context("Failed to establish TLS session")?; let mut stream = BufReader::new(stream); let request = self.receive_request(&mut stream).await .context("Failed to receive request")?; Result::<_, anyhow::Error>::Ok((request, stream)) }; // Wait for the request to be parsed let (mut request, mut stream) = { #[cfg(feature = "gemini_srv")] { // Use a timeout for interacting with the client let fut_accept_request = timeout(self.timeout, fut_accept_request); fut_accept_request.await .context("Client timed out while waiting for response")?? } #[cfg(feature = "scgi_srv")] fut_accept_request.await? }; // Determine the remote client's IP address for logging and ratelimiting let peer_addr = { #[cfg(feature = "gemini_srv")] { stream.get_ref() .get_ref() .0 .peer_addr()? .ip() } #[cfg(feature = "scgi_srv")] { SocketAddr::from_str( request.headers() .get("REMOTE_ADDR") .ok_or(ParseError::Malformed("REMOTE_ADDR header not received"))? .as_str() ).context("Received malformed IP address from upstream")? .ip() } }; #[cfg(feature = "ratelimiting")] // Perform ratelimiting checks if let Some(resp) = self.check_rate_limits(peer_addr, &request) { // Log warning warn!( "Client from {} requesting {} was turned away by ratelimiting", peer_addr, request.uri() ); // Send error response self.send_response(resp, &mut stream).await .context("Failed to send response")?; // Exit return Ok(()) } info!("{} requested: {}", peer_addr, request.uri()); // Identify the client certificate from the tls stream. This is the first // certificate in the certificate chain. #[cfg(feature = "gemini_srv")] { // This is done earlier for `scgi_srv` let client_cert = stream.get_ref() .get_ref() .1 .get_peer_certificates() .and_then(|mut v| if v.is_empty() {None} else {Some(v.remove(0))}); request.set_cert(client_cert); } let response = if let Some((trailing, handler)) = self.routes.match_request(&request) { request.set_trailing(trailing); handler.handle(request).await } else { Response::not_found() }; self.send_response(response, &mut stream).await .context("Failed to send response")?; Ok(()) } async fn send_response(&self, mut response: Response, stream: &mut (impl AsyncWrite + Unpin)) -> Result<()> { let maybe_body = response.take_body(); let header = response.header(); let use_complex_timeout = header.status.is_success() && maybe_body.is_some() && header.meta.as_str() != "text/plain" && header.meta.as_str() != "text/gemini" && self.complex_timeout.is_some(); let send_general_timeout; let send_header_timeout; let send_body_timeout; if use_complex_timeout { send_general_timeout = None; send_header_timeout = Some(self.timeout); send_body_timeout = self.complex_timeout; } else { send_general_timeout = Some(self.timeout); send_header_timeout = None; send_body_timeout = None; } opt_timeout(send_general_timeout, async { // Send the header opt_timeout(send_header_timeout, send_response_header(response.header(), stream)) .await .context("Timed out while sending response header")? .context("Failed to write response header")?; // Send the body opt_timeout(send_body_timeout, maybe_send_response_body(maybe_body, stream)) .await .context("Timed out while sending response body")? .context("Failed to write response body")?; Ok::<_,Error>(()) }) .await .context("Timed out while sending response data")??; Ok(()) } #[cfg(feature="ratelimiting")] fn check_rate_limits(&self, addr: IpAddr, req: &Request) -> Option { if let Some((_, limiter)) = self.rate_limits.match_request(req) { if let Err(when) = limiter.check_key(addr) { return Some(Response::new(ResponseHeader { status: Status::SLOW_DOWN, meta: Meta::new(when.as_secs().to_string()).unwrap() })) } } None } #[cfg(feature = "gemini_srv")] async fn receive_request( &self, stream: &mut (impl AsyncBufRead + Unpin), ) -> Result { const HEADER_LIMIT: usize = REQUEST_URI_MAX_LEN + "\r\n".len(); let mut stream = stream.take(HEADER_LIMIT as u64); let mut uri = Vec::new(); stream.read_until(b'\n', &mut uri).await?; if !uri.ends_with(b"\r\n") { if uri.len() < REQUEST_URI_MAX_LEN { bail!("Request header not terminated with CRLF") } else { bail!("Request URI too long") } } // Strip CRLF uri.pop(); uri.pop(); let uri = URIReference::try_from(&*uri) .context("Request URI is invalid")? .into_owned(); Request::new( uri, #[cfg(feature="user_management")] self.manager.clone(), ).context("Failed to create request from URI") } #[cfg(feature = "scgi_srv")] async fn receive_request( &self, stream: &mut (impl AsyncBufRead + Unpin), ) -> Result { let mut buff = Vec::with_capacity(4); #[allow(clippy::char_lit_as_u8)] // Read the length of the header netstring (e.g. "120:") stream.read_until(':' as u8, &mut buff).await?; buff.pop(); // Remove the trailing ':' let len = std::str::from_utf8(&*buff) .ok() .and_then(|s| usize::from_str(s).ok()) .ok_or(ParseError::Malformed("netstring length"))?; // Read in the headers buff.clear(); buff.resize(len + 1, 0); stream.read_exact(buff.as_mut()).await?; buff.truncate(len - 1); // Remove the final \x00, // Parse the headers let (maybe_trailing, headers) = buff.split(|b| *b == 0) // Headers are null delimiited .map(|bytes| // Convert to an &str std::str::from_utf8(bytes) .map_err(|_| ParseError::Malformed("scgi headers")) .map(str::trim) ) .try_fold( // Turn the array of [header, value, header, ...] into a map (Option::<&str>::None, HashMap::::with_capacity(16)), |(last_header, mut headers), s| { s.map(|text| { match last_header { None => (Some(text), headers), Some(header) => { headers.insert(header.to_string(), text.to_string()); (None, headers) } } }) } )?; // If there's not the same number of headers as values, that's a problem if maybe_trailing.is_some() { bail!(ParseError::Malformed("trailing header")); } // Check the content length info let cont_len_val = headers.get("CONTENT_LENGTH") .ok_or(ParseError::Malformed("No content length header!"))?; let cont_len = usize::from_str(cont_len_val) .map_err(|_| ParseError::Malformed("Malformed content length"))?; if cont_len > 0 { bail!(ParseError::Malformed("Gemini SCGI requests should not have a body")); } // Spec requires setting an SCGI header to one if *headers.get("SCGI").ok_or(ParseError::Malformed("No SCGI header"))? != "1" { bail!(ParseError::Malformed("SCGI header not set to \"1\"")); } trace!("Headers received: {:?}", headers); Ok(Request::new(headers)?) } } #[derive(Debug)] #[cfg(feature = "scgi_srv")] enum ParseError { IO(io::Error), Malformed(&'static str), } #[cfg(feature = "scgi_srv")] impl From for ParseError { fn from(e: io::Error) -> Self { Self::IO(e) } } #[cfg(feature = "scgi_srv")] impl std::fmt::Display for ParseError { fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { match self { Self::IO(e) => write!(f, "IO Error while parsing and responding SCGI: {}", e), Self::Malformed(e) => write!(f, "SCGI request malformed at {}", e), } } } #[cfg(feature = "scgi_srv")] impl std::error::Error for ParseError { fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { if let Self::IO(e) = self { Some(e) } else { None } } } pub struct Server { timeout: Duration, complex_body_timeout_override: Option, routes: RoutingNode, #[cfg(feature = "gemini_srv")] cert_path: PathBuf, #[cfg(feature = "gemini_srv")] key_path: PathBuf, #[cfg(feature="ratelimiting")] rate_limits: RoutingNode>, #[cfg(feature="user_management")] data_dir: PathBuf, #[cfg(feature="user_management")] database: Option, #[cfg(feature="certgen")] certgen_mode: CertGenMode, } impl Server { pub fn new() -> Self { Self { timeout: Duration::from_secs(1), complex_body_timeout_override: Some(Duration::from_secs(30)), routes: RoutingNode::default(), #[cfg(feature = "gemini_srv")] cert_path: PathBuf::from("cert/cert.pem"), #[cfg(feature = "gemini_srv")] key_path: PathBuf::from("cert/key.pem"), #[cfg(feature="ratelimiting")] rate_limits: RoutingNode::default(), #[cfg(feature="user_management")] data_dir: "data".into(), #[cfg(feature="user_management")] database: None, #[cfg(feature="certgen")] certgen_mode: CertGenMode::Interactive, } } #[cfg(feature="user_management")] /// Sets the directory to store user data in /// /// This will only be used if a database is not provided with [`set_database()`]. /// /// Defaults to `./data` if not specified /// /// [`set_database()`]: Self::set_database() pub fn set_database_dir(mut self, path: impl Into) -> Self { self.data_dir = path.into(); self } #[cfg(feature="user_management")] /// Sets a specific database to use /// /// This opens to trees within the database, both namespaced to avoid collisions. /// /// If this is not provided, a database will be opened at the directory provided by /// [`set_database_dir()`] /// /// [`set_database_dir()`]: Self::set_database_dir() pub fn set_database(mut self, db: sled::Db) -> Self { self.database = Some(db); self } #[cfg(feature="certgen")] /// Determine where certificate config comes from, if generation is required /// /// If a certificate & keyfile are not available on the provided path, they can be /// generated. If this happens, several modes can be used to generate them, including /// not generating them and just erroring, interactively prompting the user for /// information, and using pre-provided information. pub fn set_certificate_generation_mode(mut self, mode: CertGenMode) -> Self { self.certgen_mode = mode; self } #[cfg(feature = "gemini_srv")] /// Sets the directory that kochab should look for TLS certs and keys into /// /// Northstar will look for files called `cert.pem` and `key.pem` in the provided /// directory. /// /// This does not need to be set if both [`set_cert()`](Self::set_cert()) and /// [`set_key()`](Self::set_key()) have been called. /// /// If not set, the default is `cert/` pub fn set_tls_dir(self, dir: impl Into) -> Self { let dir = dir.into(); self.set_cert(dir.join("cert.pem")) .set_key(dir.join("key.pem")) } #[cfg(feature = "gemini_srv")] /// Set the path to the TLS certificate kochab will use /// /// This defaults to `cert/cert.pem`. /// /// This does not need to be called it [`set_tls_dir()`](Self::set_tls_dir()) has been /// called. pub fn set_cert(mut self, cert_path: impl Into) -> Self { self.cert_path = cert_path.into(); self } #[cfg(feature = "gemini_srv")] /// Set the path to the ertificate key kochab will use /// /// This defaults to `cert/key.pem`. /// /// This does not need to be called it [`set_tls_dir()`](Self::set_tls_dir()) has been /// called. /// /// This should of course correspond to the key set in /// [`set_cert()`](Self::set_cert()) pub fn set_key(mut self, key_path: impl Into) -> Self { self.key_path = key_path.into(); self } /// Set the timeout on incoming requests /// /// Note that this timeout is applied twice, once for the delivery of the request, and /// once for sending the client's response. This means that for a 1 second timeout, /// the client will have 1 second to complete the TLS handshake and deliver a request /// header, then your API will have as much time as it needs to handle the request, /// before the client has another second to receive the response. /// /// If you would like a timeout for your code itself, please use /// [`tokio::time::Timeout`] to implement it internally. /// /// **The default timeout is 1 second.** As somewhat of a workaround for /// shortcomings of the specification, this timeout, and any timeout set using this /// method, is overridden in special cases, specifically for MIME types outside of /// `text/plain` and `text/gemini`, to be 30 seconds. If you would like to change or /// prevent this, please see /// [`override_complex_body_timeout`](Self::override_complex_body_timeout()). pub fn set_timeout(mut self, timeout: Duration) -> Self { self.timeout = timeout; self } /// Override the timeout for complex body types /// /// Many clients choose to handle body types which cannot be displayed by prompting /// the user if they would like to download or open the request body. However, since /// this prompt occurs in the middle of receiving a request, often the connection /// times out before the end user is able to respond to the prompt. /// /// As a workaround, it is possible to set an override on the request timeout in /// specific conditions: /// /// 1. **Only override the timeout for receiving the body of the request.** This will /// not override the timeout on sending the request header, nor on receiving the /// response header. /// 2. **Only override the timeout for successful responses.** The only bodies which /// have bodies are successful ones. In all other cases, there's no body to /// timeout for /// 3. **Only override the timeout for complex body types.** Almost all clients are /// able to display `text/plain` and `text/gemini` responses, and will not prompt /// the user for these response types. This means that there is no reason to /// expect a client to have a human-length response time for these MIME types. /// Because of this, responses of this type will not be overridden. /// /// This method is used to override the timeout for responses meeting these specific /// criteria. All other stages of the connection will use the timeout specified in /// [`set_timeout()`](Self::set_timeout()). /// /// If this is set to [`None`], then the client will have the default amount of time /// to both receive the header and the body. If this is set to [`Some`], the client /// will have the default amount of time to recieve the header, and an *additional* /// alotment of time to recieve the body. /// /// The default timeout for this is 30 seconds. pub fn override_complex_body_timeout(mut self, timeout: Option) -> Self { self.complex_body_timeout_override = timeout; self } /// Add a handler for a route /// /// A route must be an absolute path, for example "/endpoint" or "/", but not /// "endpoint". Entering a relative or malformed path will result in a panic. /// /// For more information about routing mechanics, see the docs for [`RoutingNode`]. pub fn add_route(mut self, path: &'static str, handler: impl Into) -> Self { self.routes.add_route(path, handler.into()); self } #[cfg(feature="ratelimiting")] /// Add a rate limit to a route /// /// The server will allow at most `burst` connections to any endpoints under this /// route in a period of `period`. All extra requests will recieve a `SLOW_DOWN`, and /// not be sent to the handler. /// /// A route must be an absolute path, for example "/endpoint" or "/", but not /// "endpoint". Entering a relative or malformed path will result in a panic. /// /// For more information about routing mechanics, see the docs for [`RoutingNode`]. pub fn ratelimit(mut self, path: &'static str, burst: usize, period: Duration) -> Self { let limiter = RateLimiter::new(period, burst); self.rate_limits.add_route(path, limiter); self } fn build(mut self) -> Result { #[cfg(feature = "gemini_srv")] let config = tls_config( &self.cert_path, &self.key_path, #[cfg(feature="certgen")] self.certgen_mode ).context("Failed to create TLS config")?; self.routes.shrink(); #[cfg(feature="user_management")] let data_dir = self.data_dir; Ok(ServerInner { routes: Arc::new(self.routes), timeout: self.timeout, complex_timeout: self.complex_body_timeout_override, #[cfg(feature = "gemini_srv")] tls_acceptor: TlsAcceptor::from(config), #[cfg(feature="ratelimiting")] rate_limits: Arc::new(self.rate_limits), #[cfg(feature="user_management")] manager: UserManager::new( self.database.unwrap_or_else(move|| sled::open(data_dir).unwrap()) )?, }) } /// Start serving requests on a given bound address & port /// /// `addr` can be anything `tokio` can parse, including just a string like /// "localhost:1965" pub async fn serve_ip(self, addr: impl ToSocketAddrs) -> Result<()> { let server = self.build()?; let socket = TcpListener::bind(addr).await?; server.serve_ip(socket).await } #[cfg(feature = "scgi_srv")] /// Start serving requests on a given unix socket /// /// Requires an address in the form of a path to bind to. This is only available when /// in `scgi_srv` mode. pub async fn serve_unix(self, addr: impl AsRef) -> Result<()> { let server = self.build()?; let socket = UnixListener::bind(addr)?; server.serve_unix(socket).await } } impl Default for Server { fn default() -> Self { Self::new() } } async fn send_response_header(header: &ResponseHeader, stream: &mut (impl AsyncWrite + Unpin)) -> Result<()> { let header = format!( "{status} {meta}\r\n", status = header.status.code(), meta = header.meta.as_str(), ); stream.write_all(header.as_bytes()).await?; stream.flush().await?; Ok(()) } async fn maybe_send_response_body(maybe_body: Option, stream: &mut (impl AsyncWrite + Unpin)) -> Result<()> { if let Some(body) = maybe_body { send_response_body(body, stream).await?; } Ok(()) } async fn send_response_body(body: Body, stream: &mut (impl AsyncWrite + Unpin)) -> Result<()> { match body { Body::Bytes(bytes) => stream.write_all(&bytes).await?, Body::Reader(mut reader) => { io::copy(&mut reader, stream).await?; }, } stream.flush().await?; Ok(()) } #[cfg(feature="ratelimiting")] /// Every 5 minutes, remove excess keys from all ratelimiters async fn prune_ratelimit_log(rate_limits: Arc>>) -> Never { let mut interval = interval(tokio::time::Duration::from_secs(10)); let log = rate_limits.as_ref(); loop { interval.tick().await; log.iter().for_each(RateLimiter::trim_keys_verbose); } } #[cfg(feature = "gemini_srv")] fn tls_config( cert_path: &PathBuf, key_path: &PathBuf, #[cfg(feature = "certgen")] mode: CertGenMode, ) -> Result> { let mut config = ServerConfig::new(AllowAnonOrSelfsignedClient::new()); #[cfg(feature = "certgen")] mode.load_or_generate(&mut config, cert_path, key_path)?; #[cfg(not(feature = "certgen"))] { let cert_chain = load_cert_chain(cert_path) .context("Failed to load TLS certificate")?; let key = load_key(key_path) .context("Failed to load TLS key")?; config.set_single_cert(cert_chain, key) .context("Failed to use loaded TLS certificate")?; } Ok(config.into()) } #[cfg(feature = "gemini_srv")] fn load_cert_chain(cert_path: &PathBuf) -> Result> { let certs = std::fs::File::open(cert_path) .with_context(|| format!("Failed to open `{:?}`", cert_path))?; let mut certs = std::io::BufReader::new(certs); let certs = rustls::internal::pemfile::certs(&mut certs) .map_err(|_| anyhow!("failed to load certs `{:?}`", cert_path))?; Ok(certs) } #[cfg(feature = "gemini_srv")] fn load_key(key_path: &PathBuf) -> Result { let keys = std::fs::File::open(key_path) .with_context(|| format!("Failed to open `{:?}`", key_path))?; let mut keys = std::io::BufReader::new(keys); let mut keys = rustls::internal::pemfile::pkcs8_private_keys(&mut keys) .map_err(|_| anyhow!("failed to load key `{:?}`", key_path))?; ensure!(!keys.is_empty(), "no key found"); let key = keys.swap_remove(0); Ok(key) } #[cfg(feature = "gemini_srv")] /// A client cert verifier that accepts all connections /// /// Unfortunately, rustls doesn't provide a ClientCertVerifier that accepts self-signed /// certificates, so we need to implement this ourselves. struct AllowAnonOrSelfsignedClient { } #[cfg(feature = "gemini_srv")] impl AllowAnonOrSelfsignedClient { /// Create a new verifier fn new() -> Arc { Arc::new(Self {}) } } #[cfg(feature = "gemini_srv")] impl ClientCertVerifier for AllowAnonOrSelfsignedClient { fn client_auth_root_subjects( &self, _: Option<&webpki::DNSName> ) -> Option { Some(Vec::new()) } fn client_auth_mandatory(&self, _sni: Option<&webpki::DNSName>) -> Option { Some(false) } // the below methods are a hack until webpki doesn't break with certain certs fn verify_client_cert( &self, _: &[Certificate], _: Option<&webpki::DNSName> ) -> Result { Ok(ClientCertVerified::assertion()) } fn verify_tls12_signature( &self, _message: &[u8], _cert: &Certificate, _dss: &DigitallySignedStruct, ) -> Result { Ok(HandshakeSignatureValid::assertion()) } fn verify_tls13_signature( &self, _message: &[u8], _cert: &Certificate, _dss: &DigitallySignedStruct, ) -> Result { Ok(HandshakeSignatureValid::assertion()) } } #[cfg(feature = "ratelimiting")] enum Never {}